celery结合django使用配置

1.安装插件

pip install -U "celery[redis]"
pip3 install django-celery-beat
并执行python3 manage migrate 就自动生成数据库了

2.在settings里配置

CELERY_BROKER_URL = 'redis://127.0.0.1'
CELERY_RESULT_BACKEND = 'redis://127.0.0.1'
app里添加django_celery_beat

3.和settings同级创建celery配置文件

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'video_board.settings')
app = Celery('proj')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

@app.task(bind=True)
def debug_task(self):
print('Request: {0!r}'.format(self.request))

4.在__init__文件里面

from __future__ import absolute_import, unicode_literals

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

__all__ = ['celery_app']

5.在项目里创建tasks.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


@shared_task
def xsum(numbers):
return sum(numbers)

6.进入django admin后台创建计划任务

7.views代码

from app01 import tasks
from celery.result import AsyncResult

def celery_call(request):
    
    #t = tasks.add.delay(random.randint(1,100),6)
    t = tasks.add.delay(228,28)
    #print("randomn",t.get())
    return HttpResponse(t.id)

def celery_result(request):
    task_id = request.GET.get('id')
    res = AsyncResult(id = task_id)
    if res.ready():
        return HttpResponse(res.get())
    else:
        return HttpResponse(res.ready())

8.启动celery

celery multi start w1 -A video_board -l info 后台
celery -A video_board worker -l debug 前台
celery multi stop w1 -A video_board -l info 停止celery

9.启动计划任务

celery -A celery_pro.periodic_tasks beat -l debug 项目启动计划任务
celery -A video_board beat -l debug -S django 结合django启动计划任务