Python 爬虫实战：综合案例 - 企业级异步分布式爬虫系统搭建

前言

在前述系列文章中，我们分别掌握了 Scrapy 分布式爬虫、中间件开发、Pipeline 数据持久化、aiohttp 异步爬虫及 asyncio 任务调度的核心技术。在企业级爬虫开发场景中，单一技术栈往往无法满足 “高并发、高可用、可扩展、易维护” 的需求，需整合多技术栈构建完整的爬虫系统。本文将以 电商商品评论采集 为实战场景，整合 Scrapy、aiohttp、asyncio、Redis、MySQL、MongoDB 等技术，搭建一套企业级异步分布式爬虫系统，覆盖任务调度、反爬对抗、数据清洗、持久化存储、监控告警全流程，解决企业级爬虫开发的综合难题。

摘要

本文聚焦企业级异步分布式爬虫系统的全流程搭建，首先剖析系统架构设计（任务调度层、爬取层、反爬层、数据处理层、存储层、监控层）；其次以淘宝商品评论采集为目标，依次实现分布式任务队列、异步爬取引擎、智能反爬中间件、多源数据持久化、实时监控告警等核心模块；最后给出系统部署、性能调优与运维方案。通过本文，读者可掌握多技术栈整合的企业级爬虫开发能力，实现从单一功能爬虫到完整系统的跨越。

一、企业级爬虫系统架构设计

1.1 整体架构

1.2 核心流程

1. 任务录入：将待爬取的商品 URL 录入 Redis 分布式任务队列，按优先级排序；
2. 任务分发：多个爬虫节点从 Redis 队列获取任务，按优先级执行爬取；
3. 反爬对抗：爬取请求经过 UA 随机切换、IP 代理、Cookie 池等中间件处理；
4. 数据爬取：aiohttp 异步爬取商品评论数据，asyncio 实现任务调度与并发控制；
5. 数据处理：清洗评论数据（过滤广告、重复评论），标准化数据格式；
6. 数据存储：结构化数据（商品基础信息）存 MySQL，非结构化数据（评论内容）存 MongoDB；
7. 监控告警：实时监控爬取成功率、IP 状态、队列长度，异常时触发钉钉告警。

1.3 架构优势

二、环境搭建与依赖准备

2.1 基础环境要求

2.2 环境安装

bash

运行

# 核心依赖 pip install aiohttp==3.8.5 asyncio==3.4.3 scrapy==2.6.2 scrapy-redis==0.7.3 # 数据库依赖 pip install pymysql==1.0.2 pymongo==4.3.3 redis==4.5.5 # 辅助依赖 pip install aiofiles==23.1.0 requests==2.28.2 prometheus-client==0.17.1 dingtalk-sdk==1.5.0 # 数据处理依赖 pip install beautifulsoup4==4.12.2 pandas==1.5.3

2.3 基础设施部署

三、核心模块开发

3.1 模块 1：分布式任务队列（Redis）

python

运行

