【2022 年】Python3 爬虫教程 - 当爬虫遇见 RabbitMQ 消息队列

爬虫系列文章总目录：【2022 年】Python3 爬虫学习教程，本教程内容多数来自于《Python3网络爬虫开发实战（第二版）》一书，目前截止 2022 年，可以将爬虫基本技术进行系统讲解，同时将最新前沿爬虫技术如异步、JavaScript 逆向、AST、安卓逆向、Hook、智能解析、群控技术、WebAssembly、大规模分布式、Docker、Kubernetes 等，市面上目前就仅有《Python3 网络爬虫开发实战（第二版）》一书了，点击了解详情。

在数据爬取过程中，我们可能需要进行一些任务间通信机制的实现。比如说：

• 一个进程负责构造爬取请求，另一个进程负责执行爬取请求。
• 某个爬取任务进程完成了，通知另外一个进程进行数据处理。
• 某个进程新建了一个爬取任务，就通知另外一个进程开始数据爬取。

所以，为了降低这些进程的耦合度，就需要一个类似消息队列的中间件来存储和分发这些消息实现进程间的通信。

有了消息队列中间件之后，以上的两个任务就可以独立运行，通过消息队列来通信即可：

• 一个进程将需要爬取的任务构造请求对象放入队列，另一个进程从队列中取出请求对象并执行爬取。

• 某个爬取任务进程完成了，完成时就向消息队列发一个消息，另一个进程监听到这类消息，那就开始数据处理。

• 某个进程新建了一个爬取任务，那就向消息队列发一个消息，另一个负责爬取的进程监听到这类消息，那就开始数据爬取。

那这个消息队列用什么来实现呢？业界比较流行的有 RabbitMQ、RocketMQ、Kafka 等等，RabbitMQ 作为一个开源、可靠、灵活的消息队列中间件倍受青睐，本节我们就来了解下 RabbitMQ 的用法。

注意：前面我们了解了一些数据存储库的用法，基本都用于持久化存储一些数据。但本节介绍的是一个消息队列组件，虽然其主要应用于数据消息通信，但由于其也有存储信息的能力，所以将其归类于本章进行介绍。

1. RabbitMQ 介绍

RabbitMQ 是使用 Erlang 语言开发的开源消息队列系统，基于 AMQP 协议实现。AMQP 的全称是 Advanced Message Queue，即高级消息队列协议，它的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。

RabbitMQ 最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。具体特点包括：

• 可靠性（Reliability）：RabbitMQ 使用一些机制来保证可靠性，如持久化、传输确认、发布确认。
• 灵活的路由（Flexible Routing）：在消息进入队列之前，通过 Exchange 来路由消息的。对于典型的路由功能，RabbitMQ 已经提供了一些内置的 Exchange 来实现。针对更复杂的路由功能，可以将多个 Exchange 绑定在一起，也通过插件机制实现自己的 Exchange 。
• 消息集群（Clustering）：多个 RabbitMQ 服务器可以组成一个集群，形成一个逻辑 Broker 。
• 高可用（Highly Available Queues）：队列可以在集群中的机器上进行镜像，使得在部分节点出问题的情况下队列仍然可用。
• 多种协议支持（Multi-protocol）：RabbitMQ 支持多种消息队列协议，比如 STOMP、MQTT 等等。
• 多语言客户端（Many Clients）：RabbitMQ 几乎支持所有常用语言，比如 Java、.NET、Ruby 等等。
• 管理界面（Management UI）：RabbitMQ 提供了一个易用的用户界面，使得用户可以监控和管理消息 Broker 的许多方面。
• 跟踪机制（Tracing）：如果消息异常，RabbitMQ 提供了消息跟踪机制，使用者可以找出发生了什么。
• 插件机制（Plugin System）：RabbitMQ 提供了许多插件，来从多方面进行扩展，也可以编写自己的插件。

2. 准备工作

在本节开始之前，请确保已经正确安装好了 RabbitMQ，安装方式可以参考：https://setup.scrape.center/rabbitmq，确保其可以在本地正常运行。

