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