Celery 是一个分布式队列的管理工具,它封装了操作常见任务队列的各种操作，可以用 Celery 提供的接口快速实现并管理一个分布式的任务队列 , 我们使用它可以快速进行任务队列的使用与管理,同时也支持任务调度。

使用原因

程序运行过程中，要执行耗时的任务，但又不想主程序被阻塞，常见的方法是多线程。可是当并发量过大时，多线程压力很大，必须要用线程池来限制并发个数，而且多线程对共享资源的使用也是很麻烦的事情。当然也可以使用协程，但是协程毕竟还是在同一线程内执行的，如果一个任务本身就要执行很长时间，而不是因为等待IO被挂起，那其他协程照样无法得到运行。

常见使用场景

开发中经常碰到需要使用异步任务的场景，比如一个WEB请求中有运行时间很长的业务运算，如果不采用异步任务，会阻塞当前的web请求，影响用户体验。我们有多种不同的任务，这些任务优先级不同，比如我们有视频上传和压缩任务等，照片压缩上传等任务还有其他不重要的任务。这些任务耗时需要使用不同的worker去处理。，等到Celery执行完成以后通知用户。在运维场景下需要批量在几百台机器执行某些命令或者任务，此时Celery可以轻松搞定。

Celery的架构组成

一个完整的Celery分布式队列架构应该包含一下几个模块：

Celery由以下三部分构成：任务发布者，任务调度器，消息中间件(Broker)、任务执行单元(Worker)、任务结果存储(Backend)，如图：

producer: 任务发布者, 通过调用API向celery发布任务的程序。

celery beat:任务调度器。beat进程会读取配置文件里的内容。(celerybeat_schedule里设置)，周期性的将配置中到期需要执行的任务发送到任务队列。 根据配置文件发布定时任务。

Broker，即为任务调度队列，接收任务生产者发来的消息(即任务)，将任务存入队列。接受任务消息，存入队列再按顺序分发给worker执行。Celery 本身不提供队列服务，官方推荐使用 RabbitMQ 和 Redis 等。

Worker 是执行任务的处理单元，实际执行任务的程序，它实时监控消息队列，获取队列中调度的任务，并执行它。

Backend 用来存储Worker执行的任务的结果，Celery存储默认支持包括AMQP, Redis，memcached, MongoDB，Django ORM，SQLAlchemy,等。

安装Celery

可以通过Python包索引(PyPI)或从源安装Celery

要使用pip安装：

pip install Celery

要使用easy_install进行安装：

easy_install Celery

Celery配置

常见的配置有三种方式，以redis为例BROKER_URL = 'redis://ip(地址):6379/0'。celery broker 的地址和选择的 redis db，默认是 0。

第一种：

第二种：

第三种：

config中指定broker_ur和result_backend

app = Celery('tasks')

app.config_from_object('config')

添加异步任务(task)

使用@app.task() 添加task

测试代码：

@app.task()

def add_test(x, y):

return x + y

Celery 启动

测试代码：

if __name__ == "__main__":

app.start()

或在shell中执行：

celery worker -A task -l info

参数：-A 选项指定 celery 实例 app 的位置， 可以直接指定：

celery worker -A task.app -l info

-l 选项指定日志级别， -l 是 --loglevel 的缩略形式。

Celery监控

celery flower是python的用于监控celery的模块。比如可以在web页面实时显示celery worker的状态、任务的状态等，flower还提供了REST API用于方便第三方应用直接触发任务执行、获取任务执行结果等，这也极大地扩展了celery的应用范围，比如一个Java应用也可以通过flower的REST接口使用celery的异步任务系统。

Celery集成flask

创建celery的实例对象的名字必须是flask应用程序app的名字,否则celery启动会失败。

安装Flask-Celery-Helper

pip install Flask-Celery-Helper

测试例子：

Celery集成djongo

安装依赖库

pip install django

pip install django-celery

pip install celery-with-redis

测试例子：