除了安装 RabbitMQ 之外，我们还需要安装一个操作 RabbitMQ 的 Python 库，叫做 pika，使用 pip3 安装即可：

pip3 install pika

更详细的安装说明可以参考：https://setup.scrape.center/pika。

以上二者都安装好之后，我们就可以开始本节的学习了。

3. 基本使用

首先，RabbitMQ 提供的是队列的功能，我们要实现进程间通信，其本质上就是实现一个生产者-消费者模型，即一个进程作为生产者放入消息，另外一个进程作为消费者监听并处理消息，实现过程主要有 3 个关键点：

• 声明队列：通过指定队列的一些参数，将队列创建出来。
• 生产内容：生产者根据队列的连接信息连接队列，向队列中放入对应的内容。
• 消费内容：消费者根据队列的连接信息连接队列，从队列中取出对应的内容。

下面我们先来声明一个队列，相关代码如下：

import pika

QUEUE_NAME = 'scrape'
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue=QUEUE_NAME)

由于 RabbitMQ 运行在本地，所以这里直接使用 localhost 即可连接 RabbitMQ 服务，得到一个连接对象 connection。接下来我们需要声明一个频道，即 channel，利用它我们可以操作队列内容的生产和消费，接着我们调用 channel 方法的 queue_declare 来声明一个队列，队列名称叫作 scrape。

接着我们尝试向队列中添加一个内容：

channel.basic_publish(exchange='',
                      routing_key=QUEUE_NAME,
                      body='Hello World!')

这里我们调用了 channel 的 basic_publish 方法，向队列发布了一个内容，其中 routing_key 就是指队列的名称，body 就是真实的内容。

以上代码可以写入一个文件，取名为 producer.py，即生产者。

现在前两步——声明队列、生产内容其实已经完成了，接下来就是消费者从队列中获取内容了。

其实也很简单。消费者用同样的方式连接到队列，代码如下：

import pika

QUEUE_NAME = 'scrape'
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue=QUEUE_NAME)

然后从队列中获取数据，代码如下：

def callback(ch, method, properties, body):
    print(f"Get {body}")

channel.basic_consume(queue='scrape',
                      auto_ack=True,
                      on_message_callback=callback)
channel.start_consuming()

这里我们调用了 channel 的 basic_consume 进行消费，同时指定了回调方法 on_message_callback 的名称为 callback，另外还指定了 auto_ack 为 True，这代表消费者获取信息之后会自动通知消息队列，表示这个消息已经被处理了，当前消息可以从队列中移除。

最后将以上代码保存为 consumer.py 并运行，它会监听 scrape 这个队列的变动，如果有消息进入，就会获取并进行消费，回调 callback 方法，打印输出结果。

然后运行一下 producer.py，运行之后会连接刚才的队列，同时在该队列中加入一条消息，内容为 Hello World!。

这时候我们再返回 consumer，可以发现输出如下：

Get Hello World!

这就说明消费者成功收到了生产者发送的消息，消息成功被生产者放入消息队列，然后被消费者捕获并输出。

另外我们再次运行 producer.py，每运行一次，生产者都会向其中放入一个消息，消费者便会收到该消息并输出。

以上便是最基本的 RabbitMQ 的用法。

4. 随用随取

接下来我们来尝试实现一个简单的爬取队列，即一个进程负责构造爬取请求并将请求放入队列，另一个进程从队列中取出请求并执行爬取。

刚才我们仅仅是完成了基于 RabbitMQ 的最简单的生产者和消费者的通信，但是这种实现如果用在爬虫上是不太现实的，因为这里我们把消费者实现为了“订阅”的模式，也就是说，消费者会一直监听队列的变化，一旦队列中有了消息，消费者便要立马进行处理，消费者是无法主动控制它取用消息的时机的。

但实际上，假如我们要基于 RabbitMQ 实现一个爬虫的爬取队列的话，RabbitMQ 存的会是待执行/爬取的请求对象，生产者往里面放置请求对象，消费者获取到请求对象之后就执行这个请求，发起 HTTP 请求到服务器获取响应。但发起到获取响应的过程所消耗的时间是消费者无法控制的，这取决于服务器返回时间的长短。因此，消费者并不一定能够很快地将消息处理完，所以，消费者应该也有权利来控制取用的频率，这就是随用随取。

