Python袋里IP采集怎样减少无效请求？6步优化脚本效率

采集脚本接入袋里后失败率不降反升，是个很常见的现象。原因通常不是袋里质量差，而是脚本缺少"识别失败类型 → 判断处理动作 → 针对性重试"这条闭环。本文按改造顺序拆成六步，每一步给出可运行代码，并说明它消除的是哪一类无效请求。

环境与前提

代码在以下环境验证：

Python 3.11、requests 2.31、aiohttp 3.9、urllib3 2.x 目标端：自建静态站点（用于可控压测） 若干公开数据接口 袋里：极安袋里，测试中用到两种接入形态 短效袋里：单 IP 存活 1–15 分钟（五档可选），到期自动失效 隧道袋里：脚本侧固定一个入口地址，出口 IP 由服务端轮换

后文第 4 步的 Session 复用粒度、第 6 步的冷却期取值，都是从上面这两个参数推出来的。换成其他供应商时，需要按各自的 IP 存活时长和接入方式重新标定这两个数——这也是本文把环境写在前面的原因。

合规提醒：以下方法只适用于公开数据采集。执行前请确认目标站点的 robots.txt 与服务条款，遵守其中声明的 Crawl-delay，不采集个人信息、不绕过登录态与付费墙、不对目标造成可感知的负载压力。

0. 先把失败分类，再谈优化

绝大多数"优化"失败，是因为改造前后只有一个总失败率数字，无法判断哪一步起了作用。所以第一件事是打点，把失败拆成可归因的类别。

import threadingfrom collections import Counterimport requestsclass FailureStats:"""线程安全的失败分类计数器。"""def __init__(self):self._counter = Counter()self._lock = threading.Lock()def record(self, category: str) -> None:with self._lock:self._counter[category]  = 1def report(self) -> dict:with self._lock:total = sum(self._counter.values())if not total:return { }return { k: { "count": v, "ratio": round(v / total, 4)}for k, v in self._counter.most_common()}def classify(status: int | None = None, exc: BaseException | None = None) -> str:"""把一次请求结果归到一个可决策的类别上。"""if status is not None:if 200 <= status < 300:return "ok"if status in (401, 403, 451):return "target_reject"if status == 404:return "not_found"if status == 429:return "rate_limited"if 500 <= status < 600:return "target_5xx"return f"http_{status}"# 注意判断顺序：ProxyError / SSLError / ConnectTimeout# 都是 ConnectionError 或 Timeout 的子类，先具体后笼统。if isinstance(exc, requests.exceptions.ProxyError):return "proxy_error"if isinstance(exc, requests.exceptions.SSLError):return "ssl_error"if isinstance(exc, requests.exceptions.ConnectTimeout):return "connect_timeout"if isinstance(exc, requests.exceptions.ReadTimeout):return "read_timeout"if isinstance(exc, requests.exceptions.ConnectionError):return "conn_error"return "unknown"

有了这张分布表，后面每一步改造带来的变化才是可验证的：预检生效，proxy_error 占比应当下降；分级重试生效，target_reject 的重试次数应当归零。

六步改造与它们各自消除的失败类别：

步骤	改造点	主要消除
1	入池前连通性预检	proxy_error、connect_timeout
2	超时分层	长尾 read_timeout 拖慢吞吐
3	按类别分级重试	对 target_reject / not_found 的无意义重试
4	Session 与连接池复用	握手开销、连接数膨胀
5	并发上限约束	rate_limited、target_5xx
6	失败 IP 退池与冷却	同一坏 IP 的重复命中（雪崩）

1. 入池前的连通性预检

预检要回答的是"这个 IP 此刻能不能用"，不是"以后能不能用"。所以它必须足够便宜：单次、不重试、短超时、并发受控。

from concurrent.futures import ThreadPoolExecutorimport requests# 建议换成自建的 echo 端点，公共服务本身会限速，# 预检结果会被它的限速污染。PRECHECK_URL = "https://your-own-echo.example.com/ip"def precheck(proxy: str) -> str | None:try:resp = requests.head(PRECHECK_URL,proxies={ "http": proxy, "https": proxy},timeout=(2, 3),allow_redirects=False,)return proxy if resp.status_code < 400 else Noneexcept requests.RequestException:return Nonedef precheck_batch(proxy_list: list[str], workers: int = 20) -> list[str]:with ThreadPoolExecutor(max_workers=workers) as pool:return [p for p in pool.map(precheck, proxy_list) if p]

几个容易踩的点：

用 HEAD 而非 GET，只验通路不拉正文； allow_redirects=False，跳转本身不是可用性信号； 预检不重试。预检里重试等于把"筛选"变成了"抢救"，成本会翻倍； 预检并发别开太高，20 上下即可，否则预检自己会成为瓶颈。

