企业级分布式爬虫系统：从架构设计到代码落地-育师

在数据驱动的时代，大规模爬虫早已不是单机脚本就能胜任的工作。面对目标网站日益严苛的反爬机制，一套稳定、高效、可扩展的分布式爬虫系统，是企业数据团队的必备基础设施。

本文将结合实战经验，从架构设计到完整代码实现，手把手带你搭建这样一套系统。核心武器只有三个：代理池、Cookie池和随机延迟器，三者配合使用时效果是乘法关系而非加法——代理换了IP但Cookie没变，或者延迟不够随机，反爬系统都能通过关联分析识别出来。只有三管齐下，才能真正做到"像真人一样访问"。

一、先理清架构：从"单兵作战"到"集团军"

一个企业级的分布式爬虫系统，通常包含以下角色：

任务调度器：接收抓取需求，生成任务队列，根据策略分配给不同的工作节点
爬虫工作节点：实际执行 HTTP 请求、解析页面、提取数据的进程或容器
代理池：为每个工作节点动态提供出口 IP，避免单一 IP 被封锁
Cookie池：维护大量有效登录态，模拟不同用户，降低频率限制与风控拦截
监控与日志：实时观察各模块健康状态，为自动降级、报警提供依据

整体架构如下：

三层分离：Master 负责资源管理和任务分发，Worker 负责执行采集，Redis 作为中间协调层。我们接下来逐个拆解。

二、代理池：让每次请求都"换一张脸"

代理池的目标是高可用、低延迟、自动轮换。常用的实现思路是利用 Redis 作为存储中心，结合定时验证与自动补充机制。

2.1 存储与获取

使用 Redis 的集合（Set）或有序集合（Sorted Set）存储代理 IP，每条记录为ip:port字符串。同时记录每个代理的可用次数、成功率、最后验证时间等元数据，方便调度时择优选择。

获取方面，从付费代理服务商 API 批量拉取，写入 Redis。免费代理稳定性差，企业场景不推荐。

2.2 验证机制

启动一个独立的验证服务，定时从 Redis 中随机抽取代理，请求目标网站的某个轻量接口（如首页），如果返回状态码 200 且响应时间在阈值内，则标记为有效；否则降低权重或直接移除。

2.3 调度策略

工作节点每次发起请求前，调用代理池接口随机获取一个有效代理。采用加权随机：根据代理的历史成功率和响应速度分配概率，让优质代理承担更多流量，同时保证所有代理都有机会被使用，避免局部过热。

代码实现

