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 httpxproxies = [ "https://www./link/dfcfbb196720bb3febac626b8b9d082d", "https://www./link/07f2d8dbef3b2aeca9cb258091bc3dba", "https://www./link/d8d7e7b1982462cff20f9d893c472d70" ]
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")
使用
resp = request_with_failover("https://www./link/dc17d9b4862d86f8054735577c04462a", proxies) print(resp.json())
3. 支持异步 + 自动重试（推荐生产使用）
异步方式更高效，配合 asyncio 循环和 httpx.AsyncClient 可自然支持并发轮换。还可集成指数退避或限制最大重试次数：
import asyncio import httpxasync def async_request_with_failover(url, proxies, max_retries=3): for i, proxy in enumerate(proxies max_retries): # 轮完再重试 if i >= len(proxies) 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")
使用
async def main(): resp = await async_request_with_failover( "https://www./link/dc17d9b4862d86f8054735577c04462a", ["https://www./link/5e59e49f5fd2ec69f4522f445c6fa9dd", "https://www./link/420d174d96d5cc6642bc8c1e765074b2"] ) print(resp.json())
asyncio.run(main())
4. 进阶：代理健康检查与动态剔除
为避免反复尝试已失效代理，可在初始化时或运行中做轻量健康检查（如请求 http://httpbin.org/get），并将失败代理临时移出列表或打上“禁用”标记。也可用 functools.lru_cache 缓存代理可用性状态，或结合 Redis 实现多进程共享代理状态。
简单标记示例：
from typing import List, Dict, Optionalclass ProxyManager: def init(self, proxy_list: List[str]): self.proxies = proxy_list.copy() self.unhealthy: set = set()
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)不复杂但容易忽略的是：代理轮换必须配合明确的异常类型判断和可控的重试边界，否则可能陷入无限循环或掩盖真实错误。只要把代理切换逻辑收口到一次请求的生命周期内，并做好失败隔离，就能稳定支撑高可用爬取或测试场景。
