分布式异步任务框架之Celery定义、异步任务框架特点、架构、使用场景、安装配置、基本使用、多任务结构使用、延时任务、定时任务及django中使用celery

1、定义

python中的一个分布式异步任务框架
Celery是一个简单、灵活且可靠的,处理大量消息的分布式系统
专注于实时处理的异步任务队列
同时也支持任务调度
(1) 执行异步任务(对立: 同步任务),解决耗时任务,将耗时操作任务提交给Celery去异步执行,比如发送短信/邮件、消息推送、音视频处理等
(2) 执行延时任务(比如5分钟后干一件事): 解决延迟任务
(3) 执行定时任务: 每天隔几分钟干什么事,解决周期(周期)任务,比如每天数据统计Celery 官网: http://www.celeryproject.org/
Celery 官方文档英文版: http://docs.celeryproject.org/en/latest/index.html
Celery 官方文档中文版: http://docs.jinkan.org/docs/celery/


2、Celery异步任务框架特点

(1) 可以不依赖任何服务器,通过自身命令,启动服务(内部支持socket)
(2) celery服务为为其他项目服务提供异步解决任务需求的注: 会有两个服务同时运行,一个是项目服务,一个是celery服务,项目服务将需要异步处理的任务交给celery服务,celery就会在需要时异步完成项目的需求
人是一个独立运行的服务 | 医院也是一个独立运行的服务
正常情况下,人可以完成所有健康情况的动作,不需要医院的参与,但当人生病时,就会被医院接收,解决人生病问题
人生病的处理方案交给医院来解决,所有人不生病时,医院独立运行,人生病时,医院就来解决人生病的需求


3、Celery架构

Celery的架构由三部分组成&＃xff0c;消息中间件&＃xff08;message broker&＃xff09;、任务执行单元&＃xff08;worker&＃xff09;和任务执行结果存储&＃xff08;task result store&＃xff09;组成

1、消息中间件
Celery本身不提供消息服务,但是可以方便的和第三方提供的消息中间件集成,包括RabbitMQ、Redis等2、任务执行单元
Worker是Celery提供的任务执行的单元,worker并发的运行在分布式的系统节点中3、任务结果存储
Task result store用来存储Worker执行的任务的结果,Celery支持以不同方式存储任务的结果,包括AMQP, redis等


4、使用场景

异步执行: 解决耗时任务,将耗时操作任务提交给Celery去异步执行,比如发送短信/邮件、消息推送、音视频处理等
延迟执行: 解决延迟任务
定时执行: 解决周期(周期)任务,比如每天数据统计关于秒杀系统可以使用celery不能秒超,使用锁机制(mysql悲观锁,乐观锁),redis锁提高并发量 ---> 把同步做成异步 ---> 使用celery前端点击秒杀按钮,向后端发送秒杀请求 ---> 同步操作同步操作请求来到后端,判断数量是否够,如果够要生成订单(mysql),订单状态是待支付状态 请求返回,告诉前端秒杀成功异步操作请求来到后端,提交一个celery任务 ---> celery任务异步的执行判断数量是否够,如果够,要生成订单(mysql)秒杀是否成功的结果还没有,直接返回了(返回任务id)前端启动一个定时任务,每隔5s,向后台发送一个查询请求,查询秒杀任务是否执行完成(带着任务id查)如果是未执行状态或者执行中 ---> 返回给前端,前端不处理,定时任务继续执行又隔了5s,发送查询,查询到秒杀成功的结果,返回给前端,秒杀成功


5、Celery的安装配置

pip install celery
消息中间件: RabbitMQ/Redis
app&＃61;Celery(‘任务名’, broker&＃61;’xxx’, backend&＃61;’xxx’)


6、基本使用

