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 捕获连接类异常,在失败时更新当前代理并重试。推荐使用指数退避 + 最大重试次数防止死循环:
<font color="#888">import httpx
import time
from itertools import cycle
<p>proxies = [
"<a href="https://www.php.cn/link/dfcfbb196720bb3febac626b8b9d082d">https://www.php.cn/link/dfcfbb196720bb3febac626b8b9d082d</a>",
"<a href="https://www.php.cn/link/07f2d8dbef3b2aeca9cb258091bc3dba">https://www.php.cn/link/07f2d8dbef3b2aeca9cb258091bc3dba</a>",
"socks5://proxy3:1080"
]</p><p>proxy_cycle = cycle(proxies)
client = httpx.Client(timeout=10.0)</p><p>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</font>使用异步 client(httpx.AsyncClient)时的等效写法
异步版本逻辑一致,只需用 async/await 和捕获对应异常(注意 httpx.ConnectTimeout、httpx.ReadTimeout 等更细粒度异常):
立即学习“Python免费学习笔记(深入)”;
<font color="#888">import asyncio
import httpx
<p>async 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")</font>进阶建议:代理健康状态缓存与动态剔除
真实场景中,不应每次失败都轮到下一个——而应标记已知失效代理,跳过它们,提升成功率:









