Python分布式爬虫如何持续从队列中不间断地取数据

帖子标题：Python分布式爬虫如何持续从队列中不间断地取数据

核心思路就是用一个死循环，在消费者节点里不断地从消息队列（比如Redis或RabbitMQ）里pop任务，处理完一个立马去取下一个。这里用Redis的blpop实现，它会阻塞直到有数据，这样既不会空转消耗CPU，又能保证不间断。

import redis
import json
import time
from your_spider_module import process_task  # 你的任务处理函数

# 连接Redis
redis_client = redis.Redis(host='localhost', port=6379, db=0)
TASK_QUEUE = 'distributed_spider:task_queue'

def worker():
    print(f"Worker started. Waiting for tasks from queue: {TASK_QUEUE}")
    while True:
        # blpop是关键：阻塞式左端弹出，超时时间设为0表示一直等
        # 返回一个元组 (queue_name, task_data)
        task_data = redis_client.blpop(TASK_QUEUE, timeout=0)
        
        if task_data:
            queue_name, task_json = task_data
            try:
                task = json.loads(task_json)
                print(f"Processing task: {task}")
                # 执行实际的爬取或解析逻辑
                process_task(task)
                print(f"Task completed: {task}")
            except json.JSONDecodeError:
                print(f"Failed to decode task: {task_json}")
            except Exception as e:
                print(f"Error processing task {task_json}: {e}")
                # 根据需求，这里可以选择将失败的任务重新放回队列或记录到死信队列

if __name__ == '__main__':
    worker()

要点说明：

1. blpop 是关键：用阻塞弹出代替轮询，没任务时线程挂起，有任务时立刻唤醒，高效。
2. 外层 while True：保证处理完一个任务后，立刻回到循环开头，等待下一个。
3. 异常处理：任务处理失败时记录日志，避免崩溃。根据你的策略，可以决定是否重试。
4. 扩展性：启动多个这样的worker进程/线程，就是分布式消费者。

生产者示例（往队列里放任务）：

import redis
import json

redis_client = redis.Redis(host='localhost', port=6379, db=0)
TASK_QUEUE = 'distributed_spider:task_queue'

def add_task_to_queue(task):
    task_json = json.dumps(task)
    redis_client.rpush(TASK_QUEUE, task_json)
    print(f"Task added: {task}")

总结：用带阻塞弹出的消息队列配合死循环。