详解分布式任务队列Celery使用说明

起步

Celery 是一个简单、灵活且可靠的，处理大量消息的分布式系统，并且提供维护这样一个系统的必需工具。它是一个专注于实时处理的任务队列，同时也支持任务调度。

运行模式是生产者消费者模式：

任务队列：任务队列是一种在线程或机器间分发任务的机制。

消息队列：消息队列的输入是工作的一个单元，称为任务，独立的职程（Worker）进程持续监视队列中是否有需要处理的新任务。

Celery 用消息通信，通常使用中间人（Broker）在客户端和职程间斡旋。这个过程从客户端向队列添加消息开始，之后中间人把消息派送给职程，职程对消息进行处理。

Celery的架构由三部分组成，消息中间件（message broker），任务执行单元（worker）和任务执行结果存储（task result store）组成。

消息中间件：Celery本身不提供消息服务，但是可以方便的和第三方提供的消息中间件集成，包括，RabbitMQ, Redis, MongoDB等，本文使用 redis 。

任务执行单元：Worker是Celery提供的任务执行的单元，worker并发的运行在分布式的系统节点中

任务结果存储：Task result store用来存储Worker执行的任务的结果，Celery支持以不同方式存储任务的结果，包括Redis，MongoDB，Django ORM，AMQP等，这里我先不去看它是如何存储的，就先选用Redis来存储任务执行结果。

安装

通过 pip 命令即可安装：

pip install celery

本文使用 redis 做消息中间件，所以需要在安装：

pip install redis

redis软件也要安装，官网只提供了 linux 版本的下载:https://redis.io/download,windows 的可以到 https://github.com/MicrosoftArchive/redis 下载 exe 安装包。

简单的demo

为了运行一个简单的任务，从中说明 celery 的使用方式。在项目文件夹内创建 app.py 和 tasks.py 。tasks.py 用来定义任务：

# tasks.pyimport timefrom celery import Celerybroker = 'redis://127.0.0.1:6379/1'backend = 'redis://127.0.0.1:6379/2'app = Celery('my_tasks', broker=broker, backend=backend)@app.taskdef add(x, y):  print('enter task')  time.sleep(3)  return x + y

这些代码做了什么事。 broker 指定任务队列的消息中间件，backend 指定了任务执行结果的存储。app 就是我们创建的 Celery 对象。通过 app.task 修饰器将 add 函数变成一个一部的任务。

# app.pyfrom tasks import addif __name__ == '__main__':  print('start task')  result = add.delay(2, 18)  print('end task')  print(result)

add.delay 函数将任务序列化发送到消息中间件。终端执行 python app.py 可以看到输出一个任务的唯一识别：