大多数「突然不能用了」的 Python 爬虫，其实从一开始就没真正跑通过——它们一直在拿到状态码 200、内容却是验证页的响应，直到三周后解析出来的数据集是空的，才有人发现。轮换代理确实解决了一个真实问题，但前提是：外围代码得分得清「一个页面」和「一次拒绝」的区别。
这篇文章先讲机制——怎么把轮换代理接入 requests、httpx 和 Scrapy——然后讲大多数教程会跳过的部分：什么时候该轮换、怎么给响应分类，以及如何让流量账单和你真正采到的数据成正比。
轮换到底买到了什么
一个想限制自动化访问的目标站，手上有若干个成本很低的信号。IP 地址是它最先拿来用的那个，因为在客户端发出任何一个应用层字节之前，IP 是唯一存在的标识。限速、信誉查询、地域规则，全都挂在它上面。
轮换出口地址能破解其中最朴素的一种：按 IP 计数请求。但它对 TLS 指纹、请求头顺序、Cookie 状态和行为时序毫无作用。如果你的爬虫之所以被封，是因为它自报家门是 python-requests/2.31.0、每秒抓 40 个页面、还不带 Referer，那么加上住宅代理只会让这个问题变得更贵，而不会让它消失。
轮换是其中一层，就该把它当成一层来看。
凭据格式
各家服务商暴露轮换能力的方式略有差异，但通行做法是把会话和定位参数编码进代理用户名。UUIProxy 的住宅出口长这样：
http://USERNAME-country-us-session-abc123:[email protected]:7000这串字符里发生了三件事：
country-us把出口固定在美国。去掉它就是整个 IP 池。session-abc123让携带这个标识的所有请求复用同一个出口 IP，最长 120 分钟。去掉它则每个请求都换新 IP。- 主机和端口是网关地址，始终不变。
轮换的全部 API 面就是这些。下面所有内容，都是在讨论这串字符里该填什么、什么时候填。
requests 逐请求轮换
最简单可用的配置是完全省略会话标识，网关会为每次调用分配一个新出口：
import os
import requests
PROXY = (
f"http://{os.environ['PROXY_USER']}-country-us:"
f"{os.environ['PROXY_PASS']}@gate.uuipproxy.com:7000"
)
proxies = {"http": PROXY, "https": PROXY}
response = requests.get(
"https://example.com/catalogue",
proxies=proxies,
timeout=30,
headers={"Accept-Language": "en-US,en;q=0.9"},
)
print(response.status_code, len(response.content))有两个细节比看上去重要得多。
永远设置 timeout。 不设的话 requests 会无限等待，一个卡住的住宅出口就能把一个工作线程永久占死。代理场景下 30 秒已经很宽裕，10 秒通常就够。
让 Accept-Language 与出口国家一致。 一个法兰克福的 IP 索要 en-US 内容是明显的不一致，修起来零成本，不修则极易被识别。
用 requests.Session 做粘性会话
当一个业务流程跨越多个页面——搜索、列表、详情——你会希望它们全部来自同一个地址。为每个逻辑会话生成一个标识并复用：
import os
import uuid
import requests
def sticky_session(country: str = "us") -> requests.Session:
token = uuid.uuid4().hex[:12]
proxy = (
f"http://{os.environ['PROXY_USER']}-country-{country}-session-{token}:"
f"{os.environ['PROXY_PASS']}@gate.uuipproxy.com:7000"
)
session = requests.Session()
session.proxies = {"http": proxy, "https": proxy}
session.headers.update({
"Accept-Language": "en-US,en;q=0.9",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
})
return session
with sticky_session() as session:
session.get("https://example.com/search?q=widgets", timeout=30)
detail = session.get("https://example.com/product/1234", timeout=30)requests.Session 还会在这些调用之间保持 Cookie——这正是「看起来像一个访客而不是三个互不相干的访客」的另一半。
用 httpx 做异步轮换
一旦每分钟请求量超过几百，同步版本就不再是你该关心的瓶颈了。httpx 提供了相同的代理语义，并且支持 asyncio：
import asyncio
import os
import uuid
import httpx
BASE = "gate.uuipproxy.com:7000"
def proxy_url(country: str = "us") -> str:
token = uuid.uuid4().hex[:12]
user = f"{os.environ['PROXY_USER']}-country-{country}-session-{token}"
return f"http://{user}:{os.environ['PROXY_PASS']}@{BASE}"
async def fetch(url: str, country: str = "us") -> httpx.Response:
async with httpx.AsyncClient(
proxy=proxy_url(country),
timeout=httpx.Timeout(30.0, connect=10.0),
follow_redirects=True,
) as client:
return await client.get(url)
async def main(urls: list[str]) -> None:
semaphore = asyncio.Semaphore(20)
async def bounded(url: str):
async with semaphore:
return await fetch(url)
results = await asyncio.gather(*(bounded(url) for url in urls))
for url, response in zip(urls, results):
print(url, response.status_code)
asyncio.run(main(["https://example.com/a", "https://example.com/b"]))注意那个信号量。代理侧并发不计费，但目标站点有它自己能从容承载的上限，超过这个上限是把一个能跑的爬虫变成被封爬虫的最快方式。
Scrapy 集成
Scrapy 内置的 HttpProxyMiddleware 读取 request.meta["proxy"]，所以轮换就是一个五行的中间件：
# middlewares.py
import os
import uuid
class RotatingProxyMiddleware:
def __init__(self):
self.user = os.environ["PROXY_USER"]
self.password = os.environ["PROXY_PASS"]
def process_request(self, request, spider):
country = request.meta.get("proxy_country", "us")
token = request.meta.get("proxy_session") or uuid.uuid4().hex[:12]
user = f"{self.user}-country-{country}-session-{token}"
request.meta["proxy"] = f"http://{user}:{self.password}@gate.uuipproxy.com:7000"把它注册在内置中间件之前：
# settings.py
DOWNLOADER_MIDDLEWARES = {
"myproject.middlewares.RotatingProxyMiddleware": 340,
"scrapy.downloadermiddlewares.httpproxy.HttpProxyMiddleware": 750,
}
RETRY_TIMES = 3
RETRY_HTTP_CODES = [403, 407, 408, 429, 500, 502, 503, 504]
DOWNLOAD_TIMEOUT = 30
CONCURRENT_REQUESTS_PER_DOMAIN = 8在 request.meta 中设置 proxy_session，可以让单个 spider 把某个多步骤流程钉在同一个出口上，而其余请求继续轮换。
对响应分类，而不是对状态码分类
这是「能用的爬虫」和「看起来能用的爬虫」之间的分水岭。
验证页、空结果集、被截断的列表，全都带着 HTTP 200 返回。如果你的重试逻辑只看 response.status_code，这些全部会被记为成功，并作为缺失数据写进你的数据集。
BLOCK_MARKERS = (
"captcha",
"are you a human",
"access denied",
"unusual traffic",
"请完成验证",
)
def classify(response) -> str:
if response.status_code in (403, 407, 429):
return "blocked"
if response.status_code >= 500:
return "server_error"
if response.status_code != 200:
return "unexpected"
body = response.text.lower()
if any(marker in body for marker in BLOCK_MARKERS):
return "challenged"
if len(response.content) < 2048:
return "suspicious_short"
return "ok"长度检查能抓住最常见的静默失败：一个在浏览器里渲染正常、但完全不包含解析器所需数据的骨架页。这个阈值要用你自己目标站的已知正常响应来校准。
升级出口，而不是原地反复重试
响应分类之后，重试就从一个循环变成了一个路由决策。持续产出最低「单条数据成本」的模式是这样的：
- 请求先走数据中心代理，这是你手上最便宜的出口。
- 对响应分类。结果是
ok就结束——绝大多数请求都会是。 - 结果是
blocked或challenged，重新排队到住宅出口。 - 住宅也失败，再升级一次到动态移动代理，并在这一层封顶。
TIERS = ["datacenter", "residential", "mobile"]
def fetch_with_escalation(url: str, tier_index: int = 0):
if tier_index >= len(TIERS):
return None
response = fetch_via(url, tier=TIERS[tier_index])
if classify(response) == "ok":
return response
return fetch_with_escalation(url, tier_index + 1)在典型的电商目标上，需要离开第一层的请求通常不到五分之一。与「全部走住宅代理」相比，月账单一般能差出三到四倍。关于两类出口的取舍，可以看住宅代理与数据中心代理该怎么选。
让出口与请求相互匹配
只有在每个出口与它承载的请求内部一致时，轮换 IP 池才有意义。三类不匹配造成了大多数本可避免的失败：
地理位置与语言不符。 意大利住宅 IP 请求 Accept-Language: en-US 是明显异常。请从你固定的出口国家推导这个请求头。
地理位置与内容不符。 用巴西出口请求某平台的德国站点，往往会直接返回巴西站点，而你的解析器会不声不响地记下错误的价格。
会话与 Cookie 不符。 如果你轮换了出口 IP 却保留了 Cookie，等于告诉目标站：一个登录身份在会话中途跨越了大洲。要么一起换，要么都别换。
关于 IP 之外的其他信号，可以看爬虫被封 IP 怎么办：请求指纹实战排查指南。
按主机限速
克制不只是一种道德立场，它还是成本最低的防封手段。按主机名做令牌桶，可以避免单个 worker 的突发流量把整个采集任务拖进限速区间：
import asyncio
import time
from collections import defaultdict
class HostLimiter:
def __init__(self, per_second: float = 4.0):
self.interval = 1.0 / per_second
self.next_slot = defaultdict(float)
self.lock = asyncio.Lock()
async def acquire(self, host: str) -> None:
async with self.lock:
now = time.monotonic()
wait = max(0.0, self.next_slot[host] - now)
self.next_slot[host] = max(now, self.next_slot[host]) + self.interval
if wait:
await asyncio.sleep(wait)从每主机每秒 4 个请求起步，再根据目标站自身的响应时间调整。如果并发上去之后延迟明显上升，说明已经超出了对方能从容承载的范围。
上量之前的检查清单
把轮换代理方案投入生产负载之前，先过一遍：
- 每个请求都有明确的超时设置，且连接超时短于读取超时。
- 响应按内容分类，而不只是看状态码。
- 重试是在出口层级之间升级，而不是在同一层重复。
- 凡是跨多个页面的流程都使用了粘性会话。
Accept-Language与依赖时区的参数与出口国家一致。- 并发按主机名设上限，而不只是全局上限。
- 按目标域名统计流量用量，让昂贵的站点在账单到来之前就被看见。
- 遵守目标站的
robots.txt指令，只采集可公开访问的页面。
光是前两条，就能解决我们在技术支持里见到的大部分「代理不管用了」的反馈。轮换本身很简单，判断它是否真的生效，才是实际的工程工作。
排查「静默 200」
最贵的失败不是 403，而是状态码 200、正文却不是你要的页面。验证页、地区选择页、「请启用 JavaScript」的空壳、被截断的列表，全都共享同一个状态码。把每条响应正文的前 512 字节，连同 URL、出口国家、会话标识和耗时一起记下来。解析器开始吐空字段时，先搜这份日志，而不是先加购流量。
一个能用的做法是按类别保留滚动样本，而不是存下每一份正文。每种分类（ok、challenge、empty、wrong_locale、truncated）各留 50 个例子，有新失败再轮换。这份归档通常足够判断：拦截率上升是因为出现了新的验证页模板，还是只是更嘈杂的一个小时。
如果肉眼都分不清这些类别，说明分类器还不存在。规则要对照这些样本来写，而不是对照你想象中目标站「大概会返回什么」。
能查得动的日志
四十个 worker 的采集任务里，print 会消失。每条请求打一行固定字段的 JSON：
url、host、status、bytes、elapsed_msexit_country、session_id、tier（datacenter/residential/mobile）- 分类器给出的
class - 若发生升级，记录
retry_count和final_tier
把这些行送到你现成的采集链路即可，标准输出进收集器就够用。一周之后你真正会问的问题永远是这几个：哪些主机吃掉最多 GB、某次发布之后哪种类别在涨、住宅流量占比是不是在慢慢爬升。这些都无法从堆栈跟踪里读出来。
不要记录凭据、完整 Cookie 罐，也不要记录含账号数据的请求体。网关用户名里已经编码了国家和会话，那就是你需要的标识。
用进程，不要用线程
CPython 的 GIL 让线程池不适合「TLS 握手 + HTML 解析」这种混合负载。按 CPU 核心数起进程，每个进程自带 httpx.AsyncClient（或等价物）和自己的 HostLimiter。进程之间只共享 URL 队列和分类结果的出口，别的都不要共享。
如果必须在进程间共享登录 Cookie——也就是带登录态的抓取——把这条工作流钉在单个进程和静态出口上。把一个登录态拆到多个 worker，是账号被挑战的典型路径。其余任务都可以铺开。
一台 16 核机器跑 16 个异步 worker、每主机每秒 4 个请求，会在打满代理网关之前先打满绝大多数「有礼貌」的目标站。你应该盯着的瓶颈是限速器，不是硬件。
关于作者
UUIProxy 解决方案架构师
周美琳与亚太地区的增长团队和研究团队合作，负责账号基础设施与多市场数据采集方案。UUIProxy 的大部分入门资料出自她手，她花了很多时间解释一件事：被封禁时，答案很少是「买更贵的代理」。
