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 True

user_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，任何方面都是有过之而无不及。如果读者时间充裕，可以自行研究使用。