在价格监测、公开页面采集和站点可用性检查中，代理池的作用不是把请求随机分散出去，而是把代理地址的获取、可用性判断、失败恢复和任务调度管理起来。动态住宅 IP 只是代理池的一种上游资源，真正影响爬虫结果的还有轮换周期、会话一致性、目标站点响应和请求节奏。

不少示例只写了 random.choice(proxies)，却没有处理代理过期、认证失败、429 限流和同一会话中途换出口的问题。短脚本可以运行，任务一旦持续数小时，就会出现大量重复失败、空数据和无法解释的日志。

本文用 Python Requests 搭建一个可运行的最小框架，重点解决三个问题：动态住宅 IP 如何接入、代理节点如何冷却恢复、不同业务流程应该采用什么轮换策略。示例使用供应商无关的接口格式，实际接入时只需调整代理获取接口的字段映射。

一、动态住宅 IP 的两种接入模式

1. 轮换网关

轮换网关对客户端表现为一个固定的 host:port。客户端连接网关后，由网关从上游地址池分配出口。出口可能按请求切换，也可能按连接或会话保持，具体规则由供应商决定。

这种模式下，代码里的代理地址不变，并不等于出口 IP 不变。验证方式应该是：通过同一个测试地址连续发起多次请求，记录响应中的出口 IP，并将结果与供应商的轮换规则对照。Requests 的 Session 会复用连接，若供应商按连接切换，过度复用 Session 可能看不到预期的轮换效果。

2. 代理获取接口

代理获取接口一般返回一条临时代理地址和有效期，例如：

{
  "proxy_url": "http://user:password@host.example.com:8000",
  "expires_in": 300
}

客户端需要在使用前获取租约，在有效期内复用；租约临近过期或节点被判定为不可用时，再获取下一条。这里的“租约”是程序对临时代理地址的管理概念，不代表所有供应商接口都使用这个字段名。

二、轮换策略要和请求关系匹配

动态住宅 IP 不适合简单设置成“每个请求必换”。请求之间是否共享 Cookie、分页游标、重定向状态或业务上下文，决定了轮换边界。

场景	建议策略	原因
独立的公开页面探测	按请求或短批次轮换	请求之间没有会话依赖
列表页到详情页	按任务轮换	保持同一任务的请求上下文
需要 Cookie 的多步流程	粘性会话	中途换出口可能造成状态不一致
长时间任务	按租约有效期续租	避免复用过期地址

粘性会话并不是让一个地址永久不变，而是在服务商允许的时间窗口内保持同一出口。会话结束后应释放或重新获取，不能把一个过期租约继续交给业务层。

三、先统一代理来源，再写业务调度

供应商的接口字段、认证方式和地区参数各不相同，业务代码不应该直接依赖某个供应商的 JSON 结构。可以把上游适配成一个简单的 acquire() 方法，代理池只接收统一的 ProxyLease。

下面的示例约定接口返回 proxy_url 和可选的 expires_in，支持 HTTP/HTTPS 代理。需要使用 SOCKS5 时，应额外安装 requests[socks]，并确认供应商提供的协议和 DNS 解析行为；本文不把 SOCKS5 混入动态住宅 IP 的主流程，避免配置含义混淆。

from __future__ import annotations

import os
from dataclasses import dataclass

import requests


@dataclass(frozen=True)
class ProxyLease:
    url: str
    expires_in: int | None = None


class ProxySource:
    """将供应商接口适配成统一的代理租约。"""

    def __init__(self, api_url: str, token: str | None = None) -> None:
        self.api_url = api_url
        self.token = token

    def acquire(self) -> ProxyLease:
        headers = {"Authorization": f"Bearer {self.token}"} if self.token else {}
        response = requests.get(self.api_url, headers=headers, timeout=(5, 10))
        response.raise_for_status()

        payload = response.json()
        proxy_url = payload.get("proxy_url")
        if not isinstance(proxy_url, str) or not proxy_url.startswith(("http://", "https://")):
            raise ValueError("接口没有返回有效的 HTTP 代理地址")

        expires_in = payload.get("expires_in")
        if expires_in is not None:
            expires_in = int(expires_in)
            if expires_in <= 0:
                raise ValueError("expires_in 必须是正整数")

        return ProxyLease(proxy_url, expires_in)


def source_from_env() -> ProxySource:
    return ProxySource(
        api_url=os.environ["PROXY_API_URL"],
        token=os.getenv("PROXY_API_TOKEN"),
    )

代码没有假设某个供应商的真实接口地址。接入时只需把供应商返回的字段映射到 proxy_url；如果认证不是 Bearer Token，也只修改 headers 的构造方式。代理获取接口返回 200，只代表租约申请成功，不代表该地址已经通过目标站点测试。

四、一个可运行的代理池最小实现

代理池至少要保存节点地址、成功数、失败数、最近延迟、冷却截止时间和当前并发数。失败节点不能立即永久删除：连接抖动可能很快恢复，认证错误则需要更长冷却或重新获取租约。

from __future__ import annotations

import random
import time
from dataclasses import dataclass


