Python 代理轮换实战：requests、httpx 与 aiohttp（2026）

只要你用 Python 做稍微有规模的抓取或自动化，早晚会遇到：目标站发现所有请求都来自同一个 IP，然后把你封掉。Python 代理轮换就是把流量分散到大量 IP 上，让任何单个 IP 都不会触发限速或封禁。本文给出 requests、httpx、aiohttp 的完整代码，并讲清何时选轮换网关、何时自建 IP 池。

两个关键决策贯穿全文。其一：轮换 vs sticky 会话 —— 每次请求换新 IP，还是整个多步流程用同一个 IP。其二：住宅 vs 数据中心 IP —— 参见代理类型决策表。这两个选对了，代码反而是最简单的部分。

两种轮换方式

1. 网关轮换（推荐）。你只连接一个固定入口，供应商在每次新连接时给你一个新的出口 IP。无需维护列表，也不用清理死 IP。极豹的动态住宅池就是这种方式：同一个主机和端口，每次请求一个新 IP。

2. 自管 IP 列表。你手持一份 host:port 列表，每次请求选一个。控制力更强，但健康检查、重试、封禁跟踪都得你自己管。需要固定、稳定地址时，配合独享数据中心 IP 使用。

用 requests 轮换

最简单的场景 —— 把一个请求路由到一个会自动轮换的网关。用 socks5h://（末尾的 h 表示 DNS 在代理端解析，避免泄露查询）：

import requests

# 极豹轮换住宅网关：每次连接一个新出口 IP
PROXY = "socks5h://USERNAME:[email protected]:10001"

proxies = {"http": PROXY, "https": PROXY}

for _ in range(5):
    r = requests.get("https://httpbin.org/ip", proxies=proxies, timeout=30)
    print(r.json()["origin"])   # 每次循环 IP 都不同

要让 requests 支持 SOCKS5，装一下额外依赖：pip install "requests[socks]"。偏好 HTTP 网关的话，把 scheme 换成 http:// 即可，其余不变。

从自建 IP 池轮换

import random
import requests

# 同一网关 + 用户名参数做国家定向
PROXY_POOL = [
    "socks5h://USERNAME-country-us:[email protected]:10001",
    "socks5h://USERNAME-country-uk:[email protected]:10001",
    "socks5h://USERNAME-country-de:[email protected]:10001",
]

def fetch(url):
    proxy = random.choice(PROXY_POOL)
    return requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=30)

resp = fetch("https://example.com")
print(resp.status_code)

必须保持同一 IP 时用 sticky 会话

登录、购物车、分页结果这些流程，一旦 IP 中途改变就会断。在用户名里加一个会话 token 来锁定 IP，复用同一 token 即可在其生命周期内保持同一出口 IP：

import requests

SESSION_ID = "task-0001"
PROXY = f"socks5h://USERNAME-session-{SESSION_ID}:[email protected]:10001"

s = requests.Session()
s.proxies = {"http": PROXY, "https": PROXY}

s.post("https://example.com/login", data={"u": "me", "p": "secret"})
s.get("https://example.com/dashboard")   # 同一 IP，会话不断

用 httpx 轮换

httpx 是现代化客户端，支持 HTTP/2 和原生异步。注意新版本用单个 proxy= 参数：

import httpx

PROXY = "socks5h://USERNAME:[email protected]:10001"

with httpx.Client(proxy=PROXY, timeout=30) as client:
    for _ in range(5):
        print(client.get("https://httpbin.org/ip").json()["origin"])

httpx 的 SOCKS 支持需 pip install "httpx[socks]"。加上 http2=True 让流量更像真实浏览器，与 TLS 指纹匹配 配合效果更好。

用 aiohttp 并发轮换

追求高吞吐时，并发发起大量请求。每条经网关的连接都拿到自己的出口 IP，所以并发与轮换是同时发生的：

import asyncio
import aiohttp
from aiohttp_socks import ProxyConnector

PROXY = "socks5://USERNAME:[email protected]:10001"

async def fetch(url):
    connector = ProxyConnector.from_url(PROXY)
    async with aiohttp.ClientSession(connector=connector) as session:
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as r:
            return await r.text()

async def main():
    urls = ["https://httpbin.org/ip"] * 20
    results = await asyncio.gather(*(fetch(u) for u in urls))
    print(f"共抓取 {len(results)} 页")

asyncio.run(main())

用 pip install aiohttp_socks 装 SOCKS 连接器。每个任务新建一个连接器，让每条请求都开一个新的代理连接。

真正有用的重试逻辑

只有在新 IP 上重试失败请求，轮换才有意义。指数退避，并在每次重试时轮换会话 token：

import requests
from time import sleep

def fetch_with_retry(url, max_retries=4):
    for attempt in range(max_retries):
        proxy = f"socks5h://USERNAME-session-r{attempt}:[email protected]:10001"
        try:
            r = requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=30)
            if r.status_code in (403, 429):
                raise requests.HTTPError(f"被拦: {r.status_code}")
            r.raise_for_status()
            return r.text
        except requests.RequestException:
            sleep(2 ** attempt)        # 1s, 2s, 4s, 8s
    raise RuntimeError(f"重试 {max_retries} 次仍失败: {url}")

验证轮换是否生效

大范围跑之前，先确认 IP 确实在变。连着打几次 echo 接口，看返回的 IP 是否不同。要同时确认你的 TLS 指纹没有出卖你，用极豹免费的 JA3/TLS 指纹检测工具 跑一下 —— IP 再干净，配上一眼就能认出的 Python 指纹，照样被封。

最佳实践

• 按任务轮换，不是按步骤。任何多请求流程用 sticky 会话；独立任务之间才轮换。
• IP 类型匹配目标。宽松的 API 用数据中心（$1/GB），Cloudflare/Akamai 后面的站用住宅（$6.8/GB）。
• 每个请求都设超时。一个挂死的代理连接没超时，会拖垮整个抓取。
• 盯住流量。住宅代理按 GB 计费 —— 重定向死循环抓个大页面会快速烧掉额度。
• 尊重 robots.txt 和限速。猛打目标会让整段 IP 被标记，害人害己。

轮换稳了之后，你撞的下一面墙通常是指纹而不是 IP。如果干净的住宅 IP 仍然 403，读读如何用 curl_cffi 绕过 TLS 指纹，以及完整的 Cloudflare 绕过配方。