爬虫代理池的四个关键设计（附 Python 实现）

我第一个采集项目就是这么写崩的：每个请求前调一次提取接口拿一条 IP，用完就丢。任务量上来以后，接口开始返回 status=406 提取频率过快，采集程序整片整片地失败。当天晚上把逻辑改成「批量提取 + 本地池子」，接口调用量直接降了一个数量级，采集成功率也从 70% 多回到 99%。

下面这四个点，是我后来在好几个项目里反复验证过的。

一、别循环单条提取

提取接口基本都有频率限制，这是合理的——它是给你"取资源"用的，不是给你当每请求一次的中间件。单条提取的另一个问题是延迟长：每个采集请求前面都挂一次 10~25ms 的接口往返，QPS 直接被打下来。

经验做法：一次取 10~20 条放进本地队列，队列见底前再补一批。接口调用频率从"每请求一次"降到"每几十个请求一次"。

二、低水位补货：什么时候补、补多少

常见写法是两个阈值：

• 高水位（比如 30 条）：池子上限，别无限囤，短效 IP 放着也会过期。
• 低水位（比如 5 条）：低于它就触发补货。

低水位不能设成 1。因为补货本身要走一次网络，如果等到只剩 1 条才补，补货的那几百毫秒里池子是空的，采集线程就会阻塞或者抛异常。低水位的作用就是留出这段缓冲。

一个简单的估算：低水位 ≈ 采集并发数 × 2。10 个采集线程，低水位设 20 左右比较稳。

三、失效怎么剔除：expire 还是本地计时

短效 IP 的生命周期只有几分钟，剔除不及时就会出现"池子里全是死 IP"。两条路都可用：

方案	优点	注意
用接口返回的 expire	最准，跟服务端一致	注意时区，接口给的是本地时间
本地按时间戳淘汰	简单，不依赖字段格式	要设得比真实有效期短一点

我一般两个都上，取较小值：expire_ts = min(取回时间 + 本地上限, 服务端 expire)。本地上限设成 170 秒——短效 IP 常见有效期是 3 分钟，留 10 秒余量，免得刚好卡在过期那一秒。

四、单条 IP 给多少并发

这是最容易被忽略的一点。很多人以为"挂了代理就能随便并发"，结果目标站按 IP 限速，照样 403。我的经验值：

• 单条 IP 并发 2~5，每秒请求不超过 3 次；
• 要更高吞吐，就增加 IP 数量，而不是压榨单条 IP；
• 对同一站点，整体并发最终还是要看对方能承受多少，这一点在限速与合规那篇里展开说过。

可直接用的 Python 实现

下面这个池子做了三件事：线程安全取用、低水位自动补货、按 expire 淘汰。

import threading
import time

import requests

API = "http://api.xydaili.net:2022/tools/ip.ashx"


class ProxyPool(object):
    def __init__(self, order, qty=20, low_watermark=5,
                 area="", isp="", timeout=10, max_age=170):
        self.order = order
        self.qty = qty
        self.low_watermark = low_watermark
        self.area = area
        self.isp = isp
        self.timeout = timeout
        self.max_age = max_age

        self._items = []          # [(proxy_url, expire_ts)]
        self._lock = threading.Lock()

    def _fetch(self):
        params = {
            "action": "GetIP",
            "OrderNumber": self.order,
            "protocol": 1,
            "qty": self.qty,
            "split": "json",
        }
        if self.area:
            params["Area"] = self.area
        if self.isp:
            params["Isp"] = self.isp

        # 取 IP 必须直连：trust_env=False 让它忽略环境变量里的 http_proxy
        session = requests.Session()
        session.trust_env = False
        try:
            resp = session.get(API, params=params, timeout=self.timeout)
        finally:
            session.close()

        data = resp.json()
        if data.get("status") != 200 or not data.get("data"):
            raise RuntimeError("提取失败: %s" % data.get("msg"))

        now = time.time()
        fresh = []
        for item in data["data"]:
            proxy_url = "http://%s:%s" % (item["ip"], item["port"])
            expire_ts = now + self.max_age
            try:  # 接口返回形如 2026-09-27 10:26:31，能解析就用它，更准
                text = str(item["expire"])[:19].replace("T", " ")
                expire_ts = min(expire_ts, time.mktime(
                    time.strptime(text, "%Y-%m-%d %H:%M:%S")))
            except Exception:
                pass
            fresh.append((proxy_url, expire_ts))
        return fresh

    def get(self):
        """取一条可用代理；池子见底时自动补货。"""
        with self._lock:
            now = time.time()
            self._items = [it for it in self._items if it[1] > now]
            if len(self._items) < self.low_watermark:
                self._items.extend(self._fetch())
            return self._items.pop(0)[0]

    @property
    def size(self):
        with self._lock:
            now = time.time()
            return len([1 for _, exp in self._items if exp > now])

用法很直白：

pool = ProxyPool(order="你的订单号", qty=20, low_watermark=5)

for url in urls:
    proxy = pool.get()
    try:
        r = requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=10)
    except requests.RequestException as exc:
        # 短效 IP 失效很正常，换下一条继续，别让一次超时打断整批任务
        print("换 IP:", exc.__class__.__name__)

几个容易忽略的细节

• 取 IP 的请求别走代理。这一条踩过的人最多，容器里配了 HTTP_PROXY 环境变量，取 IP 也走代理，代理一挂整条链路就死了。
• 补货失败不要空转重试。接口报错（订单过期、频率快）时死循环重试只会让情况更糟，记日志 + 退避更靠谱。
• 把用量记下来。接口返回里的已用量字段建议打到监控里，不然跑着跑着额度用完了才发现。
• 多进程要共享池子。上面这个类只在线程内有效，多进程场景得把队列挪到 Redis 里。

代码我整理成了可直接运行的版本，除了 Python 还覆盖 Go、Node.js、Java、C# 等 18 种语言，都放在 github-xydaili-examples 里。我用的线路来自 星月代理，覆盖 300 多个城市、短效 IP 自动轮换，白名单和账号密码两种模式都支持，上面这些参数就是按它的接口字段写的。