所以，这里我们可以稍微对前面的代码进行改写，生产者可以自行控制向消息队列中放入请求对象的频率，消费者也根据自己的处理能力控制自己从队列中取出请求对象的频率。如果生产者放置速度比消费者取用速度更快，那队列中就会缓存一些请求对象，反之队列则有时候会处于闲置状态。

但总的来说，消息队列充当了缓冲的作用，使得生产者和消费者可以按照自己的节奏来工作。

好，我们先实现下刚才所述的随用随取机制，队列中的消息暂且先用字符串来表示，后面我们可以再将其更换为请求对象。

这里我们可以将生产者实现如下：

import pika

QUEUE_NAME = 'scrape'
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.queue_declare(queue=QUEUE_NAME)

while True:
    data = input()
    channel.basic_publish(exchange='',
                          routing_key=QUEUE_NAME,
                          body=data)
    print(f'Put {data}')

这里生产者的数据我们还是使用 input 方法来获取，输入的内容就是字符串，输入之后该内容会直接被放置到队列中，然后打印输出到控制台。

先运行下生产者，然后回车输入几个内容：

foo
Put foo
bar
Put bar
baz
Put baz

这里我们输入了 foo、bar、baz 三个内容，然后控制台也有对应的输出结果。

然后消费者实现如下：

import pika

QUEUE_NAME = 'scrape'
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

while True:
    input()
    method_frame, header, body = channel.basic_get(
        queue=QUEUE_NAME, auto_ack=True)
    if body:
        print(f'Get {body}')

消费者也是一样的，我们这里也是可以通过 input 方法控制何时取用下一个，获取的方法就是 basic_get ，返回一个元组，其中 body 就是真正的数据。

运行消费者，回车几下，就可以看到每次回车都可以看到从消息队列中获取了一个新的数据：

Get b'foo'
Get b'bar'
Get b'baz'

这样我们就可以实现消费者的随用随取了。

5. 优先级队列

刚才我们仅仅是了解了最基本的队列的用法，RabbitMQ 还有一些高级功能。比如说，如果我们要想生产者发送的消息有优先级的区分，希望高优先级的队列被优先接收到，这个怎么实现呢？

其实很简单，我们只需要在声明队列的时候增加一个属性即可：

MAX_PRIORITY = 100

channel.queue_declare(queue=QUEUE_NAME, arguments={
    'x-max-priority': MAX_PRIORITY
})

这里在声明队列的时候，我们增加了一个参数，叫做 x-max-priority，指定一个最大优先级，这样整个队列就支持优先级了。

这里改写下生产者，在发送消息的时候指定一个 properties 参数为 BasicProperties 对象，BasicProperties 对象里面通过 priority 参数指定了对应消息的优先级，实现如下：

import pika

MAX_PRIORITY = 100
QUEUE_NAME = 'scrape'

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.queue_declare(queue=QUEUE_NAME, arguments={
    'x-max-priority': MAX_PRIORITY
})

while True:
    data, priority = input().split()
    channel.basic_publish(exchange='',
                          routing_key=QUEUE_NAME,
                          properties=pika.BasicProperties(
                              priority=int(priority),
                          ),
                          body=data)
    print(f'Put {data}')

这里优先级我们也可以手动输入，输入的内容我们需要分为两部分，用空格隔开，运行结果如下：

foo 40
Put foo
bar 20
Put bar
baz 50
Put baz

这里我们输入了三次内容，比如第一次输入的就是 foo 40，代表 foo 这个消息的优先级是 40，然后输入 bar 这个消息，优先级是 20，最后输入 baz 这个消息，优先级是 50。

然后重新运行消费者，按几次回车，可以看到如下输出结果：

Get b'baz'

Get b'foo'

Get b'bar'

这里我们可以看到结果就按照优先级取出来了，baz 这个优先级是最高的，所以就被最先取出来。bar 这个优先级是最低的，所以被最后取出来。