1、定义一个py文件(t_celery.py)
import celery
# 消息中间件(redis)
broker&＃61;&＃39;redis://127.0.0.1:6379/1&＃39; # 1 表示使用redis 1 这个db
# 结果存储(redis)
backend&＃61;&＃39;redis://127.0.0.1:6379/2&＃39; # 2 表示使用redis 2 这个db# 实例化得到对象,指定中间件和结果存储
app&＃61;celery.Celery(&＃39;test&＃39;,broker&＃61;broker,backend&＃61;backend)# 定义任务(可以有多个)
&＃64;app.task
def add(a,b):return a&＃43;b&＃64;app.task
def mul(a,b):return a*b2、提交任务(在其它文件中,task.py)
from t_celery import add
res&＃61;add.delay(100,4)
print(res) # 任务id号3、启动worker
非windows平台: celery worker -A t_celery -l info
windows需要装eventlet模块: celery worker -A t_celery -l info -P eventlet4、查看执行结果
from t_celery import app
from celery.result import AsyncResult
# 关键字,变量不能定义为关键字
id &＃61; &＃39;5331c70b-1b51-4a15-aa17-2fa0f7952c00&＃39; # 定义任务的id号,第二步中res的值
if __name__ &＃61;&＃61; &＃39;__main__&＃39;:res &＃61; AsyncResult(id&＃61;id, app&＃61;app)if res.successful():result &＃61; res.get()print(result)elif res.failed():print(&＃39;任务失败&＃39;)elif res.status &＃61;&＃61; &＃39;PENDING&＃39;:print(&＃39;任务等待中被执行&＃39;)elif res.status &＃61;&＃61; &＃39;RETRY&＃39;:print(&＃39;任务异常后正在重试&＃39;)elif res.status &＃61;&＃61; &＃39;STARTED&＃39;:print(&＃39;任务已经开始被执行&＃39;)


app参数

celery配置文件参数
# 有些情况可以防止死锁
CELERYD_FORCE_EXECV&＃61;True
# 设置并发worker数量
CELERYD_CONCURRENCY&＃61;4
# 允许重试
CELERY_ACKS_LATE&＃61;True
# 每个worker最多执行100个任务被销毁,可以防止内存泄漏
CELERYD_MAX_TASKS_PER_CHILD&＃61;100
# 超时时间
CELERYD_TASK_TIME_LIMIT&＃61;12*30


7、celery多任务结构

目录结构:package_celery: # 项目名celery_task # celery包名__init__.py celery.py # celery 的app,必须叫celeryorder_task.py # 任务user_task.py # 任务result.py # 结果查询submit_tast.py # 提交任务


celery.py

import celerybroker &＃61; &＃39;redis://127.0.0.1:6379/1&＃39;
backend &＃61; &＃39;redis://127.0.0.1:6379/2&＃39;app &＃61; celery.Celery(broker&＃61;broker, backend&＃61;backend, include&＃61;[&＃39;celery_task.order_task&＃39;, &＃39;celery_task.user_task&＃39; # 一定要记得注册一下
])


order_task.py

from .celery import app&＃64;app.task
def cannel_order(name):return &＃39;用户{}取消订单&＃39;.format(name)


user_task.py

from .celery import app&＃64;app.task
def send_msg(phone):return &＃39;{}发送短信成功&＃39;.format(phone)


result.py

from celery_task.celery import appfrom celery.result import AsyncResult# 关键字,变量不能定义为关键字
# 发短信任务: 39744a3f-02ec-4b8b-b855-111025e4abe4
# 取消订单任务&＃xff1a;6d1e5e91-236a-449c-ad32-eac093b240bdid &＃61; &＃39;6d1e5e91-236a-449c-ad32-eac093b240bd&＃39;
if __name__ &＃61;&＃61; &＃39;__main__&＃39;:res &＃61; AsyncResult(id&＃61;id, app&＃61;app)if res.successful():result &＃61; res.get()print(result)elif res.failed():print(&＃39;任务失败&＃39;)elif res.status &＃61;&＃61; &＃39;PENDING&＃39;:print(&＃39;任务等待中被执行&＃39;)elif res.status &＃61;&＃61; &＃39;RETRY&＃39;:print(&＃39;任务异常后正在重试&＃39;)elif res.status &＃61;&＃61; &＃39;STARTED&＃39;:print(&＃39;任务已经开始被执行&＃39;)


submit_tast.py

