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 threading
from collections import Counter

import requests


class FailureStats:
    """线程安全的失败分类计数器。"""

    def __init__(self):
        self._counter = Counter()
        self._lock = threading.Lock()

    def record(self, category: str) -> None:
        with self._lock:
            self._counter[category] += 1

    def 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 ThreadPoolExecutor

import 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 None
    except requests.RequestException:
        return None


def 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 aiohttp

timeout = 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 random
from enum import Enum


class 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 action
    return action if attempt < max_attempts else Action.SKIP


def 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 fallback
    try:
        return float(value)
    except ValueError:
        return fallback   # HTTP-date 格式，按需再解析

4. Session 与连接池复用

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

import threading

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


class 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_connections
        self._pool_maxsize = pool_maxsize

    def _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 session

    def 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 asyncio

import aiohttp

MAX_PER_IP = 2
MAX_TOTAL = 30
TIMEOUT = 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 threading
import time
from collections import defaultdict


class 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 = 0
        self._fail_threshold = fail_threshold
        self._cooldown = cooldown
        self._lock = threading.Lock()

    def acquire(self) -> str | None:
        """轮询取用，避免总是命中列表头部的同一个 IP。"""
        with self._lock:
            self._recover_locked()
            if not self._available:
                return None
            self._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] = 0

    def mark_fail(self, proxy: str) -> None:
        with self._lock:
            self._fail_count[proxy] += 1
            if self._fail_count[proxy] >= self._fail_threshold:
                if proxy in self._available:
                    self._available.remove(proxy)
                    self._cursor = 0
                self._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] = 0
                self._available.append(proxy)

    @property
    def 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 time

import requests


def fetch_with_policy(
    url: str,
    pool: IPPool,
    factory: SessionFactory,
    stats: FailureStats,
    max_rounds: int = 5,
) -> str | None:
    attempt = 0
    proxy = pool.acquire()

    for _ in range(max_rounds):
        if proxy is None:
            stats.record("pool_exhausted")
            return None

        session = 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 = None

        stats.record(category)

        if category == "ok":
            pool.mark_success(proxy)
            return resp.text

        action = decide(category, attempt)
        attempt += 1

        if action is Action.SKIP:
            return None
        if action is Action.DELAY_RETRY:
            time.sleep(retry_after_seconds(resp) if resp is not None else 60.0)
            continue
        if action is Action.RETRY_SAME:
            time.sleep(backoff(attempt))
            continue
        if action is Action.RETRY_SWITCH:
            pool.mark_fail(proxy)
            factory.drop(proxy)
            proxy = pool.acquire()
            time.sleep(backoff(attempt, base=0.5))
            continue

    return None

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

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

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

具体数值与目标站点、代理形态、IP 数量强相关，建议直接跑自己的基线数据，不要照搬别人的百分比。

无效请求率归零是不可能的：目标站点的瞬时故障、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 步直接改代码，最后往往说不清是哪一步起了作用。

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

环境与前提

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

1. 入池前的连通性预检

2. 超时分层

3. 按失败类别分级重试

4. Session 与连接池复用

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

6. 失败 IP 退池与冷却

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

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

超标时的回查顺序

小结

热门文章

最新文章

相关电子书

探索云世界

热门

云计算

大数据

云原生

人工智能

数据库

开发与运维

活动广场

任务中心

训练营

直播

乘风者计划

下载

镜像站

技术资料

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

环境与前提

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

1. 入池前的连通性预检

2. 超时分层

3. 按失败类别分级重试

4. Session 与连接池复用

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

6. 失败 IP 退池与冷却

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

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

超标时的回查顺序

小结

热门文章

最新文章

相关电子书