Celery是如何工作的？
Celery 由于 其分布式体系结构，在某种程度上可能难以理解。下图是典型Django-Celery设置的高级示意图(FROM O’REILLY)：
当请求到达时，您可以在处理它时调用Celery任务。调用任务的函数会立即返回，而不会阻塞当前进程。实际上，任务尚未完成执行，但是任务消息已进入任务队列（或许多可能的任务队列之一）。
workers 是独立的进程，用于监视任务队列中是否有新任务并实际执行它们，他们拿起任务消息、处理任务、存储结果。
一、安装一个broker
Celery需要一个发送和接收消息的解决方案，即一个消息代理（message broker）服务，常用的broker包括：
RabbitMQ功能齐全，稳定，耐用且易于安装，是生产环境的绝佳选择。
Ubuntu安装：
$ sudo apt-get install rabbitmq-server
Docker安装：
$ docker run -d -p 5672:5672 rabbitmq
-
关于RabbitMQ在Celery中的高级配置，见：使用RabbitMQ
-
其他系统安装RabbitMQ，见：下载并安装RabbitMQ
Redis
Redis也具有完整的功能，但是在突然终止或电源故障的情况下更容易丢失数据。
Ubuntu安装：
$ sudo apt install redis-server
Docker安装：
$ docker run -d -p 6379:6379 redis
-
关于Redis在Celery中的高级配置：使用Redis
-
关于Redis的安装：How to Install and Secure Redis
二、安装Celery
$ pip install celery
三、编写Celery任务代码
首先导入Celery，创建一个Celery对象，这个对象将作为一个操作 Celery 的入口，如创建任务，管理workers等。
以下示例会把所有东西都写在一个模块中，但是对于大型项目，您需要创建一个专用模块。
# tasks.py
import time
from celery import Celery
app = Celery('tasks', broker='pyamqp://guest@localhost//')
@app.task
def add(x, y):
print('--------start---------')
for i in range(5):
print(f'第{i}秒')
print('--------over----------')
return x + y
第一个参数是当前模块的名称，这是唯一的必需参数。
第二个参数指定要使用的消息代理的URL。这里使用RabbitMQ（也是默认选项）。
若使用Redis：
app = Celery('tasks', broker='redis://localhost:6379/0')
四、启动 worker 进程
$ celery -A tasks worker --loglevel=INFO
在生产环境中，需要在后台将工作程序作为守护程序运行。为此，需要使用 平台提供的工具 或 类似supervisord的工具
五、调用任务
调用任务需要导入带有celery示例的模块，这里没有重新创建一个模块导入，而是使用命令行模式。要调用我们定义的任务，可以使用delay()（详情参阅 调用任务）：
>>> from tasks import add
>>> add.delay(4, 4)
调用任务将返回一个AsyncResult实例，这可用于检查任务的状态，等待任务完成或获取其返回值（或者如果任务失败，则获取异常和回溯）
默认情况下执行任务不返回结果。为了执行远程过程调用或跟踪数据库中的任务结果，需要配置result backend。
六、获取运行结果
如果要跟踪任务的状态，Celery需要将状态存储或发送到某个地方。有多个result backend可供选择：SQLAlchemy / Django ORM， MongoDB，Memcached，Redis，RPC（RabbitMQ / AMQP）等。
下面使用 RPC 作为result backend，该后端将状态作为瞬态消息发送回去。使用backend参数配置Celery对象的result backend：
app = Celery('tasks', backend='rpc://', broker='pyamqp://')
或者，如果使用 Redis 作为result backend，但仍然使用 RabbitMQ 作为 broker（流行的组合）：
app = Celery('tasks', backend='redis://localhost', broker='pyamqp://')
更多
result backend配置参阅“result backend。
我们再次调用该任务:
>>> result = add.delay(4, 4)
>>> result.ready() # 检查是否完成任务，返回布尔值
通过设置
@app.task(ignore_result=True)选项，也可以禁用单个任务返回结果,详情见有关celery.result的完整参考
七、配置Celery
对于大多数使用情况，默认配置就够了，但是可以配置更多选项使Celery根据需要工作。详细配置见“配置和默认值”。
可以直接在应用程序上设置配置，也可以使用专用的配置模块设置配置。例如配置用于序列化任务负载的默认序列化器：
# 配置一个设置：
app.conf.task_serializer = 'json'
# 一次配置许多设置，则可以使用update
app.conf.update(
task_serializer='json',
accept_content=['json'], # Ignore other content
result_serializer='json',
timezone='Europe/Oslo',
enable_utc=True,
)
对于较大的项目，建议使用专用的配置模块。
app.config_from_object('celeryconfig')
celeryconfig.py必须可用于从当前目录或Python路径中加载
celeryconfig.py
broker_url = 'pyamqp://'
result_backend = 'rpc://'
task_serializer = 'json'
result_serializer = 'json'
accept_content = ['json']
timezone = 'Europe/Oslo'
enable_utc = True
参考
- First Steps with Celery
- How Celery works
- How To Use Celery with RabbitMQ to Queue Tasks on an Ubuntu VPS
