Python 爬虫进阶：携手 RabbitMQ，解锁任务通信

后端

2024-01-08 10:04:23

在广袤无垠的网络世界中，数据爬取是一项必不可少的技能。它就好比一艘探索船，满载着对信息的求知欲，在浩瀚的网络海洋中扬帆启航。为了让这艘船驶得更远、更稳，我们需要借助各种工具和技术。其中，消息队列就是一盏明灯，指引着爬虫穿透迷雾，抵达彼岸。

今天，我们就来谈谈 RabbitMQ，一个功能强大的消息队列系统，它将为我们的 Python 爬虫注入新的活力，让任务通信变得轻而易举。

任务通信的必要性

在数据爬取过程中，我们经常会遇到需要多个任务协同工作的情况。比如，一个任务负责构建爬取请求，另一个任务负责执行爬取请求。此时，我们就需要一种机制来协调这两个任务之间的通信。

传统上，我们可能会使用共享内存或数据库来实现任务通信。然而，这些方法存在一些缺陷，比如性能低下、可靠性差等。而消息队列则可以完美地解决这些问题。

RabbitMQ 简介

RabbitMQ 是一个开源的消息队列系统，以其高性能、高可靠性、高可扩展性而闻名。它遵循 AMQP（高级消息队列协议）标准，支持多种编程语言，包括 Python。

在 RabbitMQ 中，消息被组织成队列，每个队列都有一个唯一的名称。生产者（发送消息的任务）将消息发送到队列，消费者（接收消息的任务）从队列中接收消息。

Python 爬虫与 RabbitMQ 集成

现在，让我们了解一下如何在 Python 爬虫中集成 RabbitMQ。首先，我们需要安装 pika 库，这是一个 Python 客户端，用于与 RabbitMQ 交互。

安装完成后，我们就可以创建 RabbitMQ 客户端，并连接到消息队列服务器。接下来，我们可以创建生产者和消费者，并将它们与队列关联。

import pika

# 创建 RabbitMQ 客户端
client = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

# 创建信道
channel = client.channel()

# 创建队列
channel.queue_declare(queue='my_queue')

# 创建生产者
producer = channel.basic_publish(exchange='', routing_key='my_queue', body='Hello, RabbitMQ!')

# 创建消费者
consumer = channel.basic_consume(queue='my_queue', on_message_callback=callback)

# 启动消费者
channel.start_consuming()

在上面这段代码中，我们创建了一个生产者，向 "my_queue" 队列发送了一条消息 "Hello, RabbitMQ!"。我们还创建了一个消费者，每当队列中收到消息时，就会调用 callback 函数。

实战：爬取豆瓣电影数据

现在，我们来实战一下，使用 RabbitMQ 来协调豆瓣电影数据爬取任务。

我们将创建一个生产者任务，负责从豆瓣电影网站抓取电影数据，并将其发送到 RabbitMQ 队列中。然后，我们将创建一个消费者任务，从队列中接收电影数据，并将其存储到数据库中。

# 生产者任务
import requests
import pika

client = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = client.channel()
channel.queue_declare(queue='movie_queue')

for i in range(1, 10):
    url = 'https://movie.douban.com/top250?start={}'.format(i * 25)
    response = requests.get(url)
    movies = response.json()['subjects']
    for movie in movies:
        data = json.dumps(movie)
        channel.basic_publish(exchange='', routing_key='movie_queue', body=data)

channel.close()
client.close()


# 消费者任务
import pika
import json
import pymysql

client = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = client.channel()
channel.queue_declare(queue='movie_queue')

def callback(ch, method, properties, body):
    movie = json.loads(body.decode('utf-8'))
    # 将电影数据存储到数据库中
    ...

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