@dataclass
class ProxyNode:
    lease: ProxyLease
    successes: int = 0
    failures: int = 0
    last_latency_ms: float | None = None
    cooldown_until: float = 0.0
    last_error: str | None = None
    in_flight: int = 0

    def ready(self, now: float, max_in_flight: int) -> bool:
        return self.cooldown_until <= now and self.in_flight < max_in_flight


class ProxyPool:
    def __init__(self, source: ProxySource, max_in_flight: int = 2) -> None:
        self.source = source
        self.max_in_flight = max_in_flight
        self.nodes: list[ProxyNode] = []

    def _add_node(self) -> ProxyNode:
        node = ProxyNode(self.source.acquire())
        self.nodes.append(node)
        return node

    def acquire(self) -> ProxyNode:
        now = time.time()
        candidates = [
            node for node in self.nodes
            if node.ready(now, self.max_in_flight)
        ]
        if not candidates:
            candidates = [self._add_node()]

        def weight(node: ProxyNode) -> float:
            total = node.successes + node.failures
            failure_rate = node.failures / total if total else 0.0
            latency_penalty = (node.last_latency_ms or 1000.0) / 1000.0
            return max(0.1, 1.0 - failure_rate - latency_penalty * 0.05)

        node = random.choices(candidates, weights=[weight(n) for n in candidates], k=1)[0]
        node.in_flight += 1
        return node

    def success(self, node: ProxyNode, latency_ms: float) -> None:
        node.in_flight = max(0, node.in_flight - 1)
        node.successes += 1
        node.last_latency_ms = latency_ms
        node.last_error = None

    def failure(self, node: ProxyNode, error: str) -> None:
        node.in_flight = max(0, node.in_flight - 1)
        node.failures += 1
        node.last_error = error

        cooldown_seconds = {
            "HTTP_429": 60,
            "HTTP_407": 900,
            "ConnectTimeout": 20,
            "ReadTimeout": 30,
            "ProxyError": 120,
        }.get(error, 15)
        node.cooldown_until = time.time() + cooldown_seconds

这个版本体现了两个关键点：代理池会在没有候选节点时获取新的租约；节点恢复由冷却时间控制，而不是在一次失败后直接删除。实际项目还应该为租约增加过期判断，避免 expires_in 已到期但节点仍留在候选列表中。

可以将 ProxyNode 的过期字段补充为绝对时间：

expires_at = time.time() + lease.expires_in if lease.expires_in else None

随后在 ready() 中加入 expires_at is None or expires_at > now。如果供应商返回的地址很短命，这个判断比单纯依赖 HTTP 错误更及时。

五、请求执行器：把错误分成四类

代理池是否好用，最终要通过真实请求验证。下面的执行器对状态码和网络异常分别处理，返回内容时只把字节数据交给上层，避免把已关闭的 Session 或 Response 流继续向外传递。

from __future__ import annotations

import time

import requests
from requests.exceptions import ConnectTimeout, ConnectionError, ProxyError, ReadTimeout


RETRYABLE_STATUS = {408, 425, 500, 502, 503, 504}


def fetch(pool: ProxyPool, url: str, attempts: int = 3) -> dict[str, object] | None:
    for attempt in range(attempts):
        node = pool.acquire()
        started = time.perf_counter()

        try:
            with requests.Session() as session:
                session.trust_env = False
                response = session.get(
                    url,
                    proxies={"http": node.lease.url, "https": node.lease.url},
                    timeout=(5, 20),
                    headers={"Accept": "text/html,application/xhtml+xml"},
                )
                status_code = response.status_code
                body = response.content
                retry_after = response.headers.get("Retry-After")

            latency_ms = (time.perf_counter() - started) * 1000

            if 200 <= status_code < 300:
                pool.success(node, latency_ms)
                return {"status_code": status_code, "content": body, "latency_ms": latency_ms}

            error = f"HTTP_{status_code}"
            pool.failure(node, error)

            if status_code in {403, 407}:
                return {"status_code": status_code, "content": body, "error": error}

            if status_code == 429:
                wait_seconds = int(retry_after) if retry_after and retry_after.isdigit() else 60
                time.sleep(min(wait_seconds, 120))
            elif status_code not in RETRYABLE_STATUS:
                return {"status_code": status_code, "content": body, "error": error}

        except (ConnectTimeout, ReadTimeout, ProxyError, ConnectionError) as exc:
            pool.failure(node, type(exc).__name__)

        if attempt + 1 < attempts:
            time.sleep(min(2 ** attempt, 8))

    return None

这里的错误含义需要保留到日志中：

• 407 通常指代理认证失败，应该检查账号、密码、端口和租约，而不是盲目重试。
• 403 是目标端拒绝当前请求，可能与权限、请求频率、Cookie 或目标策略有关，不能直接等同于代理失效。
• 429 表示请求频率受到限制，应优先读取 Retry-After，降低并发并等待。
• ConnectTimeout 发生在连接阶段，ReadTimeout 发生在等待响应数据阶段，两者需要分别统计。

