起步

在 《分布式任務(wù)隊(duì)列Celery使用說明》 中介紹了在 Python 中使用 Celery 來實(shí)驗(yàn)異步任務(wù)和定時(shí)任務(wù)功能。本文介紹如何在 Django 中使用 Celery。

安裝

pip install django-celery

這個(gè)命令使用的依賴是 Celery 3.x 的版本，所以會(huì)把我之前安裝的 4.x 卸載，不過對(duì)功能上并沒有什么影響。我們也完全可以僅用Celery在django中使用，但使用 django-celery 模塊能更好的管理 celery。

使用

可以把有關(guān) Celery 的配置放到 settings.py 里去，但我比較習(xí)慣單獨(dú)一個(gè)文件來放，然后在 settings.py 引入進(jìn)來：

# 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# 設(shè)置并發(fā)的worker數(shù)量CELERYD_CONCURRENCY = 4# 任務(wù)發(fā)送完成是否需要確認(rèn)，這一項(xiàng)對(duì)性能有一點(diǎn)影響CELERY_ACKS_LATE = True# 每個(gè)worker執(zhí)行了多少任務(wù)就會(huì)銷毀，防止內(nèi)存泄露，默認(rèn)是無限的CELERYD_MAX_TASKS_PER_CHILD = 40# 規(guī)定完成任務(wù)的時(shí)間CELERYD_TASK_TIME_LIMIT = 15 * 60 # 在15分鐘內(nèi)完成任務(wù)，否則執(zhí)行該任務(wù)的worker將被殺死，任務(wù)移交給父進(jìn)程# 設(shè)置默認(rèn)的隊(duì)列名稱，如果一個(gè)消息不符合其他的隊(duì)列就會(huì)放在默認(rèn)隊(duì)列里面，如果什么都不設(shè)置的話，數(shù)據(jù)都會(huì)發(fā)送到默認(rèn)的隊(duì)列中CELERY_DEFAULT_QUEUE = "default"# 設(shè)置詳細(xì)的隊(duì)列CELERY_QUEUES = { "default": { # 這是上面指定的默認(rèn)隊(duì)列  "exchange": "default",  "exchange_type": "direct",  "routing_key": "default" }, "beat_queue": {  "exchange": "beat_queue",  "exchange_type": "direct",  "routing_key": "beat_queue" }}

配置文件中設(shè)置了 CELERY_IMPORTS 導(dǎo)入的任務(wù)，所以在django app中創(chuàng)建相應(yīng)的任務(wù)文件：

# app/tasks.pyfrom celery.task import Taskimport timeclass TestTask(Task): name = 'test-task' # 給任務(wù)設(shè)置個(gè)自定義名稱 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 *

觸發(fā)任務(wù)或提交任務(wù)可以在view中來調(diào)用：

# views.pyfrom django.http import HttpResponsefrom app.tasks import TestTaskdef test_task(request): # 執(zhí)行異步任務(wù) print('start do request') t = TestTask() t.delay() print('end do request') return HttpResponse('ok')

啟動(dòng) woker 的命令是：

python manage.py celery worker -l info