celery介绍
Posted
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了celery介绍相关的知识，希望对你有一定的参考价值。
参考技术A celery是一个分布式消息队列，支持异步定时调用以rabbitMQ和redis为主
apply_async 调用异步任务
countdown: 任务延迟执行的秒数，默认立即执行
eta：任务被执行的绝对时间
crontab 计划任务
Celery框架的基本使用与介绍
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(\'任务已经开始被执行\')
3、celer包结构【使用包写一个小游戏】
什么是包结构：通过将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
# 模拟发送短信
@app.task()
def send_sms(mobile, code):
print(\'已向手机号：%s 发送短信，验证码为：%s\' % (mobile, code))
return True
- 4、第四步：开启worker
切换到celery所在的目录下，开启worker命令
- 5、第五步：提桥任务：
# 导入任务
from celery_bag.celery_task.task import send_sms, add
# 提交任务
def add_func(a, b):
return add.delay(a, b)
def send_func(mobile, code):
return send_sms.delay(mobile, code)
- 6、触发任务提交：
import os
import redis
from add_task import add_func, send_func
from get_result import res_func
POOL = redis.ConnectionPool(max_connections=100)
conn = redis.Redis(connection_pool=POOL)
func_list =
\'1\': add_func,
\'2\': send_func
if __name__ == \'__main__\':
while True:
print(\'\'\'
1、异步计算器
2、模拟发送短信
3、查看任务执行状态
4、开启worker，并查看任务执行结果（需要重启系统）
\'\'\')
user_choice = input(\'欢迎来到celery测试系统，请输入您需要执行的功能编号>>>:\').strip()
while user_choice in func_list.keys():
if user_choice == \'1\':
print(\'已进入异步计算器功能！\')
a = \'请输入数字 1 \'
b = \'请输入数字 2 \'
else:
print(\'您已进入模拟发送短信功能\')
a = \'手机号\'
b = \'短信内容 \'
args_1 = input(\'请输入%s>>>:\' % a).strip()
args_2 = input(\'请输入%s>>>:\' % b).strip()
args_1 = int(args_1)
args_2 = int(args_2)
res = func_list.get(user_choice)(args_1, args_2)
input(\'\'\'
任务提交完成！
请记录本次任务ID：%s
任意键返回上一层
\'\'\' % res)
break
if user_choice == \'3\':
id = input(\'请输入任务ID>>>:\')
res = res_func(id)
print(res)
continue
while user_choice == \'4\':
try:
os.system(\'CD D:\\djangoProject\\luffy_api\\celery_bag\\celery_task\')
os.system(\'celery -A celery_task worker -l info -P eventlet\')
except Exception as e:
print(\'开始失败，出现错误,请重启系统\')
print(str(e))
break
print(\'自动返回上一层\')
break
else:
print(\'编号输入有误，请重写输入\')
continue
- 7、第七步：查看任务执行结果：
# 导入celery实例
from celery_task.celery import app
from celery.result import AsyncResult
def res_func(id):
id = id
a = AsyncResult(id=id, app=app)
if a.successful(): # 执行完了
result = a.get()
if result: return \'执行完成\'
elif a.failed():
return \'任务失败,失败的原因可能是未开启worker\'
elif a.status == \'PENDING\':
return \'任务等待中被执行，当前任务较多或未开启worker\'
elif a.status == \'RETRY\':
return \'任务异常后正在重试\'
elif a.status == \'STARTED\':
return \'任务已经开始被执行，请稍后查询\'
以上是关于celery介绍的主要内容，如果未能解决你的问题，请参考以下文章