Celery 是一个简单、灵活且可靠的分布式消息处理系统，主要用来作为任务队列对海量消息数据进行实时的处理，在多个程序线程或者主机之间传递和分发工作任务。同时也支持计划任务等需求。

一、环境配置

Celery 框架自身并不对传入的消息进行存储，因此在使用前需要先安装第三方的 Message Broker。如 RabbitMQ 和 Redis 等。

安装 RabbitMQ

对于 Linux 系统，执行以下命令：

$ sudo apt-get install rabbitmq-server    # 安装 RabbitMQ
$ sudo rabbitmqctl add_user myuser mypassword    # 添加用户 myuser/mypassword
$ sudo rabbitmqctl add_vhost myvhost    # 添加 vhost
$ sudo rabbitmqctl set_user_tags myuser mytag
$ sudo rabbitmqctl set_permissions -p myvhost myuser ".*" ".*" ".*"    # 为用户 myuser 设置访问 myvhost 的权限

通过 Docker 安装的步骤如下：

$ docker pull rabbitmq:3.8-management    # 拉取 docker 镜像（包含 web 管理）
# 启动 rabbitmq 容器
$ docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 --hostname myRabbit \
-e RABBITMQ_DEFAULT_VHOST=myvhost \
-e RABBITMQ_DEFAULT_USER=myuser \
-e RABBITMQ_DEFAULT_PASS=mypassword rabbitmq:3.8-management

安装 Redis

$ sudo apt-get install redis-server

安装 Celery

$ pip install celery

二、创建 Celery 应用

Celery 应用是该框架所能提供的所有功能（如管理 tasks 和 workers 等）的入口，须确保它可以被其他模块导入。
以下是一段简单的 Celery app 代码 tasks.py：

# tasks.py
from celery import Celery

app = Celery('tasks',
             broker='pyamqp://myuser:mypassword@localhost:5672/myvhost',
             backend='redis://localhost:6379/0')

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

使用 RabbitMQ 作为 broker 接收和发送任务消息，使用 Redis 作为 backend 存储计算结果。

运行 Celery worker 服务

$ celery -A tasks worker --loglevel=info

$ celery -A tasks worker --loglevel=info

 -------------- celery@skitarniu-ubuntu18 v4.3.0 (rhubarb)
---- **** -----
--- * ***  * -- Linux-4.15.0-60-generic-x86_64-with-debian-buster-sid 2019-11-01 07:21:34
-- * - **** ---
- ** ---------- [config]
- ** ---------- .> app:         tasks:0x7f4f30b84a90
- ** ---------- .> transport:   amqp://myuser:**@localhost:5672/myvhost
- ** ---------- .> results:     redis://localhost:6379/0
- *** --- * --- .> concurrency: 2 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** -----
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery


[tasks]
  . tasks.add

[2019-11-01 07:21:35,316: INFO/MainProcess] Connected to amqp://myuser:**@127.0.0.1:5672/myvhost
[2019-11-01 07:21:35,367: INFO/MainProcess] mingle: searching for neighbors
[2019-11-01 07:21:36,535: INFO/MainProcess] mingle: all alone
[2019-11-01 07:21:36,782: INFO/MainProcess] celery@skitarniu-ubuntu18 ready.

任务测试

进入 Python Shell，执行以下命令发布任务并获取结果：

>>> from tasks import add
>>> result = add.delay(4, 4)
>>> result
<AsyncResult: 6f435bc7-f194-469c-837f-54d77f880ace>
>>> result.ready()
True
>>> result.get()
8
>>> result.traceback
>>>

delay() 方法用于发布任务消息，它是 apply_async() 方法的简写，即以异步的方式将任务需求提交给前面启动好的 worker 去处理。delay() 方法返回一个 AsyncResult 对象。
result.ready() 方法可以用来检查提交的任务是否已经完成，返回布尔值。

result.get() 方法则用于获取执行完成后的结果。如任务未完成，则程序会一直等待直到有结果返回。因此该方法是阻塞的，并不常用。可以传入 timeout 参数指定等待的时间上限。
如 result.get(timeout=1)，尝试获取任务执行后的结果，等待 1 秒。若 1 秒之后结果仍未返回，抛出 celery.exceptions.TimeoutError: The operation timed out. 异常。

