Django Celery 异步任务配置与运行指南

环境准备与安装

在 Django 项目中引入 Celery 异步处理机制，首先需要安装必要的依赖包。通过 pip 安装 celery、flower 以及 eventlet：

pip install celery flower eventlet

Django 项目初始化配置

为了让 Django 启动时自动导入 Celery 实例，需要在项目的根目录下的 __init__.py 文件中添加以下代码：

from __future__ import absolute_import

# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.
from .celerypro import app as celery_app

创建 Celery 实例

在项目目录下新建 celerypro.py 文件，用于初始化 Celery 应用实例。该文件负责设置 Django 的配置模块，并加载 Celery 的配置文件：

from __future__ import absolute_import
import os
from celery import Celery
from django.conf import settings

# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'voice_quality_assurance_configure.settings') #修改项目配置文件的地址
app = Celery('voice_quality_assurance_configure') #修改项目目录名称
# Using a string here means the worker will not have to
# pickle the object when using Windows.
app.config_from_object('voice_quality_assurance_configure.celeryconfig') #修改celery配置文件的地址
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

配置 Broker 与 Backend

新建 celeryconfig.py 文件，用于定义消息队列（Broker）和结果存储（Backend）。以下配置示例使用了 RabbitMQ 作为 Broker：

from kombu import Queue
BROKER_URL = 'amqp://用户名:密码@ip:5672'# 指定 Broker
CELERY_RESULT_BACKEND = 'rpc://用户名:密码@ip:5672'# 指定 Backend
CELERY_TIMEZONE='Asia/Shanghai'# 指定时区，默认是 UTC
CELERY_TASK_SERIALIZER = 'pickle'
CELERY_RESULT_SERIALIZER = 'pickle'
CELERY_ACCEPT_CONTENT = ['pickle', 'json']
CELERY_IGNORE_RESULT = True
# CELERY_TIMEZONE='UTC'
CELERY_IMPORTS = (
  # 指定导入的任务模块
  'apps.mission.tasks'
)
CELERY_QUEUES = (
  Queue('default', routing_key='default'), #声明队列和对应路由键
  Queue('worker_queue', routing_key='worker'), #声明队列和对应路由键
)
CELERY_ROUTES = {
  'apps.mission.tasks.createsingletask': {'queue': 'worker_queue', 'routing_key': 'worker'},
}

定义异步任务

在 Django 的 App 目录下新建 tasks.py 文件。注意文件名必须为 tasks，以便 Celery 自动发现任务：

from celery import shared_task
@shared_task()
def createsingletask():
  print(test)

调用异步任务

在 views.py 中导入定义好的任务，并通过 apply_async 方法调用：

from your_app.tasks import createsingletask
createsingletask.apply_async(())

启动 Celery Worker

使用 eventlet 并发模型启动多个 Worker 节点，以处理异步任务：

celery worker -A voice_quality_assurance_configure --loglevel=info -P eventlet -n worker1
celery worker -A voice_quality_assurance_configure --loglevel=info -P eventlet -n worker2
celery worker -A voice_quality_assurance_configure --loglevel=info -P eventlet -n worker3
celery worker -A voice_quality_assurance_configure --loglevel=info -P eventlet -n worker4
celery worker -A voice_quality_assurance_configure --loglevel=info -P eventlet -n worker5

以上就是 Django Celery 异步任务配置与运行的详细内容，更多关于 Django 异步处理、Celery 配置以及 flower 监控的资料请关注本站其它相关文章！