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.php.cn/link/dfcfbb196720bb3febac626b8b9d082d", "https://www.php.cn/link/07f2d8dbef3b2aeca9cb258091bc3dba", "https://www.php.cn/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.php.cn/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.php.cn/link/dc17d9b4862d86f8054735577c04462a", ["https://www.php.cn/link/5e59e49f5fd2ec69f4522f445c6fa9dd", "https://www.php.cn/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)不复杂但容易忽略的是:代理轮换必须配合明确的异常类型判断和可控的重试边界,否则可能陷入无限循环或掩盖真实错误。只要把代理切换逻辑收口到一次请求的生命周期内,并做好失败隔离,就能稳定支撑高可用爬取或测试场景。










