Celery 官网：http://www.celeryproject.org/
Celery 官方文档英文版：http://docs.celeryproject.org/en/latest/index.html
Celery 官方文档中文版：http://docs.jinkan.org/docs/celery/
正常情况下，人可以完成所有健康情况的动作，不需要医院的参与；但当人生病时，就会被医院接收，解决人生病问题
人生病的处理方案交给医院来解决，所有人不生病时，医院独立运行，人生病时，医院就来解决人生病的需求
celery本身不提供消息服务，但是可以与第三方提供的消息中间件组成，包括RabbitMQ，Redis等等
worker是celery提供的任务执行单元，worker并发的运行在分布式的系统节点中
task result store用来存储worker执行的任务的结果，celery支持以不同的方式存储任务的结果，包括AMQP， redis等
异步执行：解决耗时任务，视频同步
延迟执行：解决延迟任务，定时发送邮件
定时执行：解决周期或者间隔任务，定时更新
pip install celery
1、celery对象：Celery(...)是放在一个模块下
1. 终端切换到文件夹所对应的位置
2.执行启动worker的命名：celery worker -A 模块名 -l info -P eventlet
注：windows系统需要eventlet支持，Linux与MacOS直接执行：celery worker -A 模块名 -l info
2、如果celery对象：Celery(...)是放在一个包下
1.必须在包下建一个celery.py文件，将Celery(...)产生对象的语句放在该文件中
2.执行启动worker的命名：celery worker -A 包名 -l info -P eventlet
注：windows系统需要eventlet支持，Linux与MacOS直接执行：celery worker -A 包名 -l info
注：包名随意
project ├── celery_task # celery包 │ ├── __init__.py # 包文件 │ ├── celery.py # celery连接和配置相关文件，且名字必须交celery.py │ └── tasks.py # 所有任务函数 ├── add_task.py # 添加任务 └── get_result.py # 获取结果
1、创建app + 任务tasks
celery.py
from celery import Celery broker = ‘redis://127.0.0.1:6379/1‘ backend = ‘redis://127.0.0.1:6379/2‘ app = Celery(broker=broker, backend=backend, include=[‘celery_task.tasks‘])
tasks.py
from .celery import app import time @app.task def plus_task(n, m): print(n) print(m) time.sleep(10) print(‘n+m的结果：%s‘ % (n + m)) return n + m
2、启动celery服务：终端切换到包的文件夹
非windows # 命令：celery worker -A 包名 -l info windows： # pip3 install eventlet # celery worker -A 包名 -l info -P eventlet
3、添加任务：手动添加，要自定义添加任务的脚本，右键run执行脚本
add_task.py
from celery_task import tasks # 添加立即执行任务 t1 = tasks.plus_task.delay(10, 20)
print(t1.id)
print(t1) # 就是id # 添加延迟任务 from datetime import datetime, timedelta eta=datetime.utcnow() + timedelta(seconds=10) # 配置时区 tasks.plus_task.apply_async(args=(200, 50), eta=eta)
4、获取结果：手动获取，要自定义获取任务的脚本，右键run执行脚本
get_result.py
from celery_task.celery import app from celery.result import AsyncResult id = ‘21325a40-9d32-44b5-a701-9a31cc3c74b5‘ if __name__ == ‘__main__‘: async = AsyncResult(id=id, app=app) 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中配置好beat_schedule的配置
# 定时任务的配置 ‘‘‘ app.conf.beat_schedule = { ‘自定义定时任务名‘: { ‘task‘: ‘指向任务函数‘, ‘schedule‘: ‘下一次再添加任务的时间间隔或固定的时间配置‘, ‘args‘: ‘无名参数‘, ‘kwargs‘: ‘有名参数‘ } } ‘‘‘
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
‘low-task‘: {
‘task‘: ‘celery_task.tasks.add_task‘,
# ‘schedule‘: timedelta(seconds=3), # 每3秒执行一次
# ‘schedule‘: crontab(hour=8, day_of_week=1), # 每周一早八点
‘schedule‘: crontab(month_of_year=8, day_of_month=1, hour=12, minute=0), # 每年8月1日12点0分添加执行一次
‘args‘: (300, 150),
2、执行命令启动定时添加任务服务：celery beat -A 包名|模块名 -l info
不用使用 -P eventlet
重点：由于采用了django的反射机制，使用celery.py所在的celery_task包必须放置项目的根目录下
如果不想把celery_task包放在项目根目录，必须添加如下几句，将项目根目录要添加到环境变量中，
import sys sys.path.append(r‘C:\Users\owen\Desktop\luffy\luffyapi‘)
1、celery.py中加载django配置环境
# 一、加载django配置环境 import os os.environ.setdefault("DJANGO_SETTINGS_MODULE", "luffyapi.settings.dev")
2、创建Celery框架对象app，配置broker和backend，得到的app就是worker
3、完成提供的任务的定时配置app.conf.beat_schedule
# 二、加载celery配置环境 from celery import Celery # broker broker = ‘redis://127.0.0.1:6379/0‘ # backend backend = ‘redis://127.0.0.1:6379/1‘ # worker app = Celery(broker=broker, backend=backend, include=[‘celery_task.tasks‘]) # 时区 app.conf.timezone = ‘Asia/Shanghai‘ # 是否使用UTC app.conf.enable_utc = False # 任务的定时配置 from datetime import timedelta from celery.schedules import crontab app.conf.beat_schedule = { ‘update-banner-list‘: { ‘task‘: ‘celery_task.tasks.update_banner_list‘, ‘schedule‘: timedelta(seconds=10), ‘args‘: (), } }
4、切换到项目根目录下启动celery服务，运行worker，执行任务
celery worker -A celery_task -l info -P eventlet
5、切换到项目根目录下启动beat服务，运行beat，添加任务
celery beat -A celery_task -l info
原文：https://www.cnblogs.com/Mr-shen/p/12411479.html