爬虫架构|Celery+RabbitMQ快速入门（四）整合版本

前面用三篇文章断断续续写了Celery+RabbitMQ相关的文章。

爬虫架构|Celery+RabbitMQ快速入门（一）用工作任务分配的案例介绍了它们是如何配合工作的，如下图4-1所示：

图4-1

爬虫架构|Celery+RabbitMQ快速入门（二）讲了它们在项目中的简单使用流程，如下图4-2所示：

图4-2

1. RabbitMQ所在服务器，启动crontab设置 crontable -user user -e设置定时执行celery application应用；
2. 在task.py文件里面启动一个叫做app的Celery Application，编写一个app.task函数来produce任务到RabbitMQ；
3. 在每个worker里面通过命令启动worker消费任务；

爬虫架构|Celery+RabbitMQ快速入门（三）讲解了一个分布式爬虫需要解决的两个基本问题：不重复地分配爬取任务和将所有爬虫的结果汇总到一处。同时也提到Celery由5个主要组件组成（1、3、4都已经提到也已使用）：

1. producer: 任务发布者, 通过调用API向celery发布任务的程序
2. celery beat: 任务调度, 根据配置文件发布定时任务
3. worker: 实际执行任务的程序
4. broker: 接受任务消息，存入队列再按顺序分发给worker执行
5. backend: 存储结果的服务器了

接下来整合前面三篇文章的内容，做一个整合版本。

一、Celery简介

Celery是一个专注于实时处理和任务调度的分布式任务队列。所谓任务就是消息，消息中的有效载荷中包含要执行任务需要的全部数据。

使用Celery的常见场景如下：

1. Web应用。当用户触发的一个操作需要较长时间才能执行完成时，可以把它作为任务交给Celery去异步执行，执行完再返回给用户。这段时间用户不需要等待，提高了网站的整体吞吐量和响应时间。
2. 定时任务。生产环境经常会跑一些定时任务。假如你有上千台的服务器、上千种任务，定时任务的管理很困难，Celery可以帮助我们快速在不同的机器设定不同种任务。
3. 同步完成的附加工作都可以异步完成。比如发送短信/邮件、推送消息、清理/设置缓存等。

Celery还提供了如下的特性：

1. 方便地查看定时任务的执行情况，比如执行是否成功、当前状态、执行任务花费的时间等。
2. 可以使用功能齐备的管理后台或者命令行添加、更新、删除任务。
3. 方便把任务和配置管理相关联。
4. 可选多进程、Eventlet和Gevent三种模式并发执行。
5. 提供错误处理机制。 1）提供多种任务原语，方便实现任务分组、拆分和调用链。 2）支持多种消息代理和存储后端。

二、Celery架构

Celery包含如下组件：

1. Producer：调用了Celery提供的API、函数或者装饰器而产生任务并交给任务队列处理的都是任务生产者。
2. Celery Beat：任务调度器，Beat进程会读取配置文件的内容，周期性地将配置中到期需要执行的任务发送给任务队列。
3. Celery Worker：执行任务的消费者，通常会在多台服务器运行多个消费者来提高执行效率。
4. Broker：消息代理，或者叫作消息中间件，接受任务生产者发送过来的任务消息，存进队列再按序分发给任务消费方（通常是消息队列或者数据库）。
5. Result Backend：任务处理完后保存状态信息和结果，以供查询。Celery默认已支持Redis、RabbitMQ、MongoDB、Django ORM、SQLAlchemy等方式。

Celery的架构图如下图4-3所示：

图4-3

1. 任务发布者有两种产生任务的方式：发布者发布任务（Web应用）和任务调度按期发布任务（定时任务）。
2. 消息代理会把接受到的任务信息分发给任务消费方，我们项目实战中消息代理使用的是RabbitMQ。
3. 消费者消费任务，在多台服务器运行多个消费者来提高执行效率。
4. 存储结果到数据库。

三、选择消息代理（Broker）

Celery目前支持RabbitMQ、Redis、MongoDB、Beanstalk、SQLAlchemy、Zookeeper等作为消息代理，但适用于生产环境的只有RabbitMQ和Redis，至于其他的方式，一是支持有限，二是可能得不到更好的技术支持。

Celery官方推荐的是RabbitMQ，Celery的作者Ask Solem Hoel最初在VMware就是为RabbitMQ工作的，Celery最初的设计就是基于RabbitMQ，所以使用RabbitMQ会非常稳定，成功案例很多。如果使用Redis，则需要能接受发生突然断电之类的问题造成Redis突然终止后的数据丢失等后果。

四、Celery序列化

在客户端和消费者之间传输数据需要序列化和反序列化，Celery支持如下的序列化方案：

• pickle pickle是Python标准库中的一个模块，支持Python内置的数据结构，但是它是Python的专有协议。从Celery3.2开始，由于安全性等原因Celery将拒绝pickle这个方案。
• json json支持多种语言，可用于跨语言方案。
• yaml yaml的表达能力更强，支持的数据类型比json多，但是Python客户端的性能不如JSON。
• msgpack msgpack是一个二进制的类JSON的序列化方案，但是比JSON的数据结构更小、更快。