6. 队列持久化

除了设置优先级，我们还可以队列的持久化存储，如果不设置持久化存储，RabbitMQ 重启之后数据就没有了。

如果要开启持久化存储，可以在声明队列时指定 durable 为 True，实现如下：

channel.queue_declare(queue=QUEUE_NAME, arguments={
    'x-max-priority': MAX_PRIORITY
}, durable=True)

同时在添加消息的时候需要指定 BasicProperties 对象的 delivery_mode 为 2，实现如下：

properties=pika.BasicProperties(priority=int(priority), delivery_mode=2)

所以，这时候生产者的写法就改写如下：

import pika

MAX_PRIORITY = 100
QUEUE_NAME = 'scrape'

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.queue_declare(queue=QUEUE_NAME, arguments={
    'x-max-priority': MAX_PRIORITY
}, durable=True)

while True:
    data, priority = input().split()
    channel.basic_publish(exchange='',
                          routing_key=QUEUE_NAME,
                          properties=pika.BasicProperties(
                              priority=int(priority),
                              delivery_mode=2,
                          ),
                          body=data)
    print(f'Put {data}')

这样就可以实现队列的持久化存储了。

7. 实战

最后，我们将消息改写成前面所描述的请求对象，这里我们借助于 requests 库中的 Request 类来表示一个请求对象。

构造请求对象时，我们传入请求方法、请求 URL 即可，代码如下：

request = requests.Request('GET', url)

这样我们就构造了一个 GET 请求，然后可以通过 pickle 进行序列化然后发送到 RabbitMQ 中。

生产者实现如下：

import pika
import requests
import pickle

MAX_PRIORITY = 100
TOTAL = 100
QUEUE_NAME = 'scrape_queue'

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.queue_declare(queue=QUEUE_NAME, durable=True)

for i in range(1, TOTAL + 1):
    url = f'https://ssr1.scrape.center/detail/{i}'
    request = requests.Request('GET', url)
    channel.basic_publish(exchange='',
                          routing_key=QUEUE_NAME,
                          properties=pika.BasicProperties(
                              delivery_mode=2,
                          ),
                          body=pickle.dumps(request))
    print(f'Put request of {url}')

这里我们运行下生产者，就构造了 100 个请求对象并发送到 RabbitMQ 中了。

对于消费者来说，可以设置一个循环，一直不断地从队列中取出这些请求对象，取出一个就执行一次爬取，实现如下：

import pika
import pickle
import requests

MAX_PRIORITY = 100
QUEUE_NAME = 'scrape_queue'

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
session = requests.Session()

def scrape(request):
    try:
        response = session.send(request.prepare())
        print(f'success scraped {response.url}')
    except requests.RequestException:
        print(f'error occurred when scraping {request.url}')

while True:
    method_frame, header, body = channel.basic_get(
        queue=QUEUE_NAME, auto_ack=True)
    if body:
        request = pickle.loads(body)
        print(f'Get {request}')
        scrape(request)

这里消费者调用 basic_get 方法获取了消息，然后通过 pickle 反序列化还原成一个请求对象，然后使用 session 的 send 方法执行了该请求，进行了数据爬取，爬取成功就打印爬取成功的消息。

运行结果如下：

Get <Request [GET]>
success scraped https://ssr1.scrape.center/detail/1
Get <Request [GET]>
success scraped https://ssr1.scrape.center/detail/2
...
Get  <Request [GET]>
success scraped https://ssr1.scrape.center/detail/100

可以看到，消费者依次取出了爬取对象，然后成功完成了一个个爬取任务。

8. 总结

本节介绍了 RabbitMQ 的基本使用方法，有了它，爬虫任务间的消息通信就变得非常简单了，另外后文我们还会基于 RabbitMQ 实现分布式的爬取实战，所以本节的内容需要好好掌握。

本节代码：https://github.com/Python3WebSpider/RabbitMQTest。

本节部分内容参考来源：

• 文档 - RabbitMQ：https://www.rabbitmq.com/documentation.html
• 文档 - Pika：https://pika.readthedocs.io/