一、 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 eventletcelery worker -A celery_task -l info -P eventlet## 5.x版本pip install eventletcelery -A celery_task worker -l info -P eventlet# linux 中启动 workcelery worker -A celery_task -l infocelery -A celery_task worker -l info# 启动 beat 的命令（负责每隔几秒钟，向任务队列中提交任务）celery beat -A celery_task -l info

两种celery任务结构：一种是放在模块下，另一种是放在包下。提倡用包管理，结构更清晰。

如果 Celery对象:Celery(...) 放在一个模块下，则：1）终端切换到该模块所在文件夹位置：scripts2）执行启动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 Celerybackend='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.taskdef add(a, b):# 很耗时的任务import timetime.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-5abb963dc1d3print(res)

新建get_result.py查询任务

from celery_task import app# 自己写的appfrom 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的数据:7print(result)elif a.failed():print('任务失败')elif a.status == 'PENDING':print('任务等待中，尚未被执行')elif a.status == 'RETRY':print('任务异常后正在重试')elif a.status == 'STARTED':print('任务已经开始被执行')

启动worker

# windows中启动workpip install eventletcelery -A celery_task worker -l info -P eventlet# linux中启动workcelery -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.pyadd_task.py# 别的服务提交任务get_result.py # 别的服务获取结果

celery.py 文件

### 有app的这个py文件，必须叫celeryfrom celery import Celerybackend = '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'# 是否使用UTCapp.conf.enable_utc = False# 任务的定时配置from datetime import timedeltafrom celery.schedules import crontabapp.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 True

user_task.py

from .celery import app@app.task# user模块的任务, 发送短信def send_sms(s):import timetime.sleep(3)return '%s手机号，短信发送成功'%s

四、 celery执行异步任务、延迟任务、定时任务

4.1 异步任务

from celery_task.user_task import send_smsres=send_sms.delay('18953675221')print(res)

4.2 延迟任务

# 延迟任务，延迟5s钟后，发送短信from celery_task.user_task import send_smsfrom 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'# 是否使用UTCapp.conf.enable_utc = False​# 任务的定时配置from datetime import timedeltafrom celery.schedules import crontabapp.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，任何方面都是有过之而无不及。如果读者时间充裕，可以自行研究使用。