短效 IP 的场景下，预检和提取应该合并成一步：提取后立刻过一遍连通性再入池，中间不要有队列积压，否则 IP 在排队时就已经过期了。

2. 超时分层

timeout=10 这种写法会把"慢但可用"和"根本连不上"当成同一件事处理。requests 支持二元组，把两者拆开：

resp = requests.get(target_url,proxies={ "http": proxy, "https": proxy},timeout=(3, 15), # (connect_timeout, read_timeout)headers=default_headers,)

连接超时：3 秒内建不上连接，基本可以判定这条链路有问题，快速失败换 IP，不必等待； 读取超时：15 秒是留给慢链路的窗口，目标站点响应慢和袋里不可用是两回事。

经验区间：连接超时 2–5 秒，读取超时 10–20 秒。把 timeout 统一写成 30 秒是常见反模式——坏袋里不会被及时淘汰，还会长时间占住 worker，整体吞吐反而更低。

aiohttp 侧对应的写法粒度更细：

import aiohttptimeout = aiohttp.ClientTimeout(total=30,sock_connect=3,sock_read=15,)

3. 按失败类别分级重试

不是所有失败都值得重试。403、404 重试一百次结果一样，而 502、连接中断换个 IP 往往就通了。

类别	动作	最大次数	说明
ok	接受	—	记录成功
target_reject（401/403/451）	跳过	0	目标明确拒绝，换 IP 也无效
not_found	跳过	0	资源不存在
rate_limited（429）	延时后重试	1	优先读 Retry-After
target_5xx	原 IP 短延时重试	2	目标侧故障，与袋里无关
proxy_error / conn_error	换 IP 重试	3	链路问题
connect_timeout / read_timeout	换 IP 重试	2	袋里慢或失联
ssl_error	换 IP 重试	1	多为中间设备劫持

落成配置表而不是 if-else 链，后续调参只改数据：

import randomfrom enum import Enumclass Action(str, Enum):ACCEPT = "accept"SKIP = "skip"RETRY_SAME = "retry_same"RETRY_SWITCH = "retry_switch"DELAY_RETRY = "delay_retry"POLICY: dict[str, tuple[Action, int]] = { "ok":(Action.ACCEPT, 0),"target_reject": (Action.SKIP, 0),"not_found": (Action.SKIP, 0),"rate_limited":(Action.DELAY_RETRY,1),"target_5xx":(Action.RETRY_SAME, 2),"proxy_error": (Action.RETRY_SWITCH, 3),"conn_error":(Action.RETRY_SWITCH, 3),"connect_timeout": (Action.RETRY_SWITCH, 2),"read_timeout":(Action.RETRY_SWITCH, 2),"ssl_error": (Action.RETRY_SWITCH, 1),"unknown": (Action.RETRY_SWITCH, 1),}def decide(category: str, attempt: int) -> Action:action, max_attempts = POLICY.get(category, (Action.RETRY_SWITCH, 1))if action in (Action.ACCEPT, Action.SKIP):return actionreturn action if attempt < max_attempts else Action.SKIPdef backoff(attempt: int, base: float = 1.0, cap: float = 30.0) -> float:"""指数退避   抖动，避免多协程在同一时刻集中重试。"""exp = min(cap, base * (2 ** attempt))return exp * (0.5   random.random() * 0.5)

统一按"失败就重试 3 次"处理的脚本，重试请求里有相当大一部分是打在硬拒绝上的——这些请求全部属于无效请求，且会额外消耗 IP 配额。分级之后，重试预算才会花在真正可能成功的失败上。

429 的处理还有一个细节：优先读响应头里的 Retry-After，服务端已经给出了等待时长，自己拍一个 60 秒既可能不够也可能浪费。

def retry_after_seconds(resp, fallback: float = 60.0) -> float:value = resp.headers.get("Retry-After")if not value:return fallbacktry:return float(value)except ValueError:return fallback # HTTP-date 格式，按需再解析

4. Session 与连接池复用

每次 requests.get() 都会新建连接，重复付出 DNS 解析 TCP 握手 TLS 握手的成本。袋里链路下这段开销更明显，因为握手要多走一跳。

import threadingimport requestsfrom requests.adapters import HTTPAdapterfrom urllib3.util.retry import Retryclass SessionFactory:"""按袋里入口地址复用 Session，退池时同步关闭释放连接。"""def __init__(self, pool_connections: int = 10, pool_maxsize: int = 20):self._sessions: dict[str, requests.Session] = { }self._lock = threading.Lock()self._pool_connections = pool_connectionsself._pool_maxsize = pool_maxsizedef _build(self, proxy: str) -> requests.Session:session = requests.Session()session.proxies = { "http": proxy, "https": proxy}adapter = HTTPAdapter(pool_connections=self._pool_connections,pool_maxsize=self._pool_maxsize,# 底层重试全部关掉，重试策略由上层的 POLICY 统一决定，# 否则两套重试会叠乘，实际请求数远超预期。max_retries=Retry(total=0, backoff_factor=0, status_forcelist=[]),)session.mount("http://", adapter)session.mount("https://", adapter)return sessiondef get(self, proxy: str) -> requests.Session:with self._lock:if proxy not in self._sessions:self._sessions[proxy] = self._build(proxy)return self._sessions[proxy]def drop(self, proxy: str) -> None:with self._lock:session = self._sessions.pop(proxy, None)if session is not None:session.close()

两个关键点：

一、底层重试必须关掉。 HTTPAdapter 默认的 Retry 和第 3 步的分级重试是两套独立机制，同时开启时实际请求数是两者相乘，日志上会表现为"明明只配了 3 次重试，抓包却有 9 次"。

二、退池时要 close()。 只从字典里删掉引用，底层连接池要等 GC 才释放，长跑任务里会看到 socket 数持续上涨。

复用粒度取决于接入形态：

独立 IP 列表：Session 按 IP 建，一个 IP 一个池； 隧道形态：脚本只固定一个入口地址，出口 IP 由服务端轮换，此时整个进程可以共用一个 Session，握手成本被摊薄到接近零——这是隧道接入在工程上最实际的收益。

5. 并发上限约束（附一个高频误用）

并发不是越高越好。单 IP 对同一目标的并发超过阈值就会触发限速，rate_limited 和 target_5xx 会同时上升，整体吞吐反而下降。

定档依据：

robots.txt 声明了 Crawl-delay 的，按声明执行，不做加速； 无声明的公开数据，单 IP 并发 1–3，总并发随可用 IP 数线性扩展； 页面重、服务端渲染慢的目标，单 IP 并发压到 1。

一个非常容易写错的地方：Semaphore 必须在循环外创建。写成下面这样是无效的——

# ❌ 错误：每个 task 各自持有一个新信号量，等于没有限流for i, url in enumerate(urls):sem = asyncio.Semaphore(MAX_PER_IP)tasks.append(fetch(sessions[i % len(sessions)], url, sem))

正确写法是按袋里各建一个，并叠加一个全局信号量：

import asyncioimport aiohttpMAX_PER_IP = 2MAX_TOTAL = 30TIMEOUT = aiohttp.ClientTimeout(total=30, sock_connect=3, sock_read=15)async def fetch(session, url, ip_sem, total_sem):async with total_sem, ip_sem:try:async with session.get(url, timeout=TIMEOUT) as resp:if resp.status >= 400:return None, classify(status=resp.status)return await resp.text(), "ok"except asyncio.TimeoutError:return None, "read_timeout"except aiohttp.ClientError as exc:return None, classify(exc=exc)async def run_batch(urls: list[str], sessions: dict[str, aiohttp.ClientSession]):total_sem = asyncio.Semaphore(MAX_TOTAL)ip_sems = { proxy: asyncio.Semaphore(MAX_PER_IP) for proxy in sessions}keys = list(sessions)tasks = [fetch(sessions[keys[i % len(keys)]], url,ip_sems[keys[i % len(keys)]], total_sem)for i, url in enumerate(urls)]return await asyncio.gather(*tasks)

并发上限的另一半价值是保护本机。单进程 1000 并发时，asyncio 事件循环调度本身会成为新瓶颈，此时提高并发数只会让 P99 延迟继续恶化。

6. 失败 IP 退池与冷却

坏 IP 不退池，会被后续请求反复命中，形成雪崩：失败率上升 → 重试增多 → 又打到同一批坏 IP。

import threadingimport timefrom collections import defaultdictclass IPPool:def __init__(self, proxies, fail_threshold: int = 3, cooldown: int = 300):self._available = list(proxies)self._failed: dict[str, float] = { }self._fail_count: dict[str, int] = defaultdict(int)self._cursor = 0self._fail_threshold = fail_thresholdself._cooldown = cooldownself._lock = threading.Lock()def acquire(self) -> str | None:"""轮询取用，避免总是命中列表头部的同一个 IP。"""with self._lock:self._recover_locked()if not self._available:return Noneself._cursor = (self._cursor   1) % len(self._available)return self._available[self._cursor]def mark_success(self, proxy: str) -> None:with self._lock:self._fail_count[proxy] = 0def mark_fail(self, proxy: str) -> None:with self._lock:self._fail_count[proxy]  = 1if self._fail_count[proxy] >= self._fail_threshold:if proxy in self._available:self._available.remove(proxy)self._cursor = 0self._failed[proxy] = time.time()def _recover_locked(self) -> None:now = time.time()for proxy, failed_at in list(self._failed.items()):if now - failed_at > self._cooldown:self._failed.pop(proxy)self._fail_count[proxy] = 0self._available.append(proxy)@propertydef stats(self) -> dict:with self._lock:return { "available": len(self._available), "cooling": len(self._failed)}

几个设计取舍：

失败 3 次才退池，而不是 1 次。单次失败可能来自目标侧抖动，一次就退会误伤大量好 IP； mark_success 要清零计数，否则长跑任务里所有 IP 的失败次数只增不减，最终全池退空； 回收在 acquire 时顺带做，不额外起线程，主流程零等待。

冷却期该怎么设，本质上要看 IP 本身能活多久。对于固定长效 IP，设置 300 秒通常没什么问题；但在短效场景里，IP 往往只存活 1–15 分钟（极安袋里短效袋里提供五档可选，到期后会自动失效），这时候再给它配一个 300 秒冷却，结果往往是还没等冷却结束，IP 就已经过期了，所谓回收也就失去了实际意义。遇到这种情况，更合适的做法是把冷却时间压到 60 秒以内，把策略重点从“等它恢复”切换成“立刻换下一个”。如果是隧道形态，则没必要再保留 IPPool 这一层，退池逻辑直接交给服务端处理，脚本侧只需要保留失败计数用于告警。

整合：一个最小可运行实现

把六步串起来：

import timeimport requestsdef fetch_with_policy(url: str,pool: IPPool,factory: SessionFactory,stats: FailureStats,max_rounds: int = 5,) -> str | None:attempt = 0proxy = pool.acquire()for _ in range(max_rounds):if proxy is None:stats.record("pool_exhausted")return Nonesession = factory.get(proxy)try:resp = session.get(url, timeout=(3, 15))category = classify(status=resp.status_code)except requests.RequestException as exc:category = classify(exc=exc)resp = Nonestats.record(category)if category == "ok":pool.mark_success(proxy)return resp.textaction = decide(category, attempt)attempt  = 1if action is Action.SKIP:return Noneif action is Action.DELAY_RETRY:time.sleep(retry_after_seconds(resp) if resp is not None else 60.0)continueif action is Action.RETRY_SAME:time.sleep(backoff(attempt))continueif action is Action.RETRY_SWITCH:pool.mark_fail(proxy)factory.drop(proxy)proxy = pool.acquire()time.sleep(backoff(attempt, base=0.5))continuereturn None

改造后还剩多少无效请求？

把 FailureStats.report() 接到日志或监控上，改造前后各跑一轮同样的 URL 集合，对比分布，例如：

失败类别	改造前占比	改造后占比	主要归因
proxy_error	高	显著下降	第 1、6 步
connect_timeout	高	显著下降	第 1、2 步
rate_limited	中	明显下降	第 5 步
target_reject 上的重试量	中	归零	第 3 步
target_5xx	低	基本不变	目标侧因素，不可控

无效请求率归零是不可能的：目标站点的瞬时故障、DNS 抖动、路由波动都会产生不可避免的失败。经验上，5% 以内属于良好，10% 以内可接受，持续超过 15% 说明脚本或链路存在需要回查的问题。

超标时的回查顺序

按维度逐层收敛，不要一上来就换袋里：

看错误码分布 —— 是否集中在某一两个类别？集中在 proxy_error 查链路，集中在 rate_limited 查并发，集中在 target_reject 查请求指纹（UA、Header 顺序、Cookie）； 看时间分布 —— 是否集中在某个时段？可能是目标站点的高峰期或定时策略，也可能是自己的定时任务撞在了一起； 看目标分布 —— 是否集中在某几个 URL 或域名？那多半是目标侧规则变化，与袋里无关； 看 IP 分布 —— 是否集中在少数 IP？说明预检强度不够，或退池阈值设得太宽松。

四个维度定位到具体原因后再动手，比盲目调参有效得多。

小结

别把目光局限在“出口 IP”这一环。真正掐住无效请求率脖子的，是脚本处理失败时的逻辑：它能否精准识别故障类型？是否会在无意义的地方徒劳重试？又能否让坏掉的 IP 及时出局？这六步的顺序绝非随意排列——先建立度量体系，随后进行筛选（1）与控制单次成本（2、4），接着约束速率（5），最后才轮到失败后的处置（3、6）。跳过第 0 步直接改代码，最后往往连哪一步起了作用都说不清。