import redis import time import random import requests import threading import logging from typing import Optional, Dict, List from urllib.parse import urlparse logger = logging.getLogger(__name__) class ProxyPool: """ 基于Redis的代理池 核心功能：多源拉取、自动验证、分数调度 """ def __init__( self, redis_host: str = "localhost", redis_port: int = 6379, redis_db: int = 0, key_prefix: str = "proxy_pool", test_url: str = "https://httpbin.org/ip", test_timeout: int = 5, pool_min_size: int = 20, pool_max_size: int = 200, ): self.client = redis.Redis( host=redis_host, port=redis_port, db=redis_db, decode_responses=True ) self.key_prefix = key_prefix self.test_url = test_url self.test_timeout = test_timeout self.pool_min_size = pool_min_size self.pool_max_size = pool_max_size self._running = False self._lock = threading.Lock() def _proxy_key(self) -> str: return f"{self.key_prefix}:proxies" def _score_key(self) -> str: return f"{self.key_prefix}:scores" def add_proxy(self, ip: str, port: int, source: str = "api"): """添加代理到池中，初始分数100""" proxy_str = f"{ip}:{port}" with self._lock: self.client.zadd(self._proxy_key(), {proxy_str: 100}) self.client.hset(self._score_key(), proxy_str, f"100|{source}|{time.time()}") def add_proxies_batch(self, proxies: List[Dict]): """批量添加代理""" for p in proxies: self.add_proxy(p.get("ip"), p.get("port"), p.get("source", "batch")) def get_proxy(self) -> Optional[Dict[str, str]]: """ 获取一个可用代理（加权随机策略） 分数越高，被选中的概率越大 """ with self._lock: 获取所有代理及其分数 all_proxies = self.client.zrangebyscore( self._proxy_key(), 10, 100, withscores=True ) if not all_proxies: return None 加权随机选择：分数越高概率越大 proxies_list = [p[0] for p in all_proxies] weights = [p[1] for p in all_proxies] chosen = random.choices(proxies_list, weights=weights, k=1)[0] ip, port = chosen.split(":") return { "http": f"http://{ip}:{port}", "https": f"http://{ip}:{port}", } def report_success(self, ip: str, port: int): """报告代理使用成功，增加分数""" proxy_str = f"{ip}:{port}" current_score = self.client.zscore(self._proxy_key(), proxy_str) if current_score is not None: new_score = min(current_score + 5, 100) self.client.zadd(self._proxy_key(), {proxy_str: new_score}) def report_failure(self, ip: str, port: int): """报告代理使用失败，降低分数""" proxy_str = f"{ip}:{port}" current_score = self.client.zscore(self._proxy_key(), proxy_str) if current_score is not None: new_score = current_score - 20 if new_score <= 0: 分数归零，移出代理池 self.client.zrem(self._proxy_key(), proxy_str) self.client.hdel(self._score_key(), proxy_str) logger.info(f"代理 {proxy_str} 已被移除（分数归零）") else: self.client.zadd(self._proxy_key(), {proxy_str: new_score}) def _validate_proxy(self, proxy_str: str) -> bool: """验证单个代理是否可用""" ip, port = proxy_str.split(":") proxies = {"http": f"http://{ip}:{port}", "https": f"http://{ip}:{port}"} try: resp = requests.get( self.test_url, proxies=proxies, timeout=self.test_timeout, ) return resp.status_code == 200 except Exception: return False def _validate_loop(self): """后台验证循环""" while self._running: 随机抽取部分代理进行验证 all_proxies = self.client.zrange(self._proxy_key(), 0, -1) sample_size = min(len(all_proxies), 10) if sample_size > 0: samples = random.sample(all_proxies, sample_size) for proxy_str in samples: if self._validate_proxy(proxy_str): ip, port = proxy_str.split(":") self.report_success(ip, port) else: ip, port = proxy_str.split(":") self.report_failure(ip, port) time.sleep(30) 每30秒验证一轮 def _fetch_proxies_loop(self): """后台拉取代理循环（需子类实现具体拉取逻辑）""" while self._running: current_count = self.client.zcard(self._proxy_key()) if current_count < self.pool_min_size: logger.info(f"代理池不足（{current_count}/{self.pool_min_size}），触发拉取...") self._fetch_from_api() time.sleep(60) def _fetch_from_api(self): """ 从代理服务商API拉取代理 实际使用时替换为具体的API调用逻辑 """ 示例：付费代理API try: resp = requests.get("https://api.example-proxy.com/get?count=50", timeout=10) data = resp.json() for item in data.get("proxies", []): self.add_proxy(item["ip"], item["port"], source="api") except Exception as e: logger.error(f"代理API拉取失败: {e}") def start(self): """启动代理池后台服务""" self._running = True threading.Thread(target=self._validate_loop, daemon=True).start() threading.Thread(target=self._fetch_proxies_loop, daemon=True).start() logger.info("代理池已启动") def stop(self): """停止代理池""" self._running = False def get_stats(self) -> Dict: """获取代理池统计信息""" total = self.client.zcard(self._proxy_key()) available = self.client.zcount(self._proxy_key(), 50, 100) return {"total": total, "available": available}

三、Cookie池：模拟海量真实用户

单一 IP 配合单 Cookie 频繁抓取，依然容易被识别为爬虫。Cookie池的本质是模拟不同用户的登录状态，让每个请求看起来都像是一个独立、正常的访问者。

