httpx需手动实现代理轮换与失败重试:准备合法代理列表,捕获ConnectError/TimeoutException等异常,配合指数退避与最大重试次数切换代理;异步版本逻辑相同,需用async/await;进阶可缓存代理健康状态并动态剔除失效代理。

httpx 本身不内置代理轮换或失败重试逻辑,但可以通过封装请求、捕获异常、配合代理池实现“连接失败自动换代理”。关键在于:手动管理代理列表、捕获 httpx.ConnectError / httpx.TimeoutException 等网络异常,并在失败后切换下一个可用代理重新发起请求。
准备可用代理列表
确保你有一组格式合法、可验证的 HTTP/SOCKS 代理(如 "http://user:pass@host:port" 或 "socks5://host:port")。建议预先清洗或通过简单 HEAD 请求验证部分代理可用性,避免全量盲试。
- 代理格式必须与 httpx 的
proxies参数兼容(支持http://、https://、socks5://) - 若用认证代理,URL 中需包含用户名密码(httpx 0.24+ 支持,旧版需额外配置
auth) - 可将代理存为 list,用
itertools.cycle或索引轮询,避免重复使用失效代理
封装带代理切换的请求函数
核心是 try-except 捕获连接类异常,在失败时更新当前代理并重试。推荐使用指数退避 + 最大重试次数防止死循环:
import httpx import time from itertools import cycleproxies = [ "https://www.php.cn/link/dfcfbb196720bb3febac626b8b9d082d", "https://www.php.cn/link/07f2d8dbef3b2aeca9cb258091bc3dba", "socks5://proxy3:1080" ]
proxy_cycle = cycle(proxies) client = httpx.Client(timeout=10.0)
def request_with_failover(url, max_retries=3): last_exc = None for i in range(max_retries): proxy = next(proxy_cycle) try: r = client.get(url, proxies={"http://": proxy, "https://": proxy}) r.raise_for_status() return r except (httpx.ConnectError, httpx.TimeoutException, httpx.ReadError) as e: last_exc = e if i < max_retries - 1: time.sleep(0.5 * (2 ** i)) # 指数退避 raise last_exc
使用异步 client(httpx.AsyncClient)时的等效写法
异步版本逻辑一致,只需用 async/await 和捕获对应异常(注意 httpx.ConnectTimeout、httpx.ReadTimeout 等更细粒度异常):
立即学习“Python免费学习笔记(深入)”;
import asyncio import httpxasync def async_request_with_failover(url, proxies, max_retries=3): for i in range(max_retries): proxy = proxies[i % len(proxies)] try: async with httpx.AsyncClient(proxies={"http://": proxy, "https://": proxy}, timeout=10.0) as client: r = await client.get(url) r.raise_for_status() return r except (httpx.ConnectError, httpx.TimeoutException) as e: if i == max_retries - 1: raise e await asyncio.sleep(0.5 * (2 ** i)) raise RuntimeError("All proxies failed")
进阶建议:代理健康状态缓存与动态剔除
真实场景中,不应每次失败都轮到下一个——而应标记已知失效代理,跳过它们,提升成功率:










