Django Celery的配置

1. 在app目录下建task.py文件

   from __future__ import absolute_import, unicode_literals
   from celery import shared_task
   
   
   @shared_task
   def add(x, y):
       return x + y
   
   
   @shared_task
   def mul(x, y):
       return x * y
   

2. 在项目setting.py所在目录下建celery.py文件

   #!/usr/bin/env python3
   # -*- coding:utf-8 -*-
   # Author:wd
   from __future__ import absolute_import, unicode_literals
   import os
   from celery import Celery
   
   
   os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'taskproj.settings')  # 设置django环境
   
   app = Celery('taskproj')
   
   app.config_from_object('django.conf:settings', namespace='CELERY') #  使用CELERY_ 作为前缀，在settings中写配置
   
   app.autodiscover_tasks()  # 发现任务文件每个app下的task.py
   

3. 在setting.py中配置Broker和backend信息

   CELERY_BROKER_URL = 'redis://10.1.210.69:6379/0' # Broker配置，使用Redis作为消息中间件
   
   CELERY_RESULT_BACKEND = 'redis://10.1.210.69:6379/0' # BACKEND配置，这里使用redis
   
   CELERY_RESULT_SERIALIZER = 'json' # 结果序列化方案
   

4. 在项目文件夹中的__init__.py中配置celery app

   from __future__ import absolute_import, unicode_literals
   from .celery import app as celery_app
   __all__ = ['celery_app']
   

5. 启动worker

   celery worker -A taskproj -l debug
   

6. view.py中调用task

   from django.http import JsonResponse
   from app01 import tasks
   
   # Create your views here.
   
   def index(request,*args,**kwargs):
       res=tasks.add.delay(1,3)
       #任务逻辑
       return JsonResponse({'status':'successful','task_id':res.task_id})
   

7. 参考自：
   博客园：W-D