HTTPX不内置代理轮换,但可通过捕获httpx.ConnectError、TimeoutException、ProxyError等异常并手动切换代理实现自动轮换;需准备有效代理列表,推荐异步+重试机制,并可结合健康检查动态剔除失效代理。

HTTPX 本身不内置代理轮换逻辑,但可以通过捕获连接异常、手动切换代理并重试请求来实现“连接失败后自动轮换代理”的机制。核心思路是:封装请求逻辑,当遇到网络错误(如连接超时、拒绝连接、代理不可达)时,从代理列表中取出下一个代理,重新发起请求。
1. 准备可用的代理列表
确保你有一组格式正确、可验证的 HTTP/HTTPS 代理地址,例如:
- http://user:pass@192.168.1.100:8080
- http://192.168.1.101:3128
- https://user:pass@proxy.example.com:443
注意:HTTPX 支持 http:// 和 https:// 类型的代理(对 HTTPS 目标,建议用 https 代理或支持 CONNECT 的 http 代理);不支持 socks 代理(需借助 trio + httpx[http2] 或第三方库如 httpx-socks)。
2. 捕获连接类异常并轮换代理
HTTPX 抛出的连接失败异常主要包括:httpx.ConnectError、httpx.TimeoutException、httpx.ProxyError。应明确捕获这些异常,而非泛用 Exception。
示例逻辑(同步版):
import httpx
<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>",
"<a href="https://www.php.cn/link/d8d7e7b1982462cff20f9d893c472d70">https://www.php.cn/link/d8d7e7b1982462cff20f9d893c472d70</a>"
]</p><p>def request_with_failover(url, proxies, timeout=10):
for proxy in proxies:
try:
with httpx.Client(proxies={"all://": proxy}, timeout=timeout) as client:
r = client.get(url)
return r
except (httpx.ConnectError, httpx.TimeoutException, httpx.ProxyError) as e:
print(f"Proxy {proxy} failed: {e}")
continue
raise RuntimeError("All proxies failed")</p><h1>使用</h1><p>resp = request_with_failover("<a href="https://www.php.cn/link/dc17d9b4862d86f8054735577c04462a">https://www.php.cn/link/dc17d9b4862d86f8054735577c04462a</a>", proxies)
print(resp.json())
</p>3. 支持异步 + 自动重试(推荐生产使用)
异步方式更高效,配合 asyncio 循环和 httpx.AsyncClient 可自然支持并发轮换。还可集成指数退避或限制最大重试次数:
import asyncio
import httpx
<p>async def async_request_with_failover(url, proxies, max_retries=3):
for i, proxy in enumerate(proxies <em> max_retries): # 轮完再重试
if i >= len(proxies) </em> max_retries:
break
try:
async with httpx.AsyncClient(
proxies={"all://": proxy},
timeout=10.0
) as client:
r = await client.get(url)
return r
except (httpx.ConnectError, httpx.TimeoutException, httpx.ProxyError):
continue
raise RuntimeError("No proxy succeeded after retries")</p><h1>使用</h1><p>async def main():
resp = await async_request_with_failover(
"<a href="https://www.php.cn/link/dc17d9b4862d86f8054735577c04462a">https://www.php.cn/link/dc17d9b4862d86f8054735577c04462a</a>",
["<a href="https://www.php.cn/link/5e59e49f5fd2ec69f4522f445c6fa9dd">https://www.php.cn/link/5e59e49f5fd2ec69f4522f445c6fa9dd</a>", "<a href="https://www.php.cn/link/420d174d96d5cc6642bc8c1e765074b2">https://www.php.cn/link/420d174d96d5cc6642bc8c1e765074b2</a>"]
)
print(resp.json())</p><p>asyncio.run(main())
</p>4. 进阶:代理健康检查与动态剔除
为避免反复尝试已失效代理,可在初始化时或运行中做轻量健康检查(如请求 http://httpbin.org/get),并将失败代理临时移出列表或打上“禁用”标记。也可用 functools.lru_cache 缓存代理可用性状态,或结合 Redis 实现多进程共享代理状态。
简单标记示例:
from typing import List, Dict, Optional
<p>class ProxyManager:
def <strong>init</strong>(self, proxy_list: List[str]):
self.proxies = proxy_list.copy()
self.unhealthy: set = set()</p><pre class="brush:php;toolbar:false;">def get_next(self) -> Optional[str]:
for p in self.proxies:
if p not in self.unhealthy:
return p
return None
def mark_unhealthy(self, proxy: str):
self.unhealthy.add(proxy)
不复杂但容易忽略的是:代理轮换必须配合明确的异常类型判断和可控的重试边界,否则可能陷入无限循环或掩盖真实错误。只要把代理切换逻辑收口到一次请求的生命周期内,并做好失败隔离,就能稳定支撑高可用爬取或测试场景。










