Celery

Celery 简介

Celery是一个功能完备即插即用的任务队列。它使得我们不需要考虑复杂的问题，使用非常简单。celery看起来似乎很庞大，先对其进行简单的了解，然后再去学习其他一些高级特性。 celery适用异步处理问题，当发送邮件、或者文件上传, 图像处理等等一些比较耗时的操作，我们可将其异步执行，这样用户不需要等待很久，提高用户体验。

celery的特点是：

简单，易于使用和维护，有丰富的文档。

高效，单个celery进程每分钟可以处理数百万个任务。

灵活，celery中几乎每个部分都可以自定义扩展。

任务队列是一种跨线程、跨机器工作的一种机制.

任务队列中包含称作任务的工作单元。有专门的工作进程持续不断的监视任务队列，并从中获得新的任务并处理. celery通过消息进行通信，通常使用一个叫Broker(中间人)来协client(任务的发出者)和worker(任务的处理者). clients发出消息到队列中，broker将队列中的信息派发给worker来处理。一个celery系统可以包含很多的worker和broker，可增强横向扩展性和高可用性能。

官方文档 https://docs.celeryq.dev/en/stable/

celery 架构

celery架构分成三部分

消息中间件(broker) Celery本身不提供消息服务，但是可以方便的和第三方提供的消息中间件集成。包括， Redis等等

任务执行单元(workers) Worker是Celery提供的任务执行的单元，worker并发的运行在分布式的系统节点中(本质，一个work就是一个进程)。

任务结果存储（backend） Task result store用来存储Worker执行的任务的结果，Celery支持以不同方式存储任务的结果，包括 redis等

安装celery

pip install -U celery

也可以通过github上安装

使用

使用celery第一件事要做的最为重要的是就是要先创建一个Celery实例，我们一般叫做celery应用。或者更简单直接叫做app，app应用是我们使用celery功能的入口，比如创建任务，管理任务，在使用celery的时候， app必须能够被其他模块接入

一般celery的根目录如下


Celery_tasks/
├── config.py # 配置项
├── __init__.py
├── main.py # 主程序
└── sms	# 一个目录下可以放置多个任务，该目录下存放当前任务需要的模块和依赖
    ├── __init__.py
    └── tasks.py # 任务的文件 文件名必须是这个

1 directory, 5 files

首先 main.py


from celery import Celery
# 创建celery对象

app = Celery()

# 加载配置
app.config_from_object("Celery_tasks.config")


# 注册任务
app.autodiscover_tasks(['Celery_tasks.sms'])
# app.autodiscover_tasks(['Celery_tasks.sms','Celery_tasks_mail'])



# 通过终端 启动celery
# celery -A Celery_tasks.main worker --loglevel=info

config.py


# 设置任务队列的地址
broker_url = 'redis://127.0.0.1:6379/14'  # 消息中间件
# 设置消息队列的地质
result_backend = 'redis://127.0.0.1:6379/15'  # 结果存储

sms/tasks.py


from Celery_tasks.main import app


@app.task(name="send_sms")
def send_message():
    """发送短信"""

    return "hello,world"

启动celery


celery -A Celery_tasks.main worker --loglevel=info

启动成功

上述的 [tasks] . send_sms 是指 sms文件下的tasks 指定的 @app.task(name="send_sms")

多个tasks

可以在 send_mail中再加个task，只需要加入如下代码

mail/tasks.py


from Celery_tasks.main import app


@app.task(name="send_main")
def send_mail():
    """发送邮件"""
    return "hello,mail!!!!"

同时main.py中也要完成对tasks的注册


app.autodiscover_tasks(['Celery_tasks.sms','Celery_tasks_mail'])

调用celery


from Celery_tasks.sms.tasks import send_sms

send_sms.delay()

调用成功

redis数据库中

如果当你关闭celery 当你重新开启celery时任务将不断轮询从而执行一系列操作

Django Celery流程图