Celery第一步

Posted 2021-01-14 zydev

 

一、 Celery的架构图：

task producer - 任务生产者

       顾名思义就是发起调度任务的，然后交给任务队列去处理。简单的Python代码、耦合在Django/Flask Web 服务里请求任务比如调用备份或者调用初始化安装机器的任务，在程序里面调用Celery任务装饰的函数，产生任务并分发到任务队列处理的，我们都可以称之为任务生产者。

celery beat - 任务调度器 

       Celery beat 是 Celery 系统自带的任务生产者,它以独立进程的形式存在,该进程会读取配置文件的内容，周期性地将执行任务的请求发送给任务队列。需要注意的是在一个Celery系统中，只能存在一个 Celery beat 调度器。

broker - 任务代理 

       其实broker就是一个队列存储，是负责接收task producer发送的任务消息，存储到队列之后再进行调度，分发给任务消费方 (celery worker)。常见的broker有RabbitMQ、Redis 等。 

celery worker - 任务消费方

      Celery worker 就是任务的执行者，它负责接收任务处理中间方发来的任务处理请求，完成这些任务，并且返回任务处理的结果。Celery worker 对应的就是操作系统中的一个进程。Celery 支持分布式部署和横向扩展，我们可以在多个节点增加 Celery worker 的数量来增加系统的高可用性。在分布式系统中，我们也可以在不同节点上分配执行不同任务的 Celery worker 来达到模块化的目的。

Result Stores/backend

      存储Celery worker 执行任务之后的结果和状态信息，以供应用系统查询任务的状态信息。Celery 内置支持Django ORM，Redis，RabbitMQ 等方式来保存任务处理后的状态信息。

 

二、 安装配置Celery

• 选择RabbitMQ作为消息代理。
• RabbitMQ的Python客户端选择librabbitmq这个C库。
• 选择Msgpack做序列化。
• 选择Redis做结果存储。

安装： 

pip install "celery[librabbitmq,redi,msgpack]"

 

最简单的使用：

tasks.py

这里的backend也可以省略，不过这样就不会有结果保存了

from celery import Celery

app = Celery(‘tasks‘, broker=‘pyamqp://[email protected]//‘,backend=‘redis://localhost:6379/0‘)

@app.task
def add(x, y):
return x + y

运行Celery工作服务器：

celery -A tasks worker --loglevel=info

调动任务

>>> from tasks import add
>>> add.delay(4, 4)
>>> result.ready()
False
>>> result.get(timeout=1)
8