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
上一篇: 处暑时节吃啥
下一篇: python入门3-条件判断和循环