3.1 存储设计

同样依托 Redis。每个账号对应一组 Cookie 信息，以 JSON 形式存储，同时维护一个"可用Cookie队列"，存放所有验证通过的 Cookie 的唯一标识。

Cookie池的核心难点不在于"存"，而在于自动维护——Cookie会过期、会被标记、需要定期刷新。每个Cookie绑定一个"账号身份"，实现自动轮换和失效淘汰。

3.2 生成与更新

模拟登录：使用 Selenium 或 Playwright 等无头浏览器，配合账号密码，自动完成登录流程，获取完整的 Cookie 信息
定时检测：每个 Cookie 对应一个检测链接（比如用户个人中心页面），验证服务周期性请求该链接，若返回登录页或状态码异常，则判定失效，删除该 Cookie 并触发重新登录
补充机制：当可用 Cookie 数量低于阈值时，自动启动一批登录任务，利用账号池生成新 Cookie

3.3 使用方式

工作节点请求时，从 Cookie 池的接口随机获取一个有效 Cookie，将其附加到请求头中。如果某次请求返回"未登录"或"Cookie 过期"，立即丢弃该 Cookie 并重试。

代码实现

import redis import json import time import threading import requests from typing import Optional, Dict, List from dataclasses import dataclass, asdict @dataclass class CookieAccount: """Cookie账号实体""" account_id: str cookies: Dict[str, str] user_agent: str source: str 来源：manual / auto_login / selenium created_at: float last_used: float use_count: int max_use: int 最大使用次数 is_valid: bool class CookiePool: """ 基于Redis的Cookie池 支持自动轮换、失效淘汰、定时刷新 """ def __init__( self, redis_host: str = "localhost", redis_port: int = 6379, redis_db: int = 1, key_prefix: str = "cookie_pool", default_max_use: int = 200, ): self.client = redis.Redis( host=redis_host, port=redis_port, db=redis_db, decode_responses=True ) self.key_prefix = key_prefix self.default_max_use = default_max_use self._lock = threading.Lock() def _make_key(self, account_id: str) -> str: return f"{self.key_prefix}:account:{account_id}" def _make_queue_key(self) -> str: return f"{self.key_prefix}:queue" def add_account(self, account: CookieAccount): """添加一个Cookie账号""" account.created_at = account.created_at or time.time() account.is_valid = True self.client.hset( self._make_key(account.account_id), mapping={"data": json.dumps(asdict(account))}, ) 加入可用队列 self.client.lpush(self._make_queue_key(), account.account_id) def get_cookie(self) -> Optional[Dict]: """获取一个可用Cookie（轮询方式）""" with self._lock: 从队列右侧弹出（FIFO） account_id = self.client.rpoplpush( self._make_queue_key(), self._make_queue_key() ) if not account_id: return None key = self._make_key(account_id) raw = self.client.hget(key, "data") if not raw: return None account_data = json.loads(raw) 检查是否超限 if account_data["use_count"] >= account_data.get("max_use", self.default_max_use): self._mark_invalid(account_id) return self.get_cookie() 递归获取下一个 更新使用计数 account_data["use_count"] += 1 account_data["last_used"] = time.time() self.client.hset(key, "data", json.dumps(account_data)) return { "cookies": account_data["cookies"], "user_agent": account_data["user_agent"], "account_id": account_id, } def _mark_invalid(self, account_id: str): """标记账号失效""" key = self._make_key(account_id) raw = self.client.hget(key, "data") if raw: data = json.loads(raw) data["is_valid"] = False self.client.hset(key, "data", json.dumps(data)) 从队列中移除 self.client.lrem(self._make_queue_key(), 0, account_id) def report_failure(self, account_id: str): """报告Cookie失效（如返回登录页）""" self._mark_invalid(account_id) def get_stats(self) -> Dict: """获取Cookie池统计""" queue_len = self.client.llen(self._make_queue_key()) return { "available_accounts": queue_len, "key_prefix": self.key_prefix, } def add_accounts_batch(self, accounts: List[CookieAccount]): """批量添加账号""" for account in accounts: self.add_account(account) def auto_refresh_loop(self, refresh_func, interval: int = 1800): """ 自动刷新循环（需在子线程中运行） refresh_func: 刷新函数，接收 account_id，返回新的 CookieAccount 或 None interval: 刷新间隔（秒），默认30分钟 """ def _loop(): while True: for key in self.client.scan_iter(f"{self.key_prefix}:account:*"): raw = self.client.hget(key, "data") if not raw: continue data = json.loads(raw) if not data["is_valid"]: continue try: new_account = refresh_func(data["account_id"]) if new_account: self.add_account(new_account) except Exception: pass time.sleep(interval) thread = threading.Thread(target=_loop, daemon=True) thread.start()

