前言
标题里的「搭建」，重点不在「去哪搞到一批 IP」，而在拿到候选之后怎么把它管起来。真正的工程量集中在四件事：用什么数据结构存、多久校验一次可用性与延迟、失败几次就淘汰、以及给调用方暴露什么样的取用接口。这四件事想清楚了，池子才站得住；想不清楚，你写出来的就只是一份会过期的列表。
先把合规边界放在最前面：本文的池子服务于你自己有权发起的请求。如果目标站点已经用 403、封 IP 或验证码明确拒绝了你，那么换一个出口地址继续请求属于规避对方的技术措施，正确做法是停止对该目标的请求，改走官方 API 或联系站点申请许可。本文不会提供任何绕过封禁的成品代码，示例里的代理地址也全部是自造的本地占位数据，不指向任何真实服务。
另外，代理的来源必须是你有权使用的：自建代理、公司内部出口、你付费购买并接受其服务条款的代理服务。公开的「免费代理列表」不能作为来源——那些地址可能是别人的蜜罐，或者是会记录你全部流量的中间人，把一个不可信的中间人插进你的请求链路，风险远大于收益。
本机没有 Python 解释器，也没有装requests等第三方库，下面的代码无法在本机运行验证，只能逐行人工推演，涉及第三方库的地方会明确说明需要先pip install。正文的骨架只用标准库，这样你复制到本地就能自行验证。
一、先定接口：调用方要什么，池子给什么
搭建之前先把契约写死，后面所有实现都围绕它转。调用方需要的其实只有两个动作：
| 动作 | 签名（语义） | 返回值 | 调用方义务 |
|---|
| 取用一个出口 | acquire() | 一个代理记录，或None | 用完必须回填结果 |
| 回填使用结果 | release(record, ok, latency) | 无 | 如实回填，不要「怕淘汰」而谎报成功 |
这两个动作构成一个闭环：acquire()从池里挑一个当前最好的地址交给调用方，调用方用完把「成不成、花了多久」回填。没有回填，池子就失去了自我修正的能力——它永远不知道哪个地址已经坏了。
有个设计上的小技巧值得说：acquire()返回的不应该是一个字符串"1.2.3.4:8080"，而应该是一个记录对象（下面叫ProxyRecord）。因为调用方回填时需要指向具体那一条记录，如果只给字符串，回填时还得再查一次表，多一次出错的机会。
二、数据结构选型：字典打底，排序按需
池子的主容器用一个dict，键是host:port，值是记录对象。为什么用字典而不是列表？因为回填和淘汰都是按地址定位的，字典的查找是常数复杂度，列表则要遍历。用列表做主容器，池子上千条之后每次release都要全表扫一遍。
排序不维护，改成取用时按需排序：
# 适用于 Python 3.8+
import time
from dataclasses import dataclass, field
from typing import Dict, Optional
@dataclass
class ProxyRecord:
host: str
port: int
score: float = 1.0 # 综合评分，越大越好
latency: float = float("inf") # 最近一次成功测得的延迟（秒）
ok_count: int = 0
fail_count: int = 0
consecutive_fail: int = 0 # 连续失败次数
last_checked: float = 0.0 # 上次校验时间戳
checked: bool = False # 是否已通过过一次校验
@property
def key(self) -> str:
return f"{self.host}:{self.port}"为什么不维护一个有序结构？因为release会频繁改动分数，用堆的话每次改动都要「删除再插入」，反而更绕。池子规模在几千以内时，acquire里做一次sorted()的代价完全可以接受；真到了十万级，再考虑用带惰性删除的堆。
挑选策略用一个元组做排序键：先看分数，分数相同看延迟。这样「又稳又快」的地址自然排在前面：
# 适用于 Python 3.8+
def _rank(rec: ProxyRecord):
return (-rec.score, rec.latency)-rec.score是为了让sorted默认的升序变成「分数从高到低」；延迟越低越优先，所以延迟不取负。
三、校验：可用性与延迟，多久验一次
校验要做三件事，缺一不可：
- 连通性：能不能在超时时间内建连并拿到响应；
- 延迟：耗时多少，这个数直接进排序键；
- 来源是否被透传：有些劣质代理根本不转发，而是把你的请求直接发出去。校验时可以请求一个你控制的服务，看对方看到的来源 IP 是不是代理的出口，而不是你的真实地址。
校验的目标必须是你确信稳定且允许你去探测的地址——通常是你自己的服务。本文不给出任何具体的目标站点。
多久验一次？这里有个权衡：
| 校验频率 | 优点 | 缺点 |
|---|
| 过低（几小时一次） | 对被探测目标压力小 | 坏地址会带着旧分数被继续使用 |
| 过高（几秒一次） | 池子状态新鲜 | 把大量请求花在自我检查上，且频繁打自家探测接口 |
折中的做法是给每条记录一个校验有效期（TTL）：acquire时如果发现选中的记录已经超过 TTL 没校验过，就跳过它并顺手安排一次校验；同时后台仍有一个低频的全量校验循环兜底。这样「新的优先、旧的补验」，不会出现整个池子同一时刻全部过期。
还有一条：校验必须有超时。没有超时的校验会把刷新循环整个卡死在一条坏地址上。而且超时值要明显小于校验间隔，否则会出现「上一次还没验完，下一次又开始了」。
四、失败计数与淘汰策略
淘汰不要只看单一指标。至少同时看这三个：
- 连续失败次数：超过阈值（比如 3 次）直接移除。这是最可靠的坏地址信号。
- 分数：分数掉到阈值以下移除。它综合了成功与失败的长期表现。
- 最近成功时间：太久没成功过的，说明它要么坏了要么被限了，移除。
成功与失败的加减分不必对称——失败一次的减分要明显大于成功一次的加分（常见的做法是成功加 0.1、失败减 0.3）。原因很实际：一个坏地址造成的损失，远大于一个偶发抖动的好地址。偶发抖动的地址被多减了一点分，还能靠后续成功慢慢涨回来；而一个坏地址如果只减一点点，它会一直赖在池子里被反复选中。
分数还要随时间衰减：一条记录如果长时间没被校验（比如池子暂时空闲），它的分数应当缓慢回落，避免「几小时前很好」的地址永远钉在榜首。
淘汰可以做成「惰性」的：不必在每次release后立刻遍历全表，而是在acquire时顺手清理明显该删的记录，再配合后台循环做一次全量清理。这样高频调用下不会每次都付一次全表扫描的代价。
五、组装起来的完整骨架
下面把前四节拼成一个可读的骨架。代理列表是自造的占位数据，使用 RFC 5737 保留给文档的192.0.2.0/24网段，不会指向真实主机；校验函数留成接口，需要你按自己的授权目标实现。
# 适用于 Python 3.8+
# ProxyRecord 与排序函数 _rank 见上一节的定义，这里直接复用
import time
from typing import Dict, List, Optional
class ProxyPool:
def __init__(self, ttl: float = 300.0, min_score: float = 0.2,
max_consecutive_fail: int = 3):
self._items: Dict[str, ProxyRecord] = {}
self._ttl = ttl
self._min_score = min_score
self._max_fail = max_consecutive_fail
def load_candidates(self, records: List[ProxyRecord]) -> None:
"""从你有权使用的来源装入候选，来源不在本文讨论范围"""
for rec in records:
self._items[rec.key] = rec
def acquire(self) -> Optional[ProxyRecord]:
"""取用一个当前最优、且校验未过期的出口"""
now = time.time()
self._purge()
candidates = [r for r in self._items.values()
if r.checked and (now - r.last_checked) < self._ttl
and r.score >= self._min_score]
if not candidates:
return None
return min(candidates, key=_rank)
def release(self, rec: ProxyRecord, ok: bool,
latency: float = 0.0) -> None:
"""调用方用完回填真实结果，驱动评分与淘汰"""
if ok:
rec.ok_count += 1
rec.consecutive_fail = 0
rec.latency = latency
rec.score = min(rec.score + 0.1, 2.0)
else:
rec.fail_count += 1
rec.consecutive_fail += 1
rec.score -= 0.3
self._purge()
def mark_checked(self, rec: ProxyRecord, ok: bool,
latency: float = 0.0) -> None:
"""由校验循环调用，记录一次探活结果"""
rec.checked = True
rec.last_checked = time.time()
if ok:
rec.latency = latency
rec.score = min(rec.score + 0.05, 2.0)
rec.consecutive_fail = 0
else:
rec.consecutive_fail += 1
rec.score -= 0.2
def _purge(self) -> None:
dead = [k for k, r in self._items.items()
if r.consecutive_fail >= self._max_fail
or r.score < self._min_score]
for k in dead:
del self._items[k]
def stats(self) -> Dict[str, int]:
return {"total": len(self._items),
"checked": sum(1 for r in self._items.values() if r.checked)}几点说明。acquire()里先_purge()，再筛「已校验、未过期、分数达标」的候选，最后用min(..., key=_rank)取排名第一——因为_rank的第一个分量是-score，所以min取到的是分数最高、延迟最低的那条。release()与mark_checked()分开，是因为「调用方用了一次」和「后台主动探活」是两类事件，分开后可以给不同的权重，也方便排查问题。
入池后必须先校验一次才允许被取用——这就是checked字段存在的理由。没有这一步，你交出去的第一次请求很可能就浪费在一个从没验证过的地址上。
常见坑点
- ❌ 用列表作为池的主容器，
release时遍历整表找那条记录。
✅ 用字典以host:port为键，回填和淘汰都是常数时间定位。
- ❌ 校验函数不设超时，遇到一条坏地址就把整个刷新循环卡住。
✅ 每次校验都设超时，且超时值明显小于校验间隔；校验循环对异常做兜底。
- ❌ 刚入池就允许
acquire取用，地址从未验证过。
✅ 用一个checked标记，只有通过过一次校验的记录才进入候选集。
- ❌ 成功加 0.1 分、失败也减 0.1 分，好坏地址分数拉不开。
✅ 失败的减分明显大于成功的加分，让坏地址尽快掉到淘汰线以下。
- ❌ 只在池子初始化时全量校验一次，之后地址坏了也不知道。
✅ 给记录设 TTL，取用时检查是否过期；另有后台循环做低频全量兜底。
- ❌ 调用方谎报成功（怕地址被淘汰），或者干脆不回填结果。
✅ 回填必须如实；不回填等于让池子失去自我修正能力，坏地址会一直被选中。
- ❌ 从公开「免费代理」站抓一堆地址直接入池。
✅ 只从有权使用的来源获取；免费公开代理可能是蜜罐或中间人，风险高于收益。
- ❌ 把代理的账号密码写死在代码里并提交进仓库。
✅ 凭据走环境变量或密钥管理，配置文件加进忽略清单，日志不打印完整代理 URL。
总结
| 决策点 | 选择 | 理由 |
|---|
| 主容器 | dict，键为host:port | 回填、淘汰都按地址定位，查找要 O(1) |
| 排序 | 取用时按需sorted/min | 分数频繁变动，维护有序结构反而更绕 |
| 排序键 | (-score, latency) | 又稳又快优先；分数优先于延迟 |
| 校验时机 | 入池必验 + TTL 过期补验 + 后台全量兜底 | 兼顾新鲜度与自身开销 |
| 淘汰条件 | 连续失败次数 / 分数阈值 / 最近成功时间 | 单一指标容易误判 |
| 对外接口 | acquire()+release()闭环 | 让池子具备自我修正能力 |
搭建代理池的技术难点其实很集中：把「选择」和「评价」分开，并让评价结果真的回流。数据结构和淘汰规则都只是为这个闭环服务的。最后重复一次边界：池子服务的是你自己有权发起的请求；一旦目标站点明确拒绝了你，正确的做法是停止对该目标的请求并走官方渠道，而不是换一个出口继续打。
参考：代理与处理器相关行为以 Python 官方文档urllib.request章节为准；HTTP 状态码语义以 MDN 的状态码文档为准；本文代码未在本机运行，仅作人工推演。
