首页 > temp > python入门教程 >
-
Celery异步分布队列
一、Celery介绍celery periodic task
Celery 是一个基于python开发的分布式异步消息任务队列,通过它可以轻松的实现任务的异步处理,如果你的业务场景中需要用到异步任务,就可以考虑使用celery,例:
1.你想对100台机器执行一条批量命令,可能会花很长时间,但不想让程序等着结果返回,而是给你返回一个任务ID,你过一段时间只需要拿着这个任务id就可以拿到任务执行结果,在任务执行ing进行时,你可以继续做其它的事情。
2.你想做一个定时任务,比如每天检测一下你们所有客户的资料,如果发现今天是客户的生日,就给他发个短信祝福
Celery在执行任务时需要通过一个消息中间件和发送任务消息,以及存储任务结果,一般使用rabbitMQ or Redis
Celery有以下优点:
- 简单:一旦熟悉了celery的工作流程后,配置和使用还是比较简单的
- 高可用:当任务执行失败或执行过程中发生链接中断,celery会自动尝试重新执行任务
- 快速:一个单进程的Celery每分钟可处理上百万个任务
- 灵活:几乎celery的各个组件都可以被扩展及自定制
任务队列是一种跨线程、跨机器工作的一种机制.
任务队列中包含称作任务的工作单元。有专门的工作进程持续不断的监视任务队列,并从中获得新的任务并处理.
celery通过消息(任务)进行通信,通常使用一个叫Broker(中间人)来协助clients(任务的发出者)和worker(任务的处理者). clients发出消息到队列中,broker将队列中的信息派发给worker来处理。
Celery的架构
Celery的架构由三部分组成,消息队列(message broker),任务执行单元(worker)和任务执行结果存储(task result store)组成。
一个celery系统可以包含很多的worker和broker
Celery本身不提供消息队列功能,但是可以很方便地和第三方提供的消息中间件进行集成,包括RabbitMQ,Redis,MongoDB等
二、安装
pip install -U celery==4.4.7 -i https://pypi.tuna.tsinghua.edu.cn/simple
注意:
Celery不建议在windows系统下使用,因为Celery在4.0版本以后,不再支持windows系统,所以如果要在windows下使用只能安装4.0以前的版本,而且即便是4.0之前的版本,在windows系统下也是不能单独使用的,需要安装eventlet模块
也可从官方直接下载安装包:https://pypi.python.org/pypi/celery/
tar xvfz celery-4.4.7.tar.gz
cd celery-4.4.7
python setup.py build
python setup.py install
三、使用
使用celery第一件要做的最为重要的事情是需要先创建一个Celery实例,我们一般叫做celery应用,或者更简单直接叫做一个app。app应用是我们使用celery所有功能的入口,比如创建任务,管理任务等,在使用celery的时候,app必须能够被其他的模块导入。
一般celery任务目录直接放在项目的根目录下即可,路径:
luffyapi/
├── mycelery/
├── config.py # 配置文件
├── __init__.py
├── main.py # 主程序
└── sms/ # 一个目录可以放置多个任务,该目录下存放当前任务执行时需要的模块或依赖
└── tasks.py # 任务的文件,名称必须是这个!!!
main.py,代码:
# 主程序
from celery import Celery
# 创建celery实例对象
app = Celery("luffy")
# 通过app对象加载配置
app.config_from_object("mycelery.config")
# 自动搜索并加载任务
# 参数必须必须是一个列表,里面的每一个任务都是任务的路径名称
# app.autodiscover_tasks(["任务1","任务2",....])
app.autodiscover_tasks(["mycelery.sms","mycelery.email"])
# 启动Celery的命令
# 强烈建议切换目录到项目的根目录下启动celery!!
# celery -A mycelery.main worker --loglevel=info
配置文件config.py,代码:
# 任务队列的链接地址
broker_url = 'redis://127.0.0.1:6379/15'
# 结果队列的链接地址
result_backend = 'redis://127.0.0.1:6379/14'
创建一个任务文件sms/tasks.py,并创建任务,代码:
# celery的任务必须写在tasks.py的文件中,别的文件名称不识别!!!
from mycelery.main import app
@app.task # name表示设置任务的名称,如果不填写,则默认使用函数名做为任务名
def send_sms():
print("发送短信!!!")
@app.task(name="send_sms2") # name表示设置任务的名称,如果不填写,则默认使用函数名做为任务名
def send_sms2():
print("发送短信任务2!!!")
在程序中调用上面的异步任务,拿django进行举例:
# 调用celery执行异步任务
from my_celery.sms.tasks import send_sms
send_sms.delay(mobile)
其他参考文档:
http://docs.celeryproject.org/en/latest/getting-started/introduction.html
https://github.com/celery/celery/tree/master/examples/django/
https://www.jianshu.com/p/1840035cb510
https://flower.readthedocs.io/en/latest/screenshots.html
接下来,我们需要把celery和django组合起来一起使用。
把django和celery进行组合
在main.py主程序中对django的配置文件进行加载
import os,django
from celery import Celery
# 初始化celery对象
app = Celery("luffy")
# 初始化django
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'luffyapi.settings.dev')
django.setup()
# 加载配置
app.config_from_object("mycelery.config")
# 自动注册任务
app.autodiscover_tasks(["mycelery.sms","mycelery.email"])
# 运行celery
# 终端下: celery -A mycelery.main worker -l info
在需要使用django配置的任务中,直接加载配置,所以我们把注册的短信发送功能,整合成一个任务函数,mycelery.sms.tasks,代码:
import json,logging
from mycelery.main import app
from ronglian_sms_sdk import SmsSDK
from django.conf import settings
from luffyapi.settings import constants
log = logging.getLogger("django")
@app.task(name="send_sms")
def send_sms(mobile,code):
"""发送短信"""
sdk = SmsSDK(
settings.SMS.get("accId"),
settings.SMS.get("accToken"),
settings.SMS.get("appId"),
)
try:
resp = sdk.sendMessage(settings.SMS.get("TempId"), mobile, (code, constants.SMS_EXPIRE_TIME // 60))
resp_data = json.loads(resp)
if resp_data.get("statusCode") != "000000":
raise Exception
except Exception as exc:
log.error("短信发送失败! 手机号:%s: %s" % (mobile,exc) )
raise Exception
在这个任务中,我们需要加载短信发送的sdk和相关的配置常量,所以我们可以直接把django中的短信发送模块和相关的常量配置文件直接剪切到当前sms任务目录中
mycelery/
├── config.py
├── __init__.py
├── main.py
└── sms/
├── __init__.py
├── tasks.py
再次启动项目即可。
最终在django里面,我们调用Celery来异步执行任务。需要完成2个步骤:
# 1. 声明一个和celery一模一样的任务函数,但是我们可以导包来解决
from mycelery.sms.tasks import send_sms
# 2. 调用任务函数,发布任务
send_sms.delay(mobile,code)
# send_sms.delay() 如果调用的任务函数没有参数,则不需要填写任何内容
四、定时任务
定时任务[crontab],主要是依靠:操作系统的定时计划或者第三方软件的定时执行
定时任务的常见场景:
1. 订单超时
2. 生日邮件[例如,每天凌晨检查当天有没有用户生日,有则发送一份祝福邮件]
3. 财务统计[例如,每个月的1号,把当月的订单进行统计,生成一个财务记录,保存到数据库中]
4. 页面缓存[例如,把首页设置为每隔5分钟生成一次缓存]
在django中要实现订单的超时取消,有以下三种方式:
1. Celery本身提供了定时任务的schedules执行机制
2. 第三方模块 django-crontab 定时任务
3. redis值空间回调事件[实际上就是监控字符串的键在过期时,让python自动执行对应的代码段或者方法函数]
https://blog.csdn.net/qq_42874635/article/details/107314161
Celery官方文档中关于定时任务使用的说明:
http://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html
在实现定时任务之前,我们需要先简单使用一下。
我们需要新增一个任务目录,例如order
myceley/
├── sms/
│ ├── __init__.py
│ └── tasks.py
├── config.py
├── __init__.py
├── main.py
├── order/
│ ├── __init__.py
│ └── tasks.py
└── sms
在main.py中,注册任务目录【注意,接下来后面我们使用django的模型处理,所以必须对django的配置进行引入】
import os
from celery import Celery
# 1. 创建示例对象
app = Celery("luffy")
# 2. 加载配置
app.config_from_object("mycelery.config")
# 3. 注册任务[自动搜索并加载任务]
# 参数必须必须是一个列表,里面的每一个任务都是任务的路径名称
# app.autodiscover_tasks(["任务1","任务2"])
app.autodiscover_tasks(["mycelery.sms","mycelery.order"])
# 4. 在终端下面运行celery命令启动celery
# celery -A 主程序 worker --loglevel=info
# celery -A mycelery.main worker --loglevel=info
接下来,在order任务目录下, 创建固定名字的任务文件tasks.py,代码:
from my_celery.main import app
@app.task(name="check_order")
def check_order():
print("检查订单是否过期!!!")
接下来,我们需要把这个任务设置定时任务,所以需要借助Celery本身提供的Crontab模块。
在配置文件中,对定时任务进行注册:
# 任务队列地址
broker_url = 'redis://127.0.0.1:6379/15'
# 结果队列地址
result_backend = "redis://127.0.0.1:6379/14"
# 设置时区
timezone = 'Asia/Shanghai'
# 使用定时任务必须确保设置了时区
from .main import app
from celery.schedules import crontab
app.conf.beat_schedule = {
'check-order': {
'task': 'check_order', # 定时执行的任务
'schedule': 30.0, # 定时执行的频率,表示每30秒执行一次
# 'schedule': crontab(minute="*/5"), # 定时执行的频率,
# 'args': (16, 16) # 定时的任务参数
},
}
接下来,我们就可以重启Celery并启用Celery的定时任务调度器
先在终端下,运行celery的定时任务程序,以下命令:
celery -A mycelery.main beat # mycelery.main 是celery的主应用文件
然后再新建一个终端,运行以下命令,上面的命令必须先指定:
celery -A mycelery.main worker --loglevel=info
注意,使用的时候,如果有时区必须先配置好系统时区。
经过上面的测试以后,我们接下来只需改造上面的任务函数,用于判断修改订单是否超时。
要完成订单的任务功能,如果需要调用django框架的模型操作,那么必须针对django框架进行配置加载和初始化。
main.py,代码:
import os,django
from celery import Celery
# 初始化celery对象
app = Celery("luffy")
# 初始化django
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'luffyapi.settings.dev')
django.setup()
# 加载配置
app.config_from_object("mycelery.config")
# 自动注册任务
app.autodiscover_tasks(["mycelery.sms","mycelery.email","mycelery.order"])
# 运行celery
# 终端下: celery -A mycelery.main worker -l info
注意,因为在django中是有时区配置的,所以,我们在django框架配置中也要修改时区配置。
任务代码tasks.py的实现:
from mycelery.main import app
@app.task(name="check_order")
def check_order():
"""订单超时取消(业务逻辑)"""
pass
重新启动celery的定时任务模块和celery的主应用程序。