四、随机延迟：让机器行为更像人

即使有代理和 Cookie，如果请求间隔完全固定，仍然会被风控系统捕捉到机器特征。随机延迟就是最后一道伪装。

很多人用简单的time.sleep(random.uniform(1, 3))，但在企业级场景里，固定区间的均匀分布很容易被反爬系统通过统计学手段识别出来。真实用户的请求间隔通常符合重尾分布——少数请求间隔很短，多数请求集中在某个区间，偶尔出现长间隔。

企业级系统需要更精细的控制：

基于请求量的动态延迟：当某个域名或 API 的请求频率接近上限时，自动拉长延迟
页面间跳转延迟：模拟人类阅读页面的时间，比如在抓取详情页前，先随机等待 2~5 秒
任务级延迟：不同任务之间加入随机间隔，避免同一账号在短时间内跨多个主题抓取

延迟策略最好作为爬虫工作节点的一个可配置参数，方便根据目标站的容忍度动态调整。

代码实现

import random import time import logging from typing import Optional logger = logging.getLogger(__name__) class RandomDelayer: """ 模拟真实用户请求间隔的延迟器 - 短间隔：概率高（模拟连续点击） - 中间隔：概率最高（模拟正常浏览） - 长间隔：概率低（模拟离开/思考） """ def __init__( self, short_interval: tuple = (0.5, 2.0), mid_interval: tuple = (3.0, 8.0), long_interval: tuple = (15.0, 45.0), short_weight: float = 0.3, mid_weight: float = 0.6, long_weight: float = 0.1, ): self.intervals = [short_interval, mid_interval, long_interval] self.weights = [short_weight, mid_weight, long_weight] 归一化权重 total = sum(self.weights) self.weights = [w / total for w in self.weights] def delay(self, min_seconds: Optional[float] = None) -> float: """ 执行一次随机延迟 min_seconds: 最小延迟秒数，用于强制冷却场景 """ 按照权重随机选择一个区间 interval = random.choices(self.intervals, weights=self.weights, k=1)[0] delay_time = random.uniform(*interval) if min_seconds is not None: delay_time = max(delay_time, min_seconds) logger.debug(f"延迟 {delay_time:.2f} 秒") time.sleep(delay_time) return delay_time def adaptive_delay(self, status_code: int, consecutive_requests: int) -> float: """ 自适应延迟：根据目标站响应状态动态调整 - 200：正常延迟 - 429：触发长延迟，并用指数退避 - 403：触发强制冷却 """ if status_code == 200: if consecutive_requests > 20: 连续请求过多，增加中长间隔权重 return self._weighted_delay(short=0.1, mid=0.4, long=0.5) return self.delay() elif status_code == 429: 触发了频率限制，指数退避 backoff = min(2 ** consecutive_requests, 120) logger.warning(f"触发429限流，退避 {backoff} 秒") time.sleep(backoff) return backoff elif status_code == 403: 可能被临时封禁，强制冷却 cooldown = random.uniform(60, 180) logger.warning(f"触发403封禁，强制冷却 {cooldown:.0f} 秒") time.sleep(cooldown) return cooldown return self.delay() def _weighted_delay(self, short: float, mid: float, long: float) -> float: """临时覆盖权重执行延迟""" total = short + mid + long interval = random.choices( self.intervals, weights=[short / total, mid / total, long / total], k=1, )[0] delay_time = random.uniform(*interval) time.sleep(delay_time) return delay_time

