短效代理的成败，一半在使用方式上

国内短效代理的核心特征是 IP 生命周期短、可用池持续轮换。同一份套餐，有人成功率稳定在九成以上，有人长期在五成附近徘徊，差别通常不在 IP 本身，而在接入层是否做对了三件事：连接复用、按请求换 IP、失败重试与限速。本文给出三种典型接入写法，可直接对照自己的代码调整。

一、基础模式：Session 复用 + 单次换 IP

最常用也最容易写错的是直接用 requests.get(proxies=...) 每次新建会话。短效代理的握手成本占比很高，逐请求新建连接会把大部分时间消耗在建立链路上，表现为"能通但很慢"。正确做法是复用 Session 连接池，仅在需要换出口时切换代理。

import requests

# 会话复用：同一个 Session 内保持连接池，减少握手开销
session = requests.Session()
session.headers.update({"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"})

def fetch(url, proxy_line):
    # proxy_line 形如 1.2.3.4:8888，或 用户:密码@1.2.3.4:8888
    proxy = proxy_line if "@" in proxy_line else f"tester:tester@{proxy_line}"
    proxies = {"http": f"http://{proxy}", "https": f"http://{proxy}"}
    try:
        r = session.get(url, proxies=proxies, timeout=(5, 15))
        return r.status_code, r.text[:120]
    except Exception as e:
        return None, str(e)

要点有三个：一是代理串格式统一，账密认证要拼在小写 http:// 之后，很多人写成 socks5 导致请求超时；二是超时时间必须显式设置，短效 IP 失效往往表现为长时间挂起而不是立即报错；三是异常要吞掉并记录，单次失败不代表链路有问题，交给上层重试即可。

二、Scrapy 中间件：让每次请求自动换 IP

采集型项目多数跑在 Scrapy 上。短效代理适合放在下载中间件的 process_request 里，每个请求前从提取接口拿一条新 IP，写进 request.meta["proxy"]。

from scrapy import signals

class ShortProxyMiddleware:
    """每次请求前从提取链路换一个新 IP"""

    def __init__(self, extractor):
        self.extractor = extractor

    @classmethod
    def from_crawler(cls, crawler):
        mw = cls(crawler.settings.get("PROXY_EXTRACTOR"))
        crawler.signals.connect(mw.spider_opened, signal=signals.spider_opened)
        return mw

    def spider_opened(self, spider):
        spider.logger.info("short proxy middleware ready")

    def process_request(self, request, spider):
        ip = self.extractor.get_one()          # 返回 1.2.3.4:8888
        if not ip:
            return None                        # 没取到就直连或由重试逻辑接管
        request.meta["proxy"] = f"http://{ip}"
        request.meta["download_timeout"] = 15
        return None

这里有两个细节值得注意。第一，取不到 IP 时不要抛出异常，返回 None 让请求继续（或用直连兜底），否则会让整个调度器陷入重试风暴。第二，download_timeout 建议明显小于默认值，短效 IP 一旦失效，快速失败再换一条，永远比等满超时更划算。

如果提取接口本身有限速，建议在中间件内部做一个长度可控的本地队列，一次取 5 到 10 条放进队列，用完再取，避免每个请求都打一次接口。

三、并发池：控制线程数与重试次数

脚本化任务常用线程池。并发数与重试次数是两个必须显式设定的参数，放任默认值最容易出问题。

from concurrent.futures import ThreadPoolExecutor, as_completed

def run(urls, extractor, workers=8, retry=2):
    result = {"ok": 0, "fail": 0}
    with ThreadPoolExecutor(max_workers=workers) as pool:
        tasks = {pool.submit(fetch_with_new_ip, u, extractor, retry): u for u in urls}
        for fut in as_completed(tasks):
            ok, _ = fut.result()
            result["ok" if ok else "fail"] += 1
    return result

def fetch_with_new_ip(url, extractor, retry):
    for attempt in range(retry + 1):
        ip = extractor.get_one()
        if not ip:
            continue
        status, body = fetch(url, ip)
        if status == 200 and "验证码" not in body:
            return True, status
    return False, None

经验值：短效代理的并发不要一上来就拉到 50，先用 8 到 12 跑十分钟，观察失败率再逐步上调；重试次数控制在 3 次以内，超过这个量级说明问题在目标站的风控或 IP 池质量，而不是运气。另外，判成功不要只看状态码 200，正文里出现验证码、登录页、限流提示都应视为失败，否则统计数据会严重失真。

三个高频踩坑

  • 把短效当长效用：同一条 IP 连续请求同一目标几十次，等同于主动触发风控。短效代理的用法是"一次请求一条 IP"，长连接复用只针对 Session 对象，不是针对出口 IP。
  • 白名单与账密混用：同时开了白名单又传了账密，部分链路会直接拒绝。二选一，并在代码里保持配置单一来源。
  • 忽略地区参数：不传地区时取到的是全国随机池，命中异地节点会拉高延迟。目标站在哪个城市，就把提取参数限定在哪片区域。
判断接入是否写对，有一个简单指标：失败请求占全部请求的比例，应当与单条 IP 的失效率接近。如果明显偏高，问题在代码层；如果两者接近但都很高，才需要回到 IP 池质量上找原因。

小结

短效代理的接入并不复杂，但需要把"连接复用"和"出口轮换"两件事分开处理：连接尽可能复用，出口尽可能勤换。按这个原则改写上面三段代码中的任意一段，成功率通常会有明显变化。