欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

Django2.2中使用celery

程序员文章站 2022-10-07 21:10:13
Django2.2中使用celery1.启动Redis2.在settings中添加CELERY_BROKER_URL = 'redis://localhost:6379/0'CELERY_RESULT_BACKEND = 'redis://localhost:6379/1'CELERY_ACCEPT_CONTENT = ['application/json']CELERY_RESULT_SERIALIZER = 'json'CELERY_TASK_SERIALIZER = 'json'CEL...

Django2.2中使用celery

1.启动Redis
win10 下pip install eventlet否则报错
2.在settings中添加

CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/1'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TASK_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Shanghai'

3.在项目同名目录下创建celery.py

import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'xadmin_test.settings')  # 设置django环境

app = Celery('xadmin_test')

# 这里使用字符串以使celery的worker不用为子进程序列化配置对象。
# 命名空间 namespace='CELERY'定义所有与celery相关的配置的键名要以'CELERY_'为前缀。
app.config_from_object('django.conf:settings', namespace='CELERY') 

# 从所有django app configs中加载task模块,
# 如果你把所有的task都定义在单独的tasks.py模块中,
# 加上这句话celery会自动发现这些模块中的task,实际上这句话可以省略。
app.autodiscover_tasks()       # 发现任务文件每个app下的task.py

4.init.py

from xadmin_test.celery import app as celery_app

__all__ = ('celery_app',)

5.在app中创建task.py

from celery import shared_task
import time,datetime


@shared_task
def wait_pay(order_id):
    # now = datetime.datetime.now()
    # days = datetime.timedelta(days=1)
    # end_time = now + days
    test_num = int(order_id) + 100
    time.sleep(10)
    return test_num

6.在views中调用

def celery_test(request):
    res = tasks.wait_pay.delay(10)
    utc_time = datetime.datetime.utcfromtimestamp(执行时间.timestamp())
    res = tasks.wait_pay.apply_async(args=[10],eta=utc_time)
    return JsonResponse({'task_id':res.id})
res.get()  #获得结果
res.revoke() #取消
res.remove() #删除
from celery.result import AsyncResult
res=AsyncResult("62051878-ca77-4895-a61f-6f9525681347") # 参数为task id
res.result

7.项目启动

celery worker: celery worker -A proName -l info -P eventlet
启动Django

Django在项目启动时加载文件

from django.utils.module_loading import autodiscover_modules
autodiscover_modules('xx'
  • Django加载流程

    加载app-->autodiscover_modules-->自动发现xx.py
    

task和share_task的区别

task:

​ 将函数当作celery的任务函数

share_task:

​ 自动与所有Celery对象绑定

本文地址:https://blog.csdn.net/weixin_41615960/article/details/107314902