五、分布式协作：让一切运转起来

有了上述三大组件，需将它们组合成分布式系统。推荐以消息队列作为任务分发管道，这里用Redis的List做任务队列，LPUSH入队、BRPOP出队，天然支持多Worker并发消费。、

5.1 分布式任务队列

import redis import json import hashlib import time from typing import Optional, Dict, Any class TaskQueue: """ 基于Redis的分布式任务队列 支持优先级、去重、失败重试 """ def __init__( self, redis_host: str = "localhost", redis_port: int = 6379, redis_db: int = 2, queue_name: str = "crawl_tasks", retry_queue_name: str = "crawl_tasks_retry", max_retries: int = 3, ): self.client = redis.Redis( host=redis_host, port=redis_port, db=redis_db, decode_responses=True ) self.queue_name = queue_name self.retry_queue_name = retry_queue_name self.max_retries = max_retries 去重集合（基于任务指纹） self._dedup_key = f"{queue_name}:dedup" def push(self, task: Dict[str, Any], dedup: bool = True) -> bool: """ 添加任务到队列 dedup: 是否去重（基于URL+参数生成指纹） """ task_json = json.dumps(task, ensure_ascii=False) if dedup: fingerprint = self._fingerprint(task) if self.client.sismember(self._dedup_key, fingerprint): return False self.client.sadd(self._dedup_key, fingerprint) self.client.lpush(self.queue_name, task_json) return True def push_batch(self, tasks: list, dedup: bool = True) -> int: """批量添加任务，返回实际添加数量""" count = 0 for task in tasks: if self.push(task, dedup=dedup): count += 1 return count def pop(self, timeout: int = 5) -> Optional[Dict[str, Any]]: """ 阻塞式获取任务 timeout: 阻塞超时时间（秒），设为0表示永久阻塞 """ result = self.client.brpop(self.queue_name, timeout=timeout) if result is None: return None _, task_json = result return json.loads(task_json) def retry(self, task: Dict[str, Any]): """任务失败，放入重试队列""" retry_count = task.get("_retry_count", 0) if retry_count >= self.max_retries: 超过最大重试次数，放入死信队列 self.client.lpush( f"{self.queue_name}:dead", json.dumps(task, ensure_ascii=False) ) return task["_retry_count"] = retry_count + 1 指数退避延迟 delay = 2 ** (retry_count + 1) * 60 2分钟、4分钟、8分钟 task["_retry_at"] = time.time() + delay self.client.lpush(self.retry_queue_name, json.dumps(task, ensure_ascii=False)) def _fingerprint(self, task: Dict) -> str: """生成任务指纹（用于去重）""" raw = task.get("url", "") + str(task.get("params", {})) return hashlib.md5(raw.encode()).hexdigest() def get_queue_length(self) -> int: return self.client.llen(self.queue_name)

5.2 分布式Worker

Worker 是真正干活的部分，它从任务队列取任务，从代理池拿代理，从 Cookie 池拿 Cookie，加随机延迟，执行请求，处理结果：