五、一个例子

我们的例子选择如下方案：

1. 选择RabbitMQ作为消息代理。
2. RabbitMQ的Python客户端选择librabbitmq这个C库。
3. 选择JSON做序列化，应用跨语言。
4. 选择Redis做结果存储。

一个简单的Celery项目主要包括如下目录：

├── celeryconfig.py
├── celery.py
└── tasks.py

主程序celery.py：

from __future__ import absolute_import
from celery import Celery
app = Celery('proj', include=['proj.tasks'])
app.config_from_object('proj.celeryconfig')
if __name__ == '__main__':
    app.start()

1. "from future import absolute_import"是拒绝隐式引入，因为celery.py的名字和celery的包名冲突，需要使用这条语句让程序正确地运行。
2. app是Celery类的实例，创建的时候添加了proj.tasks这个模块，也就是包含了proj/tasks.py这个文件。
3. 把Celery配置存放进proj/celeryconfig.py文件，使用app.config_from_object加载配置。

存放任务函数的文件tasks.py：

from __future__ import absolute_import
from proj.celery import app
@app.task
def add(x, y):
    return x + y

tasks.py只有一个任务函数add，让它生效的最直接的方法就是添加app.task这个装饰器。

配置文件celeryconfig.py：

# 使用RabbitMQ作为消息代理
BROKER_URL='amqp://spider:*****@IP:端口/****' 
# 把任务结果存在了Redis
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0' 
# 任务序列化和反序列化使用JSON方案
CELERY_TASK_SERIALIZER = 'json' 
# 读取任务结果使用JSON
CELERY_RESULT_SERIALIZER = 'json' 
# 任务过期时间，不建议直接写86400，应该让这样的magic数字表述更明显
CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24 
# 指定接受的内容类型，是个数组，可以写多个
CELERY_ACCEPT_CONTENT = ['json'] 

启动消费者：

celery -A proj worker -l info

-A参数默认会寻找proj.celery这个模块，其实使用celery作为模块文件名字不怎么合理。可以使用其他名字。举个例子，假如是proj/app.py，可以使用如下命令启动：

celery -A proj.app worker -l info

上述信息提供了一些有帮助的内容，如消息代理和存储结果的地址、并发数量、任务列表、交换类型等。在对Celery不熟悉的时候可以通过如上信息判断设置和修改是否已生效。

六、指定队列

Celery非常容易设置和运行，通常它会使用默认的名为celery的队列（可以通过CELERY_DEFAULT_QUEUE修改）用来存放任务。我们可以使用优先级不同的队列来确保高优先级的任务不需要等待就得到响应。

基于proj目录下的源码，我们创建一个projq目录，并对projq/celeryconfig.py添加如下配置：

from kombu import Queue
CELERY_QUEUES = ( # 定义任务队列
    Queue('default', routing_key='task.#'), # 路由键以“task.”开头的消息都进default队列
    Queue('web_tasks', routing_key='web.#'), # 路由键以“web.”开头的消息都进web_tasks队列
)

CELERY_DEFAULT_EXCHANGE = 'tasks' # 默认的交换机名字为tasks
CELERY_DEFAULT_EXCHANGE_TYPE = 'topic' # 默认的交换类型是topic
CELERY_DEFAULT_ROUTING_KEY = 'task.default' # 默认的路由键是task.default，这个路由键符合上面的default队列

CELERY_ROUTES = {
    'projq.tasks.add': { # tasks.add的消息会进入web_tasks队列
    'queue': 'web_tasks',
    'routing_key': 'web.add',
    }
}

现在用指定队列的方式启动消费者进程：

celery -A projq worker -Q web_tasks -l info

上述worker只会执行web_tasks中的任务，我们可以合理安排消费者数量，让web_tasks中任务的优先级更高。

七、使用任务调度

之前的例子都是由发布者触发的，本节展示一下使用Celery的Beat进程自动生成任务。基于proj目录下的源码，创建一个projb目录，对projb/celeryconfig.py添加如下配置：

CELERYBEAT_SCHEDULE = {
    'add': {
        'task': 'projb.tasks.add',
        'schedule': timedelta(seconds=10),
        'args': (16, 16)
    }
}

CELERYBEAT_SCHEDULE中指定了tasks.add这个任务每10秒跑一次，执行的时候的参数是16和16。

启动Beat程序：

celery beat -A projb

然后启动Worker进程：

celery -A projb worker -l info

之后可以看到每10秒都会自动执行一次tasks.add。

注：Beat和Worker进程可以一并启动：

celery -B -A projb worker -l info

使用Django可以通过django-celery实现在管理后台创建、删除、更新任务，是因为它使用了自定义的调度类djcelery.schedulers.DatabaseScheduler，我们可以参考它实现Flask或者其他Web框架的管理后台来完成同样的功能。使用自定义调度类还可以实现动态添加任务。