from celery_task import order_task, user_taskres &＃61; order_task.cannel_order.delay(&＃39;allen&＃39;)
print(res) # 返回任务id
res &＃61; user_task.send_msg.delay(&＃39;13666666666&＃39;)
print(res) # 返回任务id


运行

# 运行worker(在package_celery目录下执行)
celery worker -A celery_task -l info -P eventlet


8、延时任务

8.1、方式一

# 发送短信为例: 2021年1月7日21点58分55秒发送短信
from datetime import datetime
from celery_task import order_task, user_task# eta: 延迟多长时间执行,eta需要传时间对象,并且是utc时间
v1 &＃61; datetime(2021, 1, 7, 21, 58, 55)
v2 &＃61; datetime.utcfromtimestamp(v1.timestamp()) # 转成utc时间,比正常时间慢8个小时
print(v2)
res &＃61; user_task.send_msg.apply_async(args&＃61;[&＃39;13666666666&＃39;, ], eta&＃61;v2)


8.2、方式二

# 发送短信为例: 以当前时间为基准,过10秒后发送短信(隔几秒后执行)
from datetime import datetime, timedeltactime &＃61; datetime.now()
utc_ctime &＃61; datetime.utcfromtimestamp(ctime.timestamp())
time_delay &＃61; timedelta(seconds&＃61;10)
task_time &＃61; utc_ctime &＃43; time_delay
res &＃61; user_task.send_msg.apply_async(args&＃61;[&＃39;13666666666&＃39;, ], eta&＃61;task_time)
print(res)


9、定时任务

# 在celery.py中配置
# 时区
app.conf.timezone &＃61; &＃39;Asia/Shanghai&＃39;
# 是否使用UTC
app.conf.enable_utc &＃61; False# 任务的定时配置
from datetime import timedelta
from celery.schedules import crontabapp.conf.beat_schedule &＃61; {&＃39;send-msg&＃39;:{&＃39;task&＃39;: &＃39;celery_task.user_task.send_msg&＃39;,&＃39;schedule&＃39;: timedelta(seconds&＃61;5), # 这里是timedelta# &＃39;schedule&＃39;: timedelta(hours&＃61;24*10), # # 这里是timedelta# &＃39;schedule&＃39;: crontab(hour&＃61;8, day_of_week&＃61;1), # 每周一早八点,这里是crontab&＃39;schedule&＃39;: crontab(hour&＃61;8, day_of_month&＃61;1), # 每月一号早八点,这里是crontab&＃39;args&＃39;: (&＃39;13666666666&＃39;,),}
}# 添加任务: 自动添加任务,所以要启动一个添加任务的服务
# 启动beat,负责每隔3s提交一个任务
celery beat -A celery_task -l info# 启动worker
celery worker -A celery_task -l info -P eventlet


10、django中使用celery()

1、celery是独立的,跟框架没有关系
2、django-celery第三方模块,兼容性不好不采用,使用通用方式
3、目录结构celery_task__init__.pycelery.pyhome_task.pyorder_task.pyuser_task.py


celery框架django项目工作流程
(1) 加载django配置环境
(2) 创建Celery框架对象app,配置broker和backend,得到的app就是worker
(3) 给worker对应的app添加可处理的任务函数,用include配置给worker的app
(4) 完成提供的任务的定时配置app.conf.beat_schedule
(5) 启动celery服务,运行worker,执行任务
(6) 启动beat服务,运行beat,添加任务重点: 由于采用了django的反射机制&＃xff0c;使用celery.py所在的celery_task包必须放置项目的根目录下


路由

path(&＃39;test_celery&＃39;, views.test_celery),


视图函数

def test_celery(request):from celery_task.celery import appfrom celery_task.user_task import send_msgfrom celery.result import AsyncResultid &＃61; request.GET.get(&＃39;id&＃39;)if id:res &＃61; AsyncResult(id&＃61;id, app&＃61;app)if res.successful():result &＃61; res.get()return HttpResponse(result)id &＃61; send_msg.delay(&＃39;13666666666&＃39;)print(id)return HttpResponse(&＃39;发送短信成功&＃39;)