import time import logging import traceback from typing import Optional logger = logging.getLogger(__name__) class DistributedWorker: """ 分布式爬虫Worker 组合代理池 + Cookie池 + 随机延迟器 + 任务队列 """ def __init__( self, worker_id: str, task_queue: TaskQueue, proxy_pool, cookie_pool: Optional[CookiePool] = None, delayer: Optional[RandomDelayer] = None, parse_func: callable = None, result_handler: callable = None, ): self.worker_id = worker_id self.task_queue = task_queue self.proxy_pool = proxy_pool self.cookie_pool = cookie_pool self.delayer = delayer or RandomDelayer() self.parse_func = parse_func 解析函数 self.result_handler = result_handler 结果处理函数 self._running = False def start(self): """启动Worker，持续消费任务""" self._running = True logger.info(f"Worker [{self.worker_id}] 已启动") consecutive_success = 0 连续成功计数 while self._running: try: 1. 从队列取任务 task = self.task_queue.pop(timeout=10) if task is None: logger.debug(f"Worker [{self.worker_id}] 等待任务...") continue 2. 获取代理 proxy = self.proxy_pool.get_proxy() if not proxy: logger.warning("代理池已空，等待补充...") time.sleep(30) self.task_queue.push(task) 任务放回队列 continue 3. 获取Cookie cookies = None headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" } if self.cookie_pool: cookie_data = self.cookie_pool.get_cookie() if cookie_data: cookies = cookie_data["cookies"] headers["User-Agent"] = cookie_data.get( "user_agent", headers["User-Agent"] ) 4. 执行请求 status_code, response_text = self._execute_request( task["url"], headers=headers, cookies=cookies, proxy=proxy, ) 5. 自适应延迟 self.delayer.adaptive_delay(status_code, consecutive_success) 6. 处理响应 if status_code == 200: consecutive_success += 1 if self.parse_func: data = self.parse_func(response_text, task) if self.result_handler: self.result_handler(data) elif status_code in [403, 429]: consecutive_success = 0 代理失效 proxy_url = proxy.get("http", "") if proxy_url: ip, port = proxy_url.replace("http://", "").split(":") self.proxy_pool.report_failure(ip, port) 任务放回重试 self.task_queue.retry(task) else: consecutive_success = 0 self.task_queue.retry(task) except KeyboardInterrupt: self._running = False except Exception: logger.error(f"Worker [{self.worker_id}] 异常: {traceback.format_exc()}") time.sleep(5) logger.info(f"Worker [{self.worker_id}] 已停止") def _execute_request( self, url: str, headers: dict, cookies: Optional[dict], proxy: dict ): """执行HTTP请求""" import requests try: resp = requests.get( url, headers=headers, cookies=cookies, proxies=proxy, timeout=15, allow_redirects=True, ) return resp.status_code, resp.text except requests.exceptions.Timeout: return 408, "" except requests.exceptions.ConnectionError: return 0, "" except Exception: return -1, "" def stop(self): self._running = False

六、主程序：一键启动分布式爬虫

把所有模块串起来，一个入口文件启动整套系统：

