标签：font ret second 固定 include 重试 timezone import 结构
一、创建celery包
1、新建一个包，任意命名，如，celery_task。
2、包内创建celery对象，指定命名为celery.py：
from celery import Celery broker = ‘redis://127.0.0.1:6379/1‘ backend = ‘redis://127.0.0.1:6379/2‘ app = Celery(__name__, broker=broker, backend=backend, include=[‘celery_task.task1‘, ‘celery_task.task2‘, ])
3、包内创建任务，以task1.py为例：
from .celery import app @app.task def add(x, y): return x + y
二、异步任务和延时任务
1、启动worker：celery worker -A celery_task -l info (-P eventlet)
2、投放任务：
from datetime import datetime, timedelta from celery_task.task1 import add ret = add.delay(10, 5) # 返回任务id eta = datetime.utcnow() + timedelta(seconds=10) # 默认要用utc时间，延时10秒 ret2 = add.apply_async(args=(60, 30), eta=eta) # 传入任务参数和执行时间，返回任务id
3、获取任务执行结果：
from celery_task import app from celery.result import AsyncResult id = ‘3e397fd7-e0c1-4c5c-999c-2655a96793bb‘ # 任务id if __name__ == ‘__main__‘: async = AsyncResult(id=id, app=app) # 填入任务编号和celery对象 if async.successful(): result = async.get() print(result) elif async.failed(): print(‘任务失败‘) elif async.status == ‘PENDING‘: print(‘任务等待中被执行‘) elif async.status == ‘RETRY‘: print(‘任务异常后正在重试‘) elif async.status == ‘STARTED‘: print(‘任务已经开始被执行‘)
三、定时任务
1、celery.py：
from datetime import timedelta from celery.schedules import crontab from celery import Celery broker = ‘redis://127.0.0.1:6379/1‘ backend = ‘redis://127.0.0.1:6379/2‘ app = Celery(__name__, broker=broker, backend=backend, include=[‘celery_task.task1‘, ‘celery_task.task2‘]) # 任务的定时配置 app.conf.timezone = ‘Asia/Shanghai‘ # 更换市区 app.conf.enable_utc = False # 不使用utc app.conf.beat_schedule = { ‘add-task‘: { ‘task‘: ‘celery_task.task1.add‘, # 指定任务 # ‘schedule‘: timedelta(seconds=3), # 每隔固定时间执行 ‘schedule‘: crontab(hour=8, day_of_week=1), # 每周一早八点 ‘args‘: (300, 150), # 任务参数 }, }
2、启动beat，启动worker：
①beat：启动后待命，到指定时间自动投放任务。
②worker：celery worker -A celery_task -l info (-P eventlet)
标签：font ret second 固定 include 重试 timezone import 结构
原文地址：https://www.cnblogs.com/caoyu080202201/p/13386400.html
