Python requests 接入住宅代理：会话保持、超时与重试的正确写法

代理 URL 的格式

requests 通过 proxies 参数走代理，格式是 协议://认证信息@主机:端口。住宅代理通常是 HTTP 网关，即使你访问的是 HTTPS 站点，代理本身也用 http://：

import requests

PROXY = "http://用户名:密码@gateway.example.com:8080"

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

resp = requests.get("https://httpbin.org/ip", proxies=proxies, timeout=30)
print(resp.json())

如果密码里包含 @、:、# 这类字符，必须做 URL 编码，否则会把代理地址解析错：

from urllib.parse import quote

user = quote("my@user", safe="")
pwd  = quote("p@ss:w0rd#1", safe="")
PROXY = f"http://{user}:{pwd}@gateway.example.com:8080"

这个问题在本地调试时经常被忽略，因为测试账号的密码往往是简单的字母数字，换个正式账号就报错。

粘性会话：让出口在一段时间内不变

住宅代理默认会轮换出口 IP。需要登录态或分页连续抓取时，必须显式要求粘性会话。做法通常是在用户名后面追加会话标识参数，具体参数名以官方文档为准，各产品线写法不同：

# 典型写法：在用户名后追加 session 参数，值相同则出口相同
SESSION_ID = "abc123xyz"
user = f"用户名-session-{SESSION_ID}"
PROXY = f"http://{user}:密码@gateway.example.com:8080"

同一个 SESSION_ID 在有效期内会映射到同一个出口 IP。需要换 IP 时更换 SESSION_ID 即可，这比等待超时更可控。

验证出口是否真的固定，连续请求两次比对即可：

for i in range(3):
    ip = requests.get("https://httpbin.org/ip", proxies=proxies, timeout=30).json()["origin"]
    print(i, ip)   # 粘性会话下应打印同一个 IP

Session 复用：连接池不是可有可无的优化

直接用 requests.get() 每次都新建连接，包括到代理的 TCP 握手。批量抓取时应该复用 Session，它内部维护连接池：

session = requests.Session()
session.proxies.update(proxies)
session.headers.update({
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
                  "(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36",
    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
})

resp = session.get("https://shop.example.com/item/123", timeout=30)

超时：一个值是不够的

timeout=30 只限制了单次读写的等待，不区分连接阶段和读取阶段。代理环境里连接阶段和读取阶段的问题性质完全不同，应该分开设置：

TIMEOUT = (10, 45)   # (连接超时, 读取超时)
resp = session.get(url, timeout=TIMEOUT)

连接超时通常意味着代理网关不可达或本地网络问题；读取超时意味着出口 IP 到目标站点这一段慢，可能是该 IP 被限速。分开设置能让日志直接指向原因。

带退避的重试

住宅代理偶发的连接失败是正常现象，关键是重试要有退避、有上限，并且在重试时换一个新出口——否则等于对同一个坏 IP 反复撞墙：

import time, random
from requests.exceptions import RequestException

def fetch(url: str, tries: int = 4):
    last = None
    for attempt in range(tries):
        try:
            # 每次重试换一个会话 ID，也就是换一个出口 IP
            sid = f"{random.randint(10**6, 10**7-1)}"
            proxies["http"] = proxies["https"] = build_proxy(sid)
            session.proxies.update(proxies)
            r = session.get(url, timeout=(10, 45))
            if r.status_code in (403, 429):
                raise RequestException(f"blocked {r.status_code}")
            return r
        except RequestException as e:
            last = e
            # 指数退避 + 抖动，避免所有重试同时打过来
            time.sleep(min(2 ** attempt, 15) + random.uniform(0, 1.5))
    raise last

两点值得注意：一是判断条件里包含 403/429，因为这两类状态码代表被风控，重试换 IP 是有意义的；二是加了随机抖动，避免多个线程的重试在同一时刻集中发出。

并发：线程数不是越高越好

requests 是同步库，通常用线程池提升吞吐。但并发过高会让同一个出口 IP 在目标站点那边形成明显的突发流量，反而提前触发限流：

from concurrent.futures import ThreadPoolExecutor, as_completed

with ThreadPoolExecutor(max_workers=8) as pool:
    futures = {pool.submit(fetch, u): u for u in urls}
    for f in as_completed(futures):
        url = futures[f]
        try:
            print(url, len(f.get().content))
        except Exception as e:
            print("FAIL", url, e)

对中等防护的站点，单出口个位数并发是比较稳妥的起点，实际值应该由成功率测试反推，而不是拍脑袋设定。

排查清单

接不上时按这个顺序查，能覆盖绝大多数情况：

1. 密码里的特殊字符是否做了 URL 编码
2. 认证方式是否匹配（IP 白名单模式不需要用户名密码，反之亦然）
3. 代理的端口和协议是否用对（住宅网关常见的是 8080 / 22225 这类端口）
4. 是否真的走了代理——用 httpbin.org/ip 确认出口 IP
5. 本地网络或云服务器安全组是否放行了出站连接
6. 目标站点是否在拦截你，而不是代理本身有问题

第 4 步最容易被跳过，但它能一步区分”代理没生效”和”目标站点拦你”这两类完全不同的问题。