Celery第一步

时间：2021-08-26 12:17:23

一、Celery是什么？

　　Celery是一个简单，灵活且可靠的分布式系统，可以处理大量消息，同时为操作提供维护该系统所需的工具。

　　这是一个任务队列，着重于实时处理，同时还支持任务调度。

二、 Celery的架构图：

Celery第一步

  • 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 内置支持AMQP, Redis等方式来保存任务处理后的状态信息。

三、 安装Celery

为了提高性能，我们选择如下方案：

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

安装： 

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

四、 第一个Celery程序

首先要服务器上已经安装了RabbitMQ、Redis

CentOS上安装配置RabbitMQ Server

Redis的安装配置

4.1. 编写tasks.py
from celery import Celery

app = Celery('tasks', broker='pyamqp://celery:celery@192.168.0.12:5672/celery_vhost',backend='redis://localhost:6379/0')

app.conf.update(

    task_serializer='json',

    accept_content=['json'],  # Ignore other content

    result_serializer='json',

    timezone='Asia/Shanghai',

    enable_utc=True,

)

@app.task

def add(x, y):

    return x + y

　第一个参数tasks是当前模块的名称。这是必需的，以便在__main__模块中定义任务时可以自动生成名称。

　第二个参数broker是消息代理，格式如下：

　　broker_url = 'amqp://myuser:mypassword@localhost:5672/myvhost'

　第三个参数backend是存储任务的执行结果和状态的，格式如下：

　　redis://:password@hostname:port/db_number

如果redis和RabbitMQ和程序在同一个机器上，并且都是默认配置，可以直接写为：

app = Celery('tasks', backend='redis://localhost', broker='pyamqp://')
4.2 运行 celery work服务器：
celery -A tasks worker --loglevel=info
celery -A tasks worker --loglevel=info

 -------------- celery@node2 v4.3.0 (rhubarb)

---- **** -----

--- * ***  * -- Linux-3.10.-957.1..el7.x86_64-x86_64-with-centos-7.6.-Core -- ::

-- * - **** ---

- ** ---------- [config]

- ** ---------- .> app:         tasks:0x7fb49c8330b8

- ** ---------- .> transport:   amqp://celery:**@192.168.0.12:5672/celery_vhost

- ** ---------- .> results:     redis://localhost:6379/0

- *** --- * --- .> concurrency:  (prefork)

-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)

--- ***** -----

 -------------- [queues]

                .> celery           exchange=celery(direct) key=celery

[tasks]

  . tasks.add

　要停止worker，只需打Control-C

4.3 后台启动和关闭Celery

默认情况下，它将在当前目录中创建pid和日志文件，以防止多个工作程序相互启动，建议将它们放在专用目录中

$ mkdir -p /var/run/celery

$ mkdir -p /var/log/celery

$ celery multi start w1 -A tasks -l info --pidfile=/var/run/celery/%n.pid \

                                        --logfile=/var/log/celery/%n%I.log

停止worker

celery multi stop w1 -A proj -l info

该stop命令是异步的，因此它不会等待工作程序关闭。您可能需要改用该stopwait命令，以确保在退出之前完成所有当前正在执行的任务：
 celery multi stopwait w1 -A proj -l info
4.4 调用任务

在另一个终端打开ipython

In [4]: result = add.delay(4, 4)

In [5]: result.task_id
Out[5]: 'd8d98d41-bb64-4c3a-9c0b-0deab6809597'

　　调用任务将返回一个AsyncResult实例。这可用于检查任务的状态，等待任务完成或获取其返回值（或者，如果任务失败，则获取异常和回溯）

在启动celery的终端可以看见已经产生了一个任务，并且返回了执行结果

[2019-10-23 17:02:43,344: INFO/MainProcess] Received task: tasks.add[d8d98d41-bb64-4c3a-9c0b-0deab6809597]
[2019-10-23 17:02:43,364: INFO/ForkPoolWorker-2] Task tasks.add[d8d98d41-bb64-4c3a-9c0b-0deab6809597] succeeded in 0.017695230431854725s: 8

我们也可以在Redis中查看保存的结果信息

127.0.0.1:6379> get celery-task-meta-d8d98d41-bb64-4c3a-9c0b-0deab6809597

"{\"status\": \"SUCCESS\", \"result\": 8, \"traceback\": null, \"children\": [], \"task_id\": \"d8d98d41-bb64-4c3a-9c0b-0deab6809597\", \"date_done\": \"2019-10-23T09:02:43.347416\"}"

　　

相关文章