Django中使用Celery的方法示例

起步

在 《分布式任务队列Celery使用说明》 中介绍了在 Python 中使用 Celery 来实验异步任务和定时任务功能。本文介绍如何在 Django 中使用 Celery。

安装

pip install django-celery

这个命令使用的依赖是 Celery 3.x 的版本，所以会把我之前安装的 4.x 卸载，不过对功能上并没有什么影响。我们也完全可以仅用Celery在django中使用，但使用 django-celery 模块能更好的管理 celery。

使用

可以把有关 Celery 的配置放到 settings.py 里去，但我比较习惯单独一个文件来放，然后在 settings.py 引入进来：

# celery_config.pyimport djceleryimport osos.environ.setdefault('FORKED_BY_MULTIPROCESSING', '1')djcelery.setup_loader()BROKER_URL = 'redis://127.0.0.1:6379/1'CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/2'# UTCCELERY_ENABLE_UTC = TrueCELERY_TIMEZONE = 'Asia/Shanghai'CELERY_IMPORTS = ( 'app.tasks',)# 有些情况可以防止死锁CELERY_FORCE_EXECV = True# 设置并发的worker数量CELERYD_CONCURRENCY = 4# 任务发送完成是否需要确认，这一项对性能有一点影响CELERY_ACKS_LATE = True# 每个worker执行了多少任务就会销毁，防止内存泄露，默认是无限的CELERYD_MAX_TASKS_PER_CHILD = 40# 规定完成任务的时间CELERYD_TASK_TIME_LIMIT = 15 * 60 # 在15分钟内完成任务，否则执行该任务的worker将被杀死，任务移交给父进程# 设置默认的队列名称，如果一个消息不符合其他的队列就会放在默认队列里面，如果什么都不设置的话，数据都会发送到默认的队列中CELERY_DEFAULT_QUEUE = "default"# 设置详细的队列CELERY_QUEUES = { "default": { # 这是上面指定的默认队列  "exchange": "default",  "exchange_type": "direct",  "routing_key": "default" }, "beat_queue": {  "exchange": "beat_queue",  "exchange_type": "direct",  "routing_key": "beat_queue" }}

配置文件中设置了 CELERY_IMPORTS 导入的任务，所以在django app中创建相应的任务文件：

# app/tasks.pyfrom celery.task import Taskimport timeclass TestTask(Task): name = 'test-task' # 给任务设置个自定义名称 def run(self, *args, **kwargs):  print('start test task')  time.sleep(4)  print('args={}, kwargs={}'.format(args, kwargs))  print('end test task')

在 settings.py 添加：

INSTALLED_APPS = [ # ... 'djcelery',]# Celeryfrom learn_django.celery_config import *

触发任务或提交任务可以在view中来调用：

# views.pyfrom django.http import HttpResponsefrom app.tasks import TestTaskdef test_task(request): # 执行异步任务 print('start do request') t = TestTask() t.delay() print('end do request') return HttpResponse('ok')

启动 woker 的命令是：

python manage.py celery worker -l info