celery：celery介绍架构基本使用，celery执行异步任务延迟任务定时任务，django中使用celery。
Posted 爱思考的实践者
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了celery：celery介绍架构基本使用，celery执行异步任务延迟任务定时任务，django中使用celery。相关的知识，希望对你有一定的参考价值。
一、 celery介绍
Celery 官网：Celery - Distributed Task Queue — Celery 5.2.7 documentation
Celery 官方文档英文版：Celery - Distributed Task Queue — Celery 5.2.7 documentation
# celery:芹菜，一个分布式的异步任务框架。
⭐celery能干什么：异步任务，延迟任务，定时任务。
-异步执行：解决耗时任务，将“耗时操作任务”提交给Celery去异步执行，比如发送短信/邮件、消息推送、音视频处理等。
-延迟执行：解决延迟任务。
-定时执行：解决周期(周期)任务，比如每天数据统计。
"""
1）可以不依赖任何服务器，通过自身命令，启动服务。
2）celery服务为其他项目服务提供“异步处理任务”的功能。
注：有两个服务同时运行，一个是项目服务，一个是celery服务。项目服务将“需要异步处理的任务”交给celery服务，celery会在需要时异步完成项目的需求。
人是一个独立运行的服务。 | 医院也是一个独立运行的服务。
正常情况下，人可以完成所有健康情况的动作，不需要医院的参与；但当人生病时，就会被医院接收，解决人生病问题。
人生病的处理方案交给医院来解决，所有人不生病时，医院独立运行；人生病时，医院就来解决人生病的需求。
"""
Celery is a project with minimal funding, so we don’t support Microsoft Windows. Please don’t open any issues related to that platform.
Sanic：python3.5以上，不支持win。
-选择使用mac开发
-装ubuntu
-win远程连接linux开发
uwgi（不支持win）+ django
二、 celery架构
celery架构中的几个主要组件为：
消息中间件(broker): Celery本身不提供消息服务，但是可以方便地和第三方提供的消息中间件集成，包括 Redis 等。
任务执行单元(worker): Worker是Celery提供的任务执行的单元，并发运行在分布式的系统节点中(本质：一个work就是一个进程)。
任务结果存储（backend）: Task result store用来存储Worker执行的任务的结果，Celery支持以不同方式存储任务的结果，包括 redis 等。Celery异步任务框架，是一个独立运行的服务（内置socket）。
使用Celery框架的步骤为：
1.安装Celery框架环境，启动Celery服务（需要提前配置Broker和Backend）；
2.手动或自动添加任务到Broker中，Worker就会自动在后台异步执行任务；
3.从Backend中获取任务执行结果。
Celery具体工作流程如下图所示：
三、 celery基本使用
# 安装：
pip install celery
# window 上启动 work
## 4.x版本及之前
pip install eventlet
celery worker -A celery_task -l info -P eventlet
## 5.x版本
pip install eventlet
celery -A celery_task worker -l info -P eventlet
# linux 中启动 work
celery worker -A celery_task -l info
celery -A celery_task worker -l info
# 启动 beat 的命令（负责每隔几秒钟，向任务队列中提交任务）
celery beat -A celery_task -l info
两种celery任务结构：一种是放在模块下，另一种是放在包下。提倡用包管理，结构更清晰。
如果 Celery对象:Celery(...) 放在一个模块下，则：
1）终端切换到该模块所在文件夹位置：scripts
2）执行启动worker的命令：celery worker -A 模块名 -l info -P eventlet
注：windows系统需要eventlet支持，Linux与MacOS直接执行：celery worker -A 模块名 -l info
注：模块名随意。
如果 Celery对象:Celery(...) 放在一个包下，则：
1）必须在这个包下建一个celery.py的文件，将Celery(...)产生对象的语句放在该文件中
2）执行启动worker的命令：celery worker -A 包名 -l info -P eventlet
注：windows系统需要eventlet支持，Linux与MacOS直接执行：celery worker -A 模块名 -l info
注：包名随意。
3.1 使用模块方式
新建celery_task.py
from celery import Celery
backend='redis://127.0.0.1:6379/1' # 结果存储
broker='redis://127.0.0.1:6379/2' # 消息中间件
# app=Celery(‘任务名’, broker=’xxx’, backend=’xxx’)
app=Celery('test', broker=broker, backend=backend) # 传一个字符串，相当于名字
@app.task
def add(a, b): # 很耗时的任务
import time
time.sleep(3)
return a + b
新建add_task.py添加任务
from celery_task import add
# res=add(1, 2) # 同步调用
res=add.delay(3,4) # 把任务提交到redis，系统返回任务uuid：b20f827d-dcf5-4fdf-a646-5abb963dc1d3
print(res)
新建get_result.py查询任务
from celery_task import app # 自己写的app
from celery.result import AsyncResult # celery模块下的
id = 'b20f827d-dcf5-4fdf-a646-5abb963dc1d3'
if __name__ == '__main__':
a = AsyncResult(id=id, app=app)
if a.successful():
result = a.get() # task中return的数据:7
print(result)
elif a.failed():
print('任务失败')
elif a.status == 'PENDING':
print('任务等待中，尚未被执行')
elif a.status == 'RETRY':
print('任务异常后正在重试')
elif a.status == 'STARTED':
print('任务已经开始被执行')
启动worker
# windows中启动work
pip install eventlet
celery -A celery_task worker -l info -P eventlet
# linux中启动work
celery -A celery_task worker -l info
3.2 使用包方式
celery_task # 包名
├── __init__.py # 包初始化文件
├── user_task.py # 任务存放路径
├── order_task.py # 任务存放路径
├── home_task.py # 任务存放路径
└── celery.py # app所在文件，celery连接和配置相关文件，必须叫celery.py
add_task.py # 别的服务提交任务
get_result.py # 别的服务获取结果
celery.py 文件
### 有app的这个py文件，必须叫celery
from celery import Celery
backend = 'redis://127.0.0.1:6379/1' # 结果存储
broker = 'redis://127.0.0.1:6379/2' # 消息中间件
app = Celery('test', broker=broker, backend=backend, include=[
'celery_task.home_task',
'celery_task.order_task',
'celery_task.user_task'
])
# 定制定时任务
# 时区(修改时区)
print(app.conf)
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False
# 任务的定时配置
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule =
'send_sms_5':
'task': 'celery_task.user_task.send_sms', # 要执行的任务
'schedule': timedelta(seconds=5),
# 'schedule': crontab(hour=8, day_of_week=1), # 每周一早八点
'args': (189533333,),
,
'add_3':
'task': 'celery_task.home_task.add', # 要执行的任务
'schedule': timedelta(seconds=3),
# 'schedule': crontab(hour=8, day_of_week=1), # 每周一早八点
'args': (6,8),
home_task.py 文件
from .celery import app
@app.task # home模块的任务
def add(a,b):
return a+b
order_task.py 文件
from .celery import app
@app.task # order模块的任务,写文件任务
def write_file(s):
with open(r'./test/log/a.txt','w', encoding='utf-8') as f:
f.write(s)
return Trueuser_task.py
from .celery import app
@app.task # user模块的任务, 发送短信
def send_sms(s):
import time
time.sleep(3)
return '%s手机号，短信发送成功'%s
四、 celery执行异步任务、延迟任务、定时任务
4.1 异步任务
from celery_task.user_task import send_sms
res=send_sms.delay('18953675221')
print(res)
4.2 延迟任务
# 延迟任务，延迟5s钟后，发送短信
from celery_task.user_task import send_sms
from datetime import datetime, timedelta
# print(datetime.utcnow()+timedelta(seconds=10)) # 打印出当前utc时间
eta=datetime.utcnow() + timedelta(seconds=10) # 当前utc时间，往后推10s，时间对象
# # args是列表，send_sms的参数，eta是延迟时间，时间对象
res=send_sms.apply_async(args=['1888888',], eta=eta)
print(res)
4.3定时任务
# 在celery.py中配置
# 修改时区配置
# 时区
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False
# 任务的定时配置
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule =
# 定时任务一，每隔3秒做一次
'task-mul':
'task': 'celery_task.user_task.mul',
'schedule': timedelta(seconds=3), # 3s后
# 'schedule': crontab(hour=8, day_of_week=1), # 每周一早八点
'args': (3, 15),
,
# 定时任务二，每隔10秒做一次
'task-add':
'task': 'celery_task.home_task.add',
'schedule': timedelta(seconds=10), # 10s后
# 'schedule': crontab(hour=8, day_of_week=1), # 每周一早八点
'args': (3, 5),
,
# 启动beat（beat负责定时提交任务）
celery beat -A celery_task -l info
# 启动worker，任务就会被worker执行了
celery -A celery_task worker -l info -P eventlet
五、 django中使用celery
# django-celery模块，年久失修
-django-celery
-django
-celery
# 自己集成（跟框架无关）
六、 首页轮播图接口加入缓存
1 如果mysql数据变化了，轮播图表数据变化了，由于缓存没有更新，就会出现问题。
2 双写一致性问题
-因为使用了缓存，mysql和redis的数据不一致了。
-如何解决：（缓存更新策略）
-加入过期时间？ 60s？ 过期时间内数据还有问题。
-定时更新缓存（每隔60s，更新一下缓存）
-数据库一更改，立马改缓存。
-先改数据库，再改缓存。
-先改缓存，再改数据库。
-数据库一改，删除缓存。
-先删缓存，再改数据库。
-先改数据库，再删缓存（稍微好一些）。
3 首页轮播图接口：缓存更新策略是定时更新。
-celery的定时任务。
-使用celery的定时任务，完成首页轮播图缓存的更新。
七、Celery常见错误排查
celery消费任务不执行或者报错NotRegistered，与很多方面有关系。在排查错误时，要从以下6方面着手：
1) 整个项目目录结构：celery的目录结构和任务函数位置，有很大影响。
2) @task入参：用户有没有主动设置装饰器的入参 name，设置了和没设置有很大不同，建议主动设置这个名字，对函数名字和所处位置依赖减小。
3) celery的配置task_queues(在3.xx叫 CELERY_QUEUES )和task_routes (在3.xx叫 task_routes)。
4) celery的配置 include （在3.xx叫 CELERY_INCLUDE）或者 imports (3.xx CELERY_IMPORTS) 或者 app.autodiscover_tasks的入参。
5) cmd命令行启动参数 --queues= 的值。
6) 用户在启动cmd命令行时候，用户所在的文件夹。
在不规范的文件夹路径下，使用celery难度很高，一般教程都没教。
[项目文件夹目录格式不规范下的celery使用演示](https://github.com/ydf0509/celery_demo) 。
国产分布式函数调度框架 https://function-scheduling-distributed-framework.readthedocs.io/zh_CN/latest/index.html ，
从用法调用难度，用户所需代码量，超高并发性能，qps控频精确程度，支持的中间件类型，任务控制方式，稳定程度等19个方面，全方位超过celery，任何方面都是有过之而无不及。如果读者时间充裕，可以自行研究使用。
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：celery介绍架构基本使用，celery执行异步任务延迟任务定时任务，django中使用celery。的主要内容，如果未能解决你的问题，请参考以下文章