目录
Celery介绍、安装、基本使用
一、Celery服务
什么是Celery:
Celery是一个简单、灵活且可靠的,处理消息的分布式系统
- Celery可以用来做什么:
- 异步任务
- 定时任务
- 延迟任务
Celery的运行原理:
- 可以不依赖任何服务,通过自身命令,启动服务
- celery服务为其他项目服务提供异步解决任务需求
# 注:会有两个服务同时运行
- 项目服务
- celery服务
项目服务将需要异步处理的任务交给celery服务,celery就会在需要时异步完成项目的需求
'''
人是一个独立运行的服务 | 医院也是一个独立运行的服务
正常情况下,人可以完成所有健康情况的动作,不需要医院的参与;但当人生病时,就会被医院接收,解决人生病问题
人生病的处理方案交给医院来解决,所有人不生病时,医院独立运行,人生病时,医院就来解决人生病的需求
'''
1、celery架构
-
消息中间件:broker
- 提交的任务【函数】都放在这里, celery本身不能提供消息中间件
- 需要借助于第三方: redis或rabbitmq
-
任务执行单元:worker
- 真正执行任务的的地方,一个个进程中执行函数
-
结果储存:backend
- 函数return的结果都存储在这里, celery本身不提供结果存储
- 需要借助于第三方: redis或rabbitmq
使用场景:
- 异步执行:解决耗时任务
- 延迟执行:解决延迟任务
- 定时执行:解决周期任务
2、celery快速使用
Celery不支持在windows上直接运行,通过eventlet支持在win上运行
安装:
pip install celery
pip install eventlet # windows需要安装
快速使用:
- 1、第一步:创建一个py文件(main.py),用于实例化celery对象,编写需要执行的函数
# 1、导入模块
from celery import Celery
# 2、指定briker,用于存放提交的异步任务
broker = 'redis://127.0.0.1:6379/1'
# 3、指定backend,用于存放函数执行结束的结果
backend = 'redis://127.0.0.1:6379/2'
# 实例化celery对象
app = Celery('test', broker=broker, backend=backend)
# 编写一个函数,装饰上celery对象
@app.task
def add(a, b):
import time
time.sleep(3)
print('add函数执行完成')
return a + b
- 2、第二步:再次创建一个py文件(run.py),用于将函数提交给celery
# 1、导入刚才编写的函数
from main import add
# 2、将任务提交给broker,函数需要的参数需要传入
res = add.delay(1, 2)
# 3、提交后可以获得该任务的ID,可通过ID可以查询任务执行结果
print(res) # 0213d2c2-453e-41a8-a171-e31f1f2f4883
- 3、第三步:使用命令开启worker (也可以提前开启,任务提交后就会直接执行)
# 启动worker命令,win需要安装eventlet
# 启动需要进入main.py文件的目录下
win:
-4.x之前版本
celery worker -A main -l info -P eventlet
-4.x之后
celery -A main worker -l info -P eventlet
mac:
celery -A main worker -l info
- 4、第四步:worker会将执行的结果存在之前指定的broker目录下(指定的redis数据库)
- 5、第五步:通过代码查看执行结果(创建新的py文件,专门用于查看执行结果)
# 1、导入celery实例的对象
from main import app
# 2、导入该模块用于查看结果
from celery.result import AsyncResult
# 3、将提交的任务编号拿过来,用于查询结果
id = '0213d2c2-453e-41a8-a171-e31f1f2f4883'
# 4、指定该文件为启动文件
if __name__ == '__main__':
# 实例化对象,将任务的ID和celery实例化对象当作参数传入
a = AsyncResult(id=id, app=app)
# 判断执行结果
if a.successful(): # 执行完了
result = a.get()
print(result)
elif a.failed():
print('任务失败')
elif a.status == 'PENDING':
print('任务等待中被执行')
elif a.status == 'RETRY':
print('任务异常后正在重试')
elif a.status == 'STARTED':
print('任务已经开始被执行')
二、Celer包结构
1、创建clery包结构
什么是包结构:通过将celery服务封装成包的形式,放在项目需要使用的时候导入即可
project
├── celery_task # celery包
│ ├── __init__.py # 包文件
│ ├── celery.py # celery连接和配置相关文件,且名字必须交celery.py
│ └── tasks.py # 所有任务函数
├── add_task.py # 添加任务
└── get_result.py # 获取结果
创建包:
创建一个包,名为:celery_task
- 1、第一步:在包下创建py文件(名字必须为celery.py)
# 导入celery模块
from celery import Celery
# 导入配置broker和backend
from .settings import BACKEND, BROKER
# 实例化celery对象
app = Celery('test',
broker=BROKER,
backend=BACKEND,
include=['celery_task.order_task',
'celery_task.user_task'])
- 2、第二步:创建settings.py,用于存放配置
BROKER = 'redis://127.0.0.1:6379/1'
BACKEND = 'redis://127.0.0.1:6379/2'
- 3、第三步,创建py文件(task.py),用于存放需要执行的异步任务
# 导入celery实例对象
from .celery import app
# 计算函数
@app.task()
def add(a, b):
print('计算结果为:', a + b)
return True
# 模拟发送短信