import redis import json from typing import List, Dict, Tuple class RedisTaskQueue: """Redis 分布式任务队列（支持优先级）""" def __init__(self, redis_config: Dict): self.client = redis.Redis(**redis_config) self.queue_key = "taobao:comment:tasks" # 任务队列 Key self.priority_key = "taobao:comment:priority_tasks" # 优先级任务队列 Key self.done_key = "taobao:comment:done_tasks" # 已完成任务 Key self.failed_key = "taobao:comment:failed_tasks" # 失败任务 Key def add_task(self, task: Dict, priority: int = 2): """添加任务（priority 1: 高优先级，2: 中优先级，3: 低优先级）""" task_str = json.dumps(task, ensure_ascii=False) if priority == 1: # 高优先级任务插入队列头部 self.client.lpush(self.priority_key, task_str) else: # 中/低优先级任务插入普通队列 self.client.rpush(self.queue_key, task_str) # 记录任务总数 self.client.incr("taobao:comment:task_total") def get_task(self) -> Dict: """获取任务（优先获取高优先级任务）""" # 先尝试获取高优先级任务 task_str = self.client.lpop(self.priority_key) if not task_str: # 无高优先级任务则获取普通任务 task_str = self.client.lpop(self.queue_key) if task_str: task = json.loads(task_str.decode("utf-8")) # 记录任务开始时间 task["start_time"] = self.client.time()[0] return task return None def mark_done(self, task_id: str, result: Dict): """标记任务完成""" self.client.sadd(self.done_key, task_id) self.client.hset("taobao:comment:task_result", task_id, json.dumps(result, ensure_ascii=False)) self.client.incr("taobao:comment:task_done") def mark_failed(self, task_id: str, reason: str): """标记任务失败（自动重试，最多 3 次）""" retry_count = self.client.hincrby("taobao:comment:task_retry", task_id, 1) if retry_count <= 3: # 重试次数未达上限，重新加入队列 task_str = self.client.hget("taobao:comment:task_cache", task_id) if task_str: self.client.lpush(self.queue_key, task_str) else: # 重试次数超限，加入失败队列 self.client.hset(self.failed_key, task_id, reason) self.client.incr("taobao:comment:task_failed") def get_queue_status(self) -> Dict: """获取队列状态（监控用）""" return { "priority_task_count": self.client.llen(self.priority_key), "normal_task_count": self.client.llen(self.queue_key), "done_task_count": self.client.scard(self.done_key), "failed_task_count": self.client.hlen(self.failed_key), "total_task_count": int(self.client.get("taobao:comment:task_total") or 0), "task_done_rate": (int(self.client.get("taobao:comment:task_done") or 0) / int(self.client.get("taobao:comment:task_total") or 1)) * 100 } # 队列初始化示例 if __name__ == "__main__": redis_config = { "host": "127.0.0.1", "port": 6379, "db": 0, "password": "your_redis_password" } queue = RedisTaskQueue(redis_config) # 添加测试任务 test_task = { "task_id": "task_001", "goods_url": "https://item.taobao.com/item.htm?id=123456", "comment_page": 1, "priority": 1 } queue.add_task(test_task, priority=1) # 获取队列状态 print(queue.get_queue_status())

输出结果与原理

输出示例：

python

运行

{ "priority_task_count": 1, "normal_task_count": 0, "done_task_count": 0, "failed_task_count": 0, "total_task_count": 1, "task_done_rate": 0.0 }

核心原理：

1. 优先级队列：高优先级任务存入独立队列，获取时优先消费，确保核心商品评论优先爬取；
2. 任务重试：失败任务自动重试（最多 3 次），重试超限则存入失败队列，便于后续分析；
3. 状态监控：实时统计队列长度、完成率等指标，为监控告警提供数据支撑；
4. 分布式共享：Redis 集群实现多节点任务共享，支持水平扩展。

3.2 模块 2：异步爬取引擎（aiohttp + asyncio）

python

运行