requests.Session 在这个执行器中按一次尝试创建并关闭，适合验证网关是否按连接分配出口。若业务流程需要保持 Cookie，应把 Session 的生命周期提升到“任务”级别，并固定一个粘性会话；不能为了轮换出口，在同一个多步流程中随意重建 Session。

六、健康检查不只是访问一个 IP 查询接口

基础检查可以使用项目允许访问的测试 URL，记录状态码和延迟；业务检查则应访问实际允许采集的目标，并验证页面或 JSON 结构。两者要分开保存：

检查层	关注内容	失败时的含义
连接层	连接、TLS、代理认证、读取耗时	代理配置或链路异常
HTTP 层	403、407、429、5xx	目标策略、认证、限流或服务端异常
内容层	标题、字段、JSON 结构	可能拿到错误页、空页或校验页

例如，代理能够访问测试 URL，但目标页面返回 403，不足以证明节点已经失效。反过来，目标返回 200 但页面没有需要的字段，也不能计入业务成功。代理池的成功率最好至少分成“连接成功率”和“有效内容成功率”，否则指标会被错误页面抬高。

延迟统计建议记录 P50、P95 和超时率。平均值容易掩盖少量长尾请求；P95 上升时，即使 P50 没有变化，爬虫的整体完成时间也可能明显变长。对于按任务轮换的场景，还可以记录单个租约的连续成功次数和连续失败次数。

七、轮换和并发设置的实际边界

动态住宅 IP 的数量不能直接换算成可承受并发。单个网关可能有连接上限，供应商可能按账号限制带宽或请求数，目标站点也可能限制单域名并发。因此，调度器应同时控制单节点并发、单域名并发和任务总速率。

Scrapy 的 AutoThrottle 采用根据响应延迟动态调整下载间隔的思路，适用于需要根据站点负载降低请求节奏的任务。Requests 项目也可以采用类似策略：当 P95 延迟、429 比例或超时率上升时，降低并发和请求频率，而不是简单增加节点数量。

建议把这些配置放在环境变量或配置文件中：

MAX_ATTEMPTS=3
MAX_IN_FLIGHT_PER_PROXY=2
CONNECT_TIMEOUT=5
READ_TIMEOUT=20
MIN_DELAY_SECONDS=0.5
COOLDOWN_ON_429=60

不同目标域名可以使用不同的并发和延迟配置。列表页采集、详情页采集和站点监控的请求形态不同，全部共用一套参数，通常会导致某些任务过慢或某些目标收到过密请求。

八、日志字段和排查方法

代理密码、令牌和完整连接字符串不应该写进普通日志。建议使用节点内部 ID 或脱敏后的哈希，并记录：

• task_id、目标域名和请求路径；
• 代理节点内部 ID、租约获取时间和过期时间；
• 状态码、异常类型、耗时和重试次数；
• 当前并发数、冷却截止时间和最终结果；
• 内容校验是否通过。

排查顺序可以按事件位置划分：获取代理接口失败，检查供应商接口和认证；连接阶段失败，检查协议、端口和网络链路；返回 407，检查代理认证；返回 403 或 429，检查目标策略和请求节奏；返回 200 但内容不对，检查解析和页面校验。每类问题的处理动作不同，日志分类越清楚，越不容易把业务问题误判成代理问题。

九、合规和工程边界

代理池只能管理网络出口和请求调度，不能替代访问授权，也不会自动赋予某个账号、接口或数据源访问权限。采集前应确认目标站点的服务条款、公开数据范围、robots.txt 和访问频率要求；需要登录或付费权限的数据，应优先使用官方接口或取得明确授权。

同样不建议把代理轮换写成“规避限制”的万能开关。合理的技术目标应该是隔离测试环境、控制连接质量、降低单节点故障影响，并让采集任务更容易观测和停止。出现连续 403、429 或内容校验失败时，应暂停任务并核对业务权限和访问策略，而不是继续提高轮换速度。

十、总结

一个可维护的 Python 代理池，至少要有四个独立边界：

1. 代理来源层：负责获取轮换网关或临时代理租约。
2. 状态层：保存成功率、延迟、错误类型、冷却时间和租约有效期。
3. 调度层：根据任务关系决定按请求、按任务还是按会话轮换。
4. 业务层：负责请求、内容校验、解析和结果落库。

动态住宅 IP 的接入并不复杂，难点在于把供应商的轮换规则映射到实际请求生命周期。公开页面探测可以使用短批次轮换；多步请求应保持粘性会话；429 要降低节奏；407 要修正认证；连接超时则需要节点冷却和有限重试。按照这套边界设计，后续将 Requests 替换为 aiohttp、httpx 或 Scrapy 时，也不必重写业务层。

参考资料

• [Requests Advanced Usage：Proxies 与 Session](https://requests.readthedocs.io/en/stable/user/advanced/)
• [Requests API：timeout、proxies 与异常类型](https://requests.readthedocs.io/en/stable/api/)
• [Scrapy AutoThrottle 官方文档](https://docs.scrapy.org/en/latest/topics/autothrottle.html)