如果任务执行过程中有抛出异常，则使用 get() 方法获取结果时会重新抛出该异常导致程序中断。可以通过修改 propagate 参数避免此情况：
result.get(propagate=False)
result.traceback 则用于获取任务的 traceback 信息。

三、Calling Tasks

Celery 定义了一些可供 task 实例调用的通用的 Calling API，包括三个方法和一些标准的执行选项：

• apply_async(args[, kwargs[, ...]])：发送任务消息给 worker
• delay(*args, **kwargs)：发送任务消息的简写形式，不支持执行选项
• calling (__call__)：即在本地进程中直接执行任务函数，不通过 worker 异步执行

以下是一些常见的调用示例：

• T.delay(arg, kwarg=value)
• T.apply_async((arg,), {'kwarg': value})
• T.apply_async(countdown=10)
  10 秒之后开始执行某个任务
• T.apply_async(eta=now + timedelta(seconds=10))
  10 秒之后开始执行某个任务
• T.apply_async(countdown=60, expires=120)
  预计 1 分钟后开始执行，但 2 分钟后还未执行则失效
• T.apply_async(expires=now + timedelta(days=2))
  2 天后失效

通过 countdown 设置任务的延迟执行：

>>> from tasks import add
>>> result = add.apply_async((2, 3))
>>> result.get()
5
>>> delay_result = add.apply_async((2, 3), countdown=15)
>>> delay_result.ready()
False
>>> delay_result.ready()
False
>>> delay_result.ready()
False
>>> delay_result.ready()
True
>>> delay_result.get()
5

还可以通过 eta（estimated time of arrival） 设置延迟执行的时间：

>>> from datetime import datetime, timedelta
>>> tomorrow = datetime.utcnow() + timedelta(days=1)
>>> add.apply_async((2, 3), eta=tomorrow)
<AsyncResult: c7dc6d7f-8b87-49d1-8077-73d7f046d709>

此时 worker 在命令行的日志输出如下：

[2019-11-06 05:16:21,362: INFO/MainProcess] Received task: tasks.add[c7dc6d7f-8b87-49d1-8077-73d7f046d709]
ETA:[2019-11-07 05:16:06.652736+00:00]

四、计划任务

Celery 允许像使用 crontab 那样按计划地定时执行某个任务。参考代码如下：

# tasks.py
from celery import Celery

app = Celery('tasks',
             broker='pyamqp://myuser:mypassword@localhost:5672/myvhost',
             backend='redis://localhost:6379/1')

app.conf.beat_schedule = {
    'add-every-60-seconds': {
        'task': 'tasks.add',
        'schedule': 60.0,
        'args': (16, 16)
    },
}
app.conf.timezone = 'UTC'

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

运行 celery -A tasks worker -B 启动 worker 服务。
-B 选项表示 beat，即 celery beat 服务，负责执行计划任务。

输出如下（每隔一分钟执行一次）：

$ celery -A tasks worker -B
...
[2019-11-06 05:41:34,057: WARNING/ForkPoolWorker-3] 32
[2019-11-06 05:42:33,998: WARNING/ForkPoolWorker-3] 32
[2019-11-06 05:43:34,056: WARNING/ForkPoolWorker-3] 32
[2019-11-06 05:44:34,105: WARNING/ForkPoolWorker-3] 32
[2019-11-06 05:45:34,157: WARNING/ForkPoolWorker-3] 32
...

同时 Celery 也支持更复杂的 crontab 类型的时间规划：

from celery.schedules import crontab

app.conf.beat_schedule = {
    # Executes every Monday morning at 7:30 a.m.
    'add-every-monday-morning': {
        'task': 'tasks.add',
        'schedule': crontab(hour=7, minute=30, day_of_week=1),
        'args': (16, 16),
    },
}

Crontab 表达式支持的语法如下：

参考资料

Celery 4.3.0 documentation