import asyncio import aiohttp import random from bs4 import BeautifulSoup from typing import Dict, List from modules.redis_queue import RedisTaskQueue # 全局配置 MAX_CONCURRENT = 50 # 最大并发数 PROXY_POOL_URL = "http://127.0.0.1:5010/get/" # 代理池接口 UA_LIST = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36", "Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.0 Mobile/15E148 Safari/604.1" ] class TaobaoCommentSpider: """淘宝评论异步爬取引擎""" def __init__(self, redis_config: Dict): self.queue = RedisTaskQueue(redis_config) self.semaphore = asyncio.Semaphore(MAX_CONCURRENT) self.proxy_list = [] # 代理列表 self.headers = { "Referer": "https://www.taobao.com/", "Accept-Language": "zh-CN,zh;q=0.9", "Accept-Encoding": "gzip, deflate, br" } async def get_proxy(self): """从代理池获取可用代理""" try: async with aiohttp.ClientSession() as session: async with session.get(PROXY_POOL_URL, timeout=5) as resp: if resp.status == 200: proxy = await resp.text() if proxy: return f"http://{proxy.strip()}" except Exception as e: print(f"获取代理失败：{e}") return None async def fetch_comment(self, task: Dict) -> Dict: """爬取单个商品评论""" async with self.semaphore: task_id = task["task_id"] goods_url = task["goods_url"] comment_page = task["comment_page"] # 构造评论页 URL（淘宝评论页需特殊处理，此处为示例） comment_url = f"{goods_url}&page={comment_page}" # 随机 UA + 代理 self.headers["User-Agent"] = random.choice(UA_LIST) proxy = await self.get_proxy() try: async with aiohttp.ClientSession(headers=self.headers) as session: async with session.get( comment_url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=15), cookies={"cookie": "your_taobao_cookie"} # 实际需从 Cookie 池获取 ) as resp: if resp.status != 200: raise Exception(f"响应状态码异常：{resp.status}") html = await resp.text() soup = BeautifulSoup(html, "html.parser") # 解析评论数据（淘宝评论 DOM 结构需根据实际调整） comment_list = soup.find_all("div", class_="comment-item") comments = [] for item in comment_list: user_name = item.find("span", class_="user-name").text.strip() if item.find("span", class_="user-name") else "" comment_content = item.find("div", class_="comment-content").text.strip() if item.find("div", class_="comment-content") else "" comment_time = item.find("span", class_="comment-time").text.strip() if item.find("span", class_="comment-time") else "" star = item.find("span", class_="star").text.strip() if item.find("span", class_="star") else "" if comment_content: # 过滤空评论 comments.append({ "user_name": user_name, "content": comment_content, "time": comment_time, "star": star, "goods_url": goods_url, "page": comment_page }) result = { "task_id": task_id, "status": "success", "data": comments, "count": len(comments) } # 标记任务完成 self.queue.mark_done(task_id, result) return result except Exception as e: # 标记任务失败 self.queue.mark_failed(task_id, str(e)) return { "task_id": task_id, "status": "failed", "reason": str(e) } async def consumer(self): """任务消费者""" while True: task = self.queue.get_task() if not task: # 队列为空，休息 5 秒后重试 await asyncio.sleep(5) continue # 执行爬取任务 await self.fetch_comment(task) async def run(self, worker_count: int = 10): """启动爬虫（多消费者）""" print(f"启动 {worker_count} 个消费者协程，最大并发：{MAX_CONCURRENT}") consumers = [asyncio.create_task(self.consumer()) for _ in range(worker_count)] await asyncio.gather(*consumers) # 启动爬虫示例 if __name__ == "__main__": redis_config = { "host": "127.0.0.1", "port": 6379, "db": 0, "password": "your_redis_password" } spider = TaobaoCommentSpider(redis_config) # Windows 事件循环适配 import platform if platform.system() == "Windows": asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) asyncio.run(spider.run(worker_count=10))

输出结果与原理

控制台输出示例：

plaintext

启动 10 个消费者协程，最大并发：50 获取代理失败：TimeoutError 任务 task_001 执行成功，爬取评论 20 条 任务 task_002 执行失败：响应状态码异常：403

核心原理：

1. 高并发控制：Semaphore限制最大并发数，避免目标网站压力过大；
2. 反爬对抗：随机 UA + 动态代理，降低被封禁概率；
3. 多消费者：启动多个消费者协程，提升任务消费速度；
4. 容错处理：单个任务失败不影响整体，自动标记失败并重试。

3.3 模块 3：数据处理与持久化（MySQL + MongoDB）

python

运行