import sys import signal import threading import logging from typing import List logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) logger = logging.getLogger("master") class DistributedCrawler: """ 分布式爬虫主控程序 负责启动代理池、Worker集群、任务分发 """ def __init__( self, redis_host: str = "localhost", redis_port: int = 6379, worker_count: int = 4, proxy_pool_config: dict = None, seed_urls: List[str] = None, parse_func: callable = None, result_handler: callable = None, ): self.redis_host = redis_host self.redis_port = redis_port self.worker_count = worker_count 初始化各模块 self.task_queue = TaskQueue(redis_host=redis_host, redis_port=redis_port) self.proxy_pool = ProxyPool( test_url="https://httpbin.org/ip", pool_min_size=proxy_pool_config.get("pool_min_size", 20) if proxy_pool_config else 20, ) self.cookie_pool = CookiePool(redis_host=redis_host, redis_port=redis_port) self.delayer = RandomDelayer() self.parse_func = parse_func self.result_handler = result_handler self.seed_urls = seed_urls or [] self.workers: List[DistributedWorker] = [] self._running = False def start(self): """启动整个爬虫系统""" self._running = True logger.info("=" * 60) logger.info("分布式爬虫系统启动中...") logger.info("=" * 60) 1. 启动代理池 logger.info("[1/4] 启动代理池...") self.proxy_pool.start() 2. 加载种子任务 logger.info(f"[2/4] 加载种子任务 ({len(self.seed_urls)} 个)...") for url in self.seed_urls: self.task_queue.push({"url": url, "depth": 0, "meta": {}}) 3. 启动Worker集群 logger.info(f"[3/4] 启动 {self.worker_count} 个Worker...") for i in range(self.worker_count): worker = DistributedWorker( worker_id=f"worker-{i:03d}", task_queue=self.task_queue, proxy_pool=self.proxy_pool, cookie_pool=self.cookie_pool, delayer=self.delayer, parse_func=self.parse_func, result_handler=self.result_handler, ) thread = threading.Thread(target=worker.start, daemon=True) thread.start() self.workers.append(worker) 4. 监控循环 logger.info("[4/4] 系统运行中，按 Ctrl+C 停止...") try: while self._running: proxy_stats = self.proxy_pool.get_stats() cookie_stats = self.cookie_pool.get_stats() queue_len = self.task_queue.get_queue_length() logger.info( f"状态: 队列剩余={queue_len} | " f"代理池={proxy_stats['available']}/{proxy_stats['total']} | " f"Cookie账号={cookie_stats['available_accounts']}" ) time.sleep(30) except KeyboardInterrupt: self.stop() def stop(self): """优雅停止""" logger.info("正在停止爬虫系统...") self._running = False for worker in self.workers: worker.stop() self.proxy_pool.stop() logger.info("爬虫系统已停止") ==================== 使用示例 ==================== if __name__ == "__main__": 自定义解析函数 def my_parse(html: str, task: dict): """解析页面，提取数据并发现新链接""" from bs4 import BeautifulSoup soup = BeautifulSoup(html, "html.parser") title = soup.title.string if soup.title else "" return {"url": task["url"], "title": title} 自定义结果处理 def my_result_handler(data: dict): """处理解析结果（存数据库、写文件等）""" logger.info(f"采集结果: {data}") 启动爬虫 crawler = DistributedCrawler( redis_host="localhost", redis_port=6379, worker_count=8, proxy_pool_config={"pool_min_size": 30}, seed_urls=[ "https://httpbin.org/ip", "https://httpbin.org/user-agent", ], parse_func=my_parse, result_handler=my_result_handler, ) 注册信号处理（优雅退出） signal.signal(signal.SIGINT, lambda s, f: crawler.stop()) signal.signal(signal.SIGTERM, lambda s, f: crawler.stop()) crawler.start()

七、实战避坑指南

账号安全：不要将所有账号密码明文写在代码或配置文件中，建议使用环境变量或密钥管理服务（如 HashiCorp Vault、AWS Secrets Manager）
请求特征伪装：除了随机延迟，还要随机化 User-Agent、Accept-Language 等请求头，甚至模拟鼠标轨迹和页面滚动（使用无头浏览器时）
限速与熔断：当某个代理或 Cookie 的错误率持续升高时，应自动熔断，暂时移出可用池，并触发报警。代码中已实现基于分数的代理淘汰机制和 429/403 的自适应退避
法律合规：务必遵守目标网站的 robots.txt 协议及用户协议，不要抓取隐私数据，控制请求频率，避免对对方服务器造成压力

八、总结

企业级分布式爬虫绝非简单的"一个脚本跑天下"，它是代理池、Cookie 池、随机延迟、分布式任务调度等多套机制的精密配合。本文从架构设计到完整代码实现，覆盖了以下核心模块：

模块	核心技术	解决什么问题
代理池	多源拉取、自动验证、分数调度	突破IP封锁，分散请求来源
Cookie池	Redis存储、自动轮换、失效淘汰	维持登录态，模拟多用户
随机延迟器	加权随机、重尾分布、自适应退避	绕过频率限制，伪装人类行为
任务队列	Redis List、去重指纹、死信重试	分布式任务分发，故障容错
Worker集群	多线程消费、异常隔离、优雅退出	横向扩展采集能力