Python爬虫实战：构建分布式爬虫系统（附完整架构与源码）

摘要

本文是\”Python爬虫实战系列\”的第五篇。在上一篇IP代理池的基础上，本文实现一个分布式爬虫系统，支持多机协作、任务调度、断点续爬和数据汇总。适用于大规模数据采集场景。

系列回顾：

• 第一篇：从0到1搭建电商价格监控系统
• 第二篇：多平台商品数据采集与价格对比分析
• 第三篇：自动化办公脚本大全（10个实用案例）
• 第四篇：构建IP代理池与智能调度系统

关键词：Python爬虫、分布式爬虫、Redis、任务队列、断点续爬

一、为什么需要分布式？

二、系统架构

┌─────────────┐
│ Master节点 │
│ 任务调度中心 │
└──────┬──────┘
│ Redis任务队列
┌────────────┼────────────┐
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ Worker 1 │ │ Worker 2 │ │ Worker N │
│ 采集+解析 │ │ 采集+解析 │ │ 采集+解析 │
└─────┬────┘ └─────┬────┘ └─────┬────┘
│ │ │
▼ ▼ ▼
┌─────────────────────────────────┐
│ 数据存储（MySQL/MongoDB） │
└─────────────────────────────────┘

三、核心代码

3.1 任务调度器（Master）

# master.py
import redis
import json
import time
from datetime import datetime

class TaskMaster:
\”\”\”任务调度器（Master节点）\”\”\”

def __init__(self, redis_host=\’localhost\’, redis_port=6379):
self.redis = redis.Redis(host=redis_host, port=redis_port, decode_responses=True)
self.task_queue = \’crawl_tasks\’
self.result_queue = \’crawl_results\’
self.status_key = \’crawl_status\’

def add_tasks(self, urls, task_config=None):
\”\”\”批量添加任务\”\”\”
task_config = task_config or {}

for i, url in enumerate(urls):
task = {
\’url\’: url,
\’task_id\’: f\”task_{int(time.time())}_{i}\”,
\’config\’: task_config,
\’status\’: \’pending\’,
\’created_at\’: datetime.now().strftime(\’%Y-%m-%d %H:%M:%S\’)
}

self.redis.rpush(self.task_queue, json.dumps(task, ensure_ascii=False))

print(f\”已添加 {len(urls)} 个任务到队列\”)

def get_result(self, timeout