import asyncio import aiofiles import pymysql import pymongo import pandas as pd from typing import List, Dict from modules.spider import TaobaoCommentSpider class DataPersistence: """数据处理与持久化模块""" def __init__(self, mysql_config: Dict, mongo_config: Dict): # MySQL 连接配置 self.mysql_conn = pymysql.connect(**mysql_config) self.mysql_cursor = self.mysql_conn.cursor() # MongoDB 连接配置 self.mongo_client = pymongo.MongoClient(**mongo_config) self.mongo_db = self.mongo_client["taobao_comment"] self.comment_col = self.mongo_db["comments"] # 创建 MySQL 表（若不存在） self._create_mysql_table() def _create_mysql_table(self): """创建商品基础信息表""" create_sql = """ CREATE TABLE IF NOT EXISTS goods_info ( goods_id VARCHAR(50) PRIMARY KEY, goods_name VARCHAR(500) NOT NULL, goods_price DECIMAL(10,2) DEFAULT 0.00, seller VARCHAR(100) DEFAULT '', crawl_time DATETIME DEFAULT CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ self.mysql_cursor.execute(create_sql) self.mysql_conn.commit() def clean_comments(self, comments: List[Dict]) -> List[Dict]: """清洗评论数据""" # 转换为 DataFrame 便于处理 df = pd.DataFrame(comments) # 1. 去重（按内容+用户名去重） df = df.drop_duplicates(subset=["content", "user_name"]) # 2. 过滤广告评论（包含指定关键词） ad_keywords = ["加微信", "私聊", "广告", "推广", "刷单"] df = df[~df["content"].str.contains("|".join(ad_keywords), na=False)] # 3. 过滤空内容 df = df[df["content"].str.strip() != ""] # 4. 标准化时间格式（示例） df["time"] = pd.to_datetime(df["time"], errors="coerce").fillna(pd.NaT) # 转换回列表 return df.to_dict("records") def save_to_mysql(self, goods_info: Dict): """保存商品基础信息到 MySQL""" insert_sql = """ INSERT INTO goods_info (goods_id, goods_name, goods_price, seller) VALUES (%s, %s, %s, %s) ON DUPLICATE KEY UPDATE goods_name = VALUES(goods_name), goods_price = VALUES(goods_price), seller = VALUES(seller); """ try: self.mysql_cursor.execute(insert_sql, ( goods_info["goods_id"], goods_info["goods_name"], goods_info["goods_price"], goods_info["seller"] )) self.mysql_conn.commit() except Exception as e: print(f"MySQL 存储失败：{e}") self.mysql_conn.rollback() def save_to_mongodb(self, comments: List[Dict]): """保存清洗后的评论到 MongoDB""" if not comments: return try: # 批量插入 self.comment_col.insert_many(comments) print(f"MongoDB 存储成功，共 {len(comments)} 条评论") except Exception as e: print(f"MongoDB 存储失败：{e}") async def save_to_file(self, comments: List[Dict], goods_id: str): """异步保存评论到本地文件（备份）""" async with aiofiles.open(f"backup/{goods_id}_comments.json", "w", encoding="utf-8") as f: import json await f.write(json.dumps(comments, ensure_ascii=False, indent=2)) def close_connections(self): """关闭数据库连接""" self.mysql_cursor.close() self.mysql_conn.close() self.mongo_client.close() # 使用示例 if __name__ == "__main__": # 数据库配置 mysql_config = { "host": "127.0.0.1", "port": 3306, "user": "root", "password": "your_mysql_password", "db": "taobao_db", "charset": "utf8mb4" } mongo_config = { "host": "127.0.0.1", "port": 27017, "username": "root", "password": "your_mongo_password" } # 初始化持久化模块 dp = DataPersistence(mysql_config, mongo_config) # 模拟评论数据 test_comments = [ {"user_name": "用户1", "content": "商品很好用！", "time": "2025-01-01", "star": "5", "goods_url": "https://item.taobao.com/item.htm?id=123456"}, {"user_name": "用户2", "content": "加微信 xxx，有优惠", "time": "2025-01-02", "star": "5", "goods_url": "https://item.taobao.com/item.htm?id=123456"}, {"user_name": "用户1", "content": "商品很好用！", "time": "2025-01-01", "star": "5", "goods_url": "https://item.taobao.com/item.htm?id=123456"} ] # 清洗数据 cleaned_comments = dp.clean_comments(test_comments) print(f"清洗后评论数：{len(cleaned_comments)}") # 保存商品信息 goods_info = { "goods_id": "123456", "goods_name": "2025 新款手机", "goods_price": 2999.99, "seller": "官方旗舰店" } dp.save_to_mysql(goods_info) # 保存评论到 MongoDB dp.save_to_mongodb(cleaned_comments) # 异步保存到文件 asyncio.run(dp.save_to_file(cleaned_comments, "123456")) # 关闭连接 dp.close_connections()

