爬虫架构|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框架的管理后台来完成同样的功能。使用自定义调度类还可以实现动态添加任务。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏信安之路

简单粗暴的文件上传漏洞

文件上传漏洞可以说是日常渗透测试用得最多的一个漏洞,因为用它获得服务器权限最快最直接。但是想真正把这个漏洞利用好却不那么容易,其中有很多技巧,也有很多需要掌握的...

1.2K00
来自专栏王磊的博客

Twitter Storm安装配置(Ubuntu系统)单机版

要使用storm首先要安装以下工具:JDK、Python、zookeeper、zeromq、jzmq、storm (注:各个模块都是独立的,如果安装失败或者卡顿...

32550
来自专栏沈唁志

创建属于自己的第一个Composer/Packagist包

Composer 是 PHP 的一个依赖管理工具,Composer 不是一个包管理器,它涉及 “packages” 和 “libraries”,但它在每个项目的...

15440
来自专栏张善友的专栏

Jexus 5.8.2 正式发布为Asp.Net Core进入生产环境提供平台支持

Jexus 是一款运行于 Linux 平台,以支持  ASP.NET、PHP 为特色的集高安全性和高性能为一体的 WEB 服务器和反向代理服务器。最新版 5.8...

22760
来自专栏我的博客

Composer

Composer是PHP中用来管理依赖(dependency)关系的工具。你可以在自己的项目中声明所依赖的外部工具库(libraries),Composer会帮...

39870
来自专栏飞雪无情的博客

Go语言经典库使用分析(三)| Gorilla Handlers 详细介绍

在我们编写web服务端程序的时候,我们可能会对一些甚至全部的Http Request统一处理,比如我们记录每个访问的Request,对提交的Form表单进行映射...

13110
来自专栏分布式系统进阶

Kafka重置消费的OffsetKafka源码分析-汇总

56620
来自专栏技术博文

H5缓存机制浅析

1 H5 缓存机制介绍 H5,即 HTML5,是新一代的 HTML 标准,加入很多新的特性。离线存储(也可称为缓存机制)是其中一个非常重要的特性。H5 引入的离...

45080
来自专栏python学习指南

python爬虫(六)_urllib2:handle处理器和自定义opener

本文将介绍handler处理器和自定义opener,更多内容请参考:python学习指南 opener和handleer 我们之前一直使用的是urllib...

30380
来自专栏Script Boy (CN-SIMO)

Qt入门之基础篇 ( 二 ) :Qt项目建立、编译、运行和发布过程解析

转载请注明出处:CN_Simo。 题解:   本篇内容主讲Qt应用从创建到发布的整个过程,旨在帮助读者能够快速走进Qt的世界。 本来计划是讲解Qt源码静态编译...

56510

扫码关注云+社区

领取腾讯云代金券