写在前面上一篇文章《Python爬虫请求头伪装后仍被反爬基于代理池随机延迟的绕过实战》里我提到过一个核心结论IP信誉评分占到了反爬拦截决策权重的60%以上。文章发出后后台收到不少留言问得最多的就是——“代理池到底怎么搭”说实话市面上免费代理一大堆但真正能用的不到10%。付费代理虽然稳定但单靠一家服务商的IP池碰上大规模采集任务照样捉襟见肘。所以这篇文章我打算手把手带你从零搭建一个属于自己的代理池。多源接入、自动验证、智能调度、全自动运行——这些功能一个不落代码全部可运行拿到就能用。一、为什么需要自己搭代理池先泼一盆冷水网上那些“一键获取免费代理”的教程十个有九个是坑。免费代理的问题在于可用率极低从公开代理站爬下来的IP能用的往往不到5%剩下的要么连不上要么延迟高到怀疑人生已被标记大量爬虫共用同一批免费代理目标网站的IP信誉系统早就把这些IP拉黑了随时失效免费代理的生命周期通常以分钟计你的爬虫刚跑起来IP就挂了付费代理能解决可用率的问题但单一服务商的IP池也有天花板——比如某代理站只提供2000个IP你的任务需要采集50万条数据每个IP即使只发50次请求也需要1万个不同的IP这中间就差了5倍的缺口。所以一个靠谱的代理池核心能力就两条多源汇聚同时接入免费源和付费源把IP池的深度和广度都拉满自动汰换实时验证、自动剔除失效IP保证池子里始终是可用的“活水”二、代理池整体架构先看一张架构图心里有个全景三个模块各司其职模块职责核心逻辑代理获取器从多个源拉取代理IP定时任务 多源并发拉取 去重代理验证器检测代理是否可用连通性测试 目标站实测 匿名性分级代理调度器分配最优代理给爬虫加权轮询 使用计数 冷却与淘汰下面我们一个模块一个模块来写。三、模块一代理获取器ProxyFetcher3.1 设计思路获取器要做的事情很简单定时从多个来源拉取代理IP去重后存入池中。这里的关键是“多源”。单一来源的IP池深度有限而且一旦服务商出问题整个爬虫就瘫痪了。多源接入相当于给代理池上了“多重保险”。3.2 免费代理源虽然是免费代理但聊胜于无。我选了几个相对稳定的免费源import requests import re import time from typing import List, Dict FREE_PROXY_SOURCES [ { name: proxy-list, url: https://www.proxy-list.download/api/v1/get?typehttp, parser: text # 返回纯文本每行一个 ip:port }, { name: proxylistplus, url: https://list.proxylistplus.com/Fresh-HTTP-Proxy-List-1, parser: html # 需要从HTML中解析 }, ] def fetch_from_free_sources() - List[Dict[str, str]]: 从免费源拉取代理返回去重后的代理列表 proxies [] seen set() for source in FREE_PROXY_SOURCES: try: resp requests.get(source[url], timeout10) if resp.status_code ! 200: continue # 简单正则提取 ip:port pattern r(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}):(\d{2,5}) matches re.findall(pattern, resp.text) for ip, port in matches: key f{ip}:{port} if key not in seen: seen.add(key) proxies.append({ip: ip, port: int(port), source: source[name]}) except Exception: continue return proxies⚠️坦诚提醒免费代理的可用率确实很低上面的代码主要是为了演示“多源汇聚”的思路。实际生产环境中建议把重心放在付费代理源上。3.3 付费代理源接入付费代理服务商通常提供API接口调用后直接返回可用IP列表。这里我封装一个通用的付费源接入器PAID_PROXY_SOURCES [ { name: provider_a, url: https://api.provider-a.com/get?num20typehttp, api_key: your_api_key_here, # 替换为你的API Key headers: {Authorization: Bearer your_token} }, # 可以继续添加更多服务商 ] def fetch_from_paid_sources() - List[Dict[str, str]]: 从付费源拉取代理 proxies [] seen set() for source in PAID_PROXY_SOURCES: try: resp requests.get( source[url], headerssource.get(headers, {}), timeout15 ) if resp.status_code ! 200: continue data resp.json() # 假设返回格式为 {data: [{ip: ..., port: ...}, ...]} for item in data.get(data, []): key f{item[ip]}:{item[port]} if key not in seen: seen.add(key) proxies.append({ ip: item[ip], port: item[port], source: source[name] }) except Exception: continue return proxies3.4 定时拉取与自动补充获取器需要定时运行持续为代理池补充新鲜血液。我用一个简单的循环来实现import threading class ProxyFetcher: def __init__(self, pool, interval: int 300): pool: 代理池实例 interval: 拉取间隔秒默认5分钟 self.pool pool self.interval interval self._running False def start(self): 启动定时拉取任务 self._running True thread threading.Thread(targetself._run, daemonTrue) thread.start() def stop(self): self._running False def _run(self): while self._running: # 拉取免费源 free_proxies fetch_from_free_sources() self.pool.add_proxies_batch(free_proxies) # 拉取付费源 paid_proxies fetch_from_paid_sources() self.pool.add_proxies_batch(paid_proxies) # 等待下一次拉取 time.sleep(self.interval)四、模块二代理验证器ProxyValidator4.1 为什么要验证代理池里塞了一堆IP不代表这些IP都能用。验证器要回答三个问题能连通吗—— 基础的TCP连接测试能访问目标站吗—— 拿目标URL做一次实际请求是匿名代理吗—— 目标站看到的IP是代理IP还是你的真实IP4.2 验证器完整实现import requests from concurrent.futures import ThreadPoolExecutor, as_completed from typing import List, Dict, Optional class ProxyValidator: def __init__(self, test_url: str https://httpbin.org/ip, timeout: int 8): test_url: 用于验证代理可用性的测试地址 timeout: 单个代理的验证超时时间 self.test_url test_url self.timeout timeout def validate_single(self, proxy: Dict[str, str]) - Optional[Dict]: 验证单个代理返回带评分的代理信息不可用则返回None proxy_url fhttp://{proxy[ip]}:{proxy[port]} proxies {http: proxy_url, https: proxy_url} score 0 try: start time.time() resp requests.get(self.test_url, proxiesproxies, timeoutself.timeout) elapsed time.time() - start if resp.status_code 200: score 1 # 连通性通过 # 检查返回的IP是否是代理IP而非真实IP origin_ip resp.json().get(origin, ) if proxy[ip] in origin_ip: score 1 # 匿名性通过 # 响应速度评分越快越好 if elapsed 2: score 2 elif elapsed 5: score 1 return { ip: proxy[ip], port: proxy[port], source: proxy.get(source, unknown), score: score, latency: round(elapsed, 2), validated_at: time.time() } except Exception: pass return None def validate_batch(self, proxies: List[Dict], max_workers: int 20) - List[Dict]: 并发验证一批代理返回验证通过的代理列表 valid_proxies [] with ThreadPoolExecutor(max_workersmax_workers) as executor: futures {executor.submit(self.validate_single, p): p for p in proxies} for future in as_completed(futures): result future.result() if result: valid_proxies.append(result) return valid_proxies评分规则说明基础连通1分匿名性验证通过1分响应时间 2秒2分响应时间 2-5秒1分满分4分代表“高速匿名代理”五、模块三代理调度器ProxyDispatcher5.1 调度策略调度器是代理池的“大脑”负责决定“下一个请求用哪个代理”。我设计了三个核心策略策略一分数优先验证器给每个代理打了分调度时优先分配高分代理保证请求质量。策略二使用均衡同一个IP不能一直用需要通过使用计数来均衡负载避免单个IP被过度使用。策略三冷却机制代理被使用后需要冷却一段时间模拟真实用户的访问间隔。冷却时间根据代理的评分动态调整——高分代理冷却短低分代理冷却长。5.2 调度器完整实现import heapq from dataclasses import dataclass, field from typing import Optional, Dict, List dataclass(orderTrue) class ProxyNode: 代理节点支持优先队列排序 priority: int # 优先级 分数 - 使用次数 * 惩罚系数 ip: str field(compareFalse) port: int field(compareFalse) score: int field(compareFalse) total_requests: int field(compareFalse) last_used: float field(compareFalse) source: str field(compareFalse) class ProxyDispatcher: def __init__(self, cooldown_base: int 30, max_use_per_ip: int 50): cooldown_base: 基础冷却时间秒 max_use_per_ip: 单个IP最大使用次数超过后自动淘汰 self.cooldown_base cooldown_base self.max_use_per_ip max_use_per_ip self._heap: List[ProxyNode] [] self._all_proxies: Dict[str, ProxyNode] {} self._lock threading.Lock() def add_proxy(self, proxy_info: Dict): 添加新代理到调度器 key f{proxy_info[ip]}:{proxy_info[port]} with self._lock: if key in self._all_proxies: return # 已存在跳过 node ProxyNode( priority0, ipproxy_info[ip], portproxy_info[port], scoreproxy_info.get(score, 1), total_requests0, last_used0, sourceproxy_info.get(source, unknown), ) self._all_proxies[key] node heapq.heappush(self._heap, node) def get_proxy(self) - Optional[Dict[str, str]]: 获取一个最优可用代理 with self._lock: now time.time() while self._heap: node heapq.heappop(self._heap) # 检查是否超过最大使用次数 if node.total_requests self.max_use_per_ip: del self._all_proxies[f{node.ip}:{node.port}] continue # 检查冷却时间 cooldown self.cooldown_base / max(node.score, 1) if now - node.last_used cooldown: # 还在冷却中放回去 heapq.heappush(self._heap, node) return None # 没有可用代理 # 更新状态 node.total_requests 1 node.last_used now node.priority node.total_requests # 使用次数越多优先级越低 heapq.heappush(self._heap, node) return { http: fhttp://{node.ip}:{node.port}, https: fhttp://{node.ip}:{node.port}, } return None def report_failure(self, ip: str, port: str): 报告代理失效从调度器中移除 key f{ip}:{port} with self._lock: if key in self._all_proxies: del self._all_proxies[key] # 重建堆简化处理生产环境可优化 self._heap [n for n in self._heap if f{n.ip}:{n.port} ! key] heapq.heapify(self._heap) def get_stats(self) - Dict: 获取当前代理池统计信息 with self._lock: total len(self._all_proxies) available sum(1 for n in self._all_proxies.values() if n.total_requests self.max_use_per_ip) return { total: total, available: available, max_use_per_ip: self.max_use_per_ip, }六、完整整合代理池主类把三个模块整合到一起封装成一个完整的代理池类import threading import time import logging from typing import List, Dict, Optional logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) class ProxyPool: 完整代理池整合获取、验证、调度三大模块 def __init__(self, test_url: str https://httpbin.org/ip, pool_min_size: int 20, fetch_interval: int 300, validate_interval: int 180): test_url: 代理验证测试地址 pool_min_size: 代理池最小容量低于此值自动补充 fetch_interval: 代理拉取间隔秒 validate_interval: 代理验证间隔秒 self.dispatcher ProxyDispatcher() self.validator ProxyValidator(test_urltest_url) self.pool_min_size pool_min_size self.fetch_interval fetch_interval self.validate_interval validate_interval self._pending_proxies: List[Dict] [] # 待验证的代理队列 self._lock threading.Lock() self._running False def add_proxies_batch(self, proxies: List[Dict]): 批量添加代理到待验证队列 with self._lock: self._pending_proxies.extend(proxies) logger.info(f已添加 {len(proxies)} 个代理到待验证队列) def _validate_loop(self): 验证循环定时验证待验证队列中的代理 while self._running: with self._lock: if self._pending_proxies: batch self._pending_proxies[:100] # 每次最多验证100个 self._pending_proxies self._pending_proxies[100:] else: batch [] if batch: valid self.validator.validate_batch(batch) for proxy in valid: self.dispatcher.add_proxy(proxy) logger.info(f验证完成{len(batch)} 个代理中 {len(valid)} 个可用) time.sleep(self.validate_interval) def _fetch_loop(self): 拉取循环定时从各源拉取新代理 while self._running: stats self.dispatcher.get_stats() if stats[available] self.pool_min_size: logger.info(f代理池可用数量不足 ({stats[available]}/{self.pool_min_size})开始拉取...) free_proxies fetch_from_free_sources() self.add_proxies_batch(free_proxies) paid_proxies fetch_from_paid_sources() self.add_proxies_batch(paid_proxies) time.sleep(self.fetch_interval) def start(self): 启动代理池服务 self._running True # 启动验证线程 validate_thread threading.Thread(targetself._validate_loop, daemonTrue) validate_thread.start() # 启动拉取线程 fetch_thread threading.Thread(targetself._fetch_loop, daemonTrue) fetch_thread.start() logger.info(代理池已启动) def stop(self): 停止代理池服务 self._running False logger.info(代理池已停止) def get_proxy(self) - Optional[Dict[str, str]]: 获取一个可用代理 return self.dispatcher.get_proxy() def report_failure(self, ip: str, port: str): 报告代理失效 self.dispatcher.report_failure(ip, port) def get_stats(self) - Dict: 获取代理池统计信息 return self.dispatcher.get_stats()七、实战测试7.1 启动代理池# 创建代理池实例 pool ProxyPool( test_urlhttps://httpbin.org/ip, pool_min_size20, fetch_interval300, # 5分钟拉取一次 validate_interval180 # 3分钟验证一次 ) # 启动代理池 pool.start() # 等待一段时间让代理池积累足够的可用IP time.sleep(30)7.2 在爬虫中使用代理池import requests import random def crawl_with_proxy_pool(url: str, pool: ProxyPool, max_retries: int 3): 使用代理池发起请求失败自动重试并切换代理 for attempt in range(max_retries): proxy pool.get_proxy() if not proxy: logger.warning(代理池已空等待补充...) time.sleep(10) continue try: resp requests.get( url, proxiesproxy, timeout10, headers{ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 } ) if resp.status_code 200: return resp.text if resp.status_code in [403, 429]: # 代理被目标站封禁报告失效 proxy_url proxy[http] ip, port proxy_url.replace(http://, ).split(:) pool.report_failure(ip, port) logger.warning(f代理 {ip}:{port} 被封锁已移除) except Exception as e: logger.warning(f请求失败: {e}尝试下一个代理) # 随机延迟避免触发频率限制 time.sleep(random.uniform(1, 3)) return None # 使用示例 html crawl_with_proxy_pool(https://www.example.com/product/12345, pool) if html: print(数据获取成功) else: print(所有代理均失败请检查代理池状态)7.3 查看代理池状态stats pool.get_stats() print(f代理池状态总数 {stats[total]}可用 {stats[available]})八、总结与优化方向8.1 你已经拥有了什么一套完整的、可运行的代理池系统包含能力实现方式多源代理拉取免费源 付费源定时自动拉取代理质量验证连通性 匿名性 响应速度三重检测智能调度分配分数优先 使用均衡 冷却机制自动汰换失效代理自动剔除超限代理自动淘汰全自动运行拉取、验证、调度三线程协同无需人工干预8.2 可以进一步优化的方向持久化存储把代理存到Redis或SQLite重启后不丢失已验证的代理自适应冷却根据目标站的返回状态码403/429/200动态调整冷却时间Web管理界面用Flask搭一个简易面板可视化展示代理池状态代理质量衰减长时间未使用的代理自动降分模拟真实IP的“休眠”行为目标站专项验证针对不同目标站分别验证因为一个代理在A站可用不代表在B站也能用8.3 最后想说搭建代理池这件事本质上是在和反爬系统打一场“信息不对称”的仗。你手里的IP越多、越分散、行为越像真人对方就越难判断你到底是爬虫还是用户。这个代理池就是你在这场博弈里最扎实的底牌。下一篇预告如何用这个代理池配合随机延迟和Cookie池搭建一套企业级的分布式爬虫系统。