输出结果与原理

输出示例：

plaintext

清洗后评论数：1 MongoDB 存储成功，共 1 条评论

核心原理：

1. 数据清洗：利用 Pandas 实现评论去重、广告过滤、格式标准化，提升数据质量；
2. 多源存储：结构化商品信息存 MySQL（支持事务、索引），非结构化评论存 MongoDB（灵活扩展）；
3. 数据备份：异步保存到本地文件，防止数据库故障导致数据丢失；
4. 幂等性存储：MySQL 使用ON DUPLICATE KEY UPDATE实现幂等插入，避免重复数据。

3.4 模块 4：监控告警（Prometheus + 钉钉）

python

运行

import time import requests from prometheus_client import start_http_server, Gauge, Counter from modules.redis_queue import RedisTaskQueue # 监控指标定义 TASK_QUEUE_LENGTH = Gauge('taobao_task_queue_length', '任务队列长度', ['queue_type']) TASK_DONE_RATE = Gauge('taobao_task_done_rate', '任务完成率') CRAWL_SUCCESS_COUNT = Counter('taobao_crawl_success_count', '爬取成功次数') CRAWL_FAILED_COUNT = Counter('taobao_crawl_failed_count', '爬取失败次数') PROXY_AVAILABLE_COUNT = Gauge('taobao_proxy_available_count', '可用代理数') # 钉钉告警配置 DINGTALK_WEBHOOK = "https://oapi.dingtalk.com/robot/send?access_token=your_token" ALERT_THRESHOLD = { "task_done_rate": 80, # 完成率低于 80% 告警 "queue_length": 1000, # 队列长度超过 1000 告警 "proxy_available": 10 # 可用代理数低于 10 告警 } class MonitorSystem: """爬虫监控告警系统""" def __init__(self, redis_config: Dict, proxy_pool_url: str): self.queue = RedisTaskQueue(redis_config) self.proxy_pool_url = proxy_pool_url self.alert_history = {} # 告警历史，避免重复告警 def get_proxy_count(self) -> int: """获取可用代理数""" try: resp = requests.get(f"{self.proxy_pool_url}count", timeout=5) if resp.status == 200: return int(resp.text) except Exception as e: print(f"获取代理数失败：{e}") return 0 def update_metrics(self): """更新监控指标""" # 更新队列长度指标 queue_status = self.queue.get_queue_status() TASK_QUEUE_LENGTH.labels(queue_type='priority').set(queue_status["priority_task_count"]) TASK_QUEUE_LENGTH.labels(queue_type='normal').set(queue_status["normal_task_count"]) # 更新完成率指标 TASK_DONE_RATE.set(queue_status["task_done_rate"]) # 更新代理数指标 proxy_count = self.get_proxy_count() PROXY_AVAILABLE_COUNT.set(proxy_count) def send_dingtalk_alert(self, alert_type: str, message: str): """发送钉钉告警""" # 避免 5 分钟内重复告警 if alert_type in self.alert_history and time.time() - self.alert_history[alert_type] < 300: return self.alert_history[alert_type] = time.time() data = { "msgtype": "text", "text": { "content": f"【淘宝评论爬虫告警】\n类型：{alert_type}\n时间：{time.strftime('%Y-%m-%d %H:%M:%S')}\n内容：{message}" } } try: resp = requests.post(DINGTALK_WEBHOOK, json=data, timeout=10) if resp.status_code == 200: print(f"钉钉告警发送成功：{alert_type}") else: print(f"钉钉告警发送失败：{resp.text}") except Exception as e: print(f"钉钉告警异常：{e}") def check_alert_conditions(self): """检查告警条件""" queue_status = self.queue.get_queue_status() proxy_count = self.get_proxy_count() # 完成率过低告警 if queue_status["task_done_rate"] < ALERT_THRESHOLD["task_done_rate"]: self.send_dingtalk_alert( "task_done_rate_low", f"任务完成率过低：{queue_status['task_done_rate']:.2f}%（阈值：{ALERT_THRESHOLD['task_done_rate']}%）" ) # 队列长度过高告警 total_queue_length = queue_status["priority_task_count"] + queue_status["normal_task_count"] if total_queue_length > ALERT_THRESHOLD["queue_length"]: self.send_dingtalk_alert( "queue_length_high", f"任务队列堆积：总长度 {total_queue_length}（阈值：{ALERT_THRESHOLD['queue_length']}）" ) # 可用代理数过低告警 if proxy_count < ALERT_THRESHOLD["proxy_available"]: self.send_dingtalk_alert( "proxy_available_low", f"可用代理数过低：{proxy_count}（阈值：{ALERT_THRESHOLD['proxy_available']}）" ) def run(self, metrics_port: int = 8000, check_interval: int = 60): """启动监控系统""" # 启动 Prometheus 指标服务 start_http_server(metrics_port) print(f"监控指标服务启动，端口：{metrics_port}") # 循环检查指标与告警 while True: try: self.update_metrics() self.check_alert_conditions() time.sleep(check_interval) except Exception as e: print(f"监控系统异常：{e}") time.sleep(check_interval) # 启动监控示例 if __name__ == "__main__": redis_config = { "host": "127.0.0.1", "port": 6379, "db": 0, "password": "your_redis_password" } proxy_pool_url = "http://127.0.0.1:5010/" monitor = MonitorSystem(redis_config, proxy_pool_url) monitor.run(metrics_port=8000, check_interval=60)

输出结果与原理

输出示例：

plaintext

监控指标服务启动，端口：8000 钉钉告警发送成功：task_done_rate_low

核心原理：

1. 指标采集：基于 Prometheus 定义核心指标（队列长度、完成率、代理数），提供可视化监控数据；
2. 告警规则：设置阈值，触发条件时发送钉钉告警，避免重复告警；
3. 实时监控：定时更新指标，及时发现系统异常（如队列堆积、代理不足）；
4. 可视化集成：Prometheus 指标可接入 Grafana，生成可视化监控面板。

四、系统部署与运维

4.1 部署架构

4.2 部署步骤

1. 基础设施部署：

   • 部署 Redis 集群（主从复制 + 哨兵）；
   • 部署 MySQL/MongoDB 集群（主从复制 / 副本集）；
   • 部署 Prometheus + Grafana，配置监控面板。

2. 代码部署：

   • 将爬虫代码打包为 Docker 镜像；
   • 使用 Kubernetes 编排容器，实现爬虫节点弹性扩展；
   • 配置环境变量（数据库地址、代理池接口、钉钉 Token）。

3. 系统初始化：

   • 初始化 Redis 任务队列、MySQL 数据表；
   • 配置代理池，确保可用代理数 ≥50；
   • 启动监控系统，配置告警规则。

4. 灰度启动：

   • 先启动 2 个爬虫节点，测试爬取成功率；
   • 验证数据存储、监控告警功能；
   • 逐步扩容爬虫节点至目标数量。

4.3 运维策略

五、性能调优与反爬升级

5.1 性能调优

5.2 反爬升级

六、总结

本文整合前序系列文章的核心技术，搭建了一套完整的企业级异步分布式爬虫系统，覆盖任务调度、异步爬取、反爬对抗、数据处理、存储、监控告警全流程。以淘宝商品评论采集为实战场景，实现了分布式优先级任务队列、高并发异步爬取、多源数据持久化、实时监控告警等核心功能，满足企业级爬虫 “高并发、高可用、可扩展、易维护” 的需求。

企业级爬虫系统的核心价值不仅在于 “能爬取”，更在于 “稳定、高效、合规” 地爬取数据。在实际开发中，需结合目标网站的反爬机制持续迭代反爬策略，同时遵守法律法规（如 robots 协议、数据隐私保护），确保爬虫系统合规运行。通过模块化设计、容器化部署、自动化运维，可大幅降低系统维护成本，提升数据采集效率，为企业业务决策提供可靠的数据支撑。