一、起因：代理验证拖垮了整个采集系统

先交代一下背景。我在一家电商公司做数据采集，核心系统是竞品价格监控——每天爬天猫、京东、拼多多的商品价格，日采集量在几十万到百万级。

刚开始做的时候，代理管理这块是比较粗糙的——抓了一批免费代理存Redis里，爬虫取的时候随机拿一个，挂了就换。看起来能跑，但实际上问题很大。

最头疼的是代理验证。当时的逻辑是这样的：用requests库写了个定时任务，每隔几分钟遍历一遍Redis里的代理，挨个发请求验证是否存活。

问题出在哪呢？requests是同步阻塞的。假设池子里有500个代理，每个代理验证超时设5秒，光一轮验证就得跑2500秒——不现实。实际把超时压到1秒，但就算这样，500个代理串行验证也要差不多一分钟。更糟的是，这一分钟里爬虫还在取代理，可能取到的就是还没验证到的失效代理。

后来代理池慢慢从免费切换到付费，稳定了一些，但体量从几百涨到上千之后，验证延迟的问题又回来了。

二、第一次尝试：用aiohttp做异步验证

第一个想到的方案是用aiohttp。异步并发验证，一次发几十个请求出去，比串行快太多了。

import aiohttp import asyncio async def validate_proxy(session, proxy): try: async with session.get( "http://httpbin.org/ip", proxy=proxy, timeout=aiohttp.ClientTimeout(total=3) ) as resp: return proxy if resp.status == 200 else None except: return None async def batch_validate(proxies): async with aiohttp.ClientSession() as session: tasks = [validate_proxy(session, p) for p in proxies] results = await asyncio.gather(*tasks, return_exceptions=True) return [r for r in results if r is not None]

改成异步之后，验证1000个代理大概十几秒就搞定了，效率提升很明显。

但这个方案用了一段时间后，发现了一个尴尬的问题：aiohttp不支持SOCKS5代理。我们有一部分代理是SOCKS5协议的，aiohttp原生不支持，得用aiohttp-socks这个第三方库，多了一层依赖不说，偶尔还会有兼容性问题。

三、最终方案：httpx.AsyncClient

后来全面切到了httpx。httpx的API和requests几乎一模一样，但支持异步，而且原生支持SOCKS5代理。

核心改动就是把requests换成httpx.AsyncClient：

import httpx import asyncio async def validate_proxy(client, proxy_url): try: resp = await client.get( "http://httpbin.org/ip", proxy=proxy_url, timeout=3.0 ) if resp.status_code == 200: # 记录响应延迟 elapsed = resp.elapsed.total_seconds() return {"proxy": proxy_url, "latency": elapsed} except Exception: pass return None async def validate_all(proxy_list, concurrency=30): sem = asyncio.Semaphore(concurrency) async def bounded_validate(p): async with sem: async with httpx.AsyncClient() as client: return await validate_proxy(client, p) tasks = [bounded_validate(p) for p in proxy_list] results = await asyncio.gather(*tasks) return [r for r in results if r is not None]

四、一个容易忽略的坑：事件循环里的同步代码

这里有个细节特别容易踩——不要以为写了async函数就万事大吉了。

我在做代理网关的时候，路由层用了FastAPI的async，但是底层调代理验证还傻乎乎用requests同步库。代码看起来没问题，压测的时候QPS一直上不去。排查了半天才反应过来：requests.get()把整个事件循环都堵死了，async等于白写。

结论：用了async，就全线用异步库。千万别混用同步库。

五、代理延迟过滤：快和准的平衡

异步验证跑起来之后，发现另一个问题：验证通过不代表"好用"。

有些代理能连通，但响应时间动不动就5秒以上。爬虫那边对时效有要求——价格数据晚几分钟就失去参考价值了。所以只验证存活不够，还得过滤掉慢的。

加了一个延迟阈值过滤：

def filter_by_latency(results, max_latency=3.0): """只保留响应时间在阈值以内的代理""" fast = [r for r in results if r["latency"] <= max_latency] slow = len(results) - len(fast) print(f"过滤前: {len(results)} 个, 过滤后: {len(fast)} 个 (剔除 {slow} 个高延迟)") return fast

阈值设了3秒。设完之后可用代理数量大概少了三四成，但实际采集成功率反而上去了——因为爬虫不再在慢代理上浪费时间。这算是我踩过的一个认知偏差：代理数量重要，但代理质量更重要。

六、验证端点的选择

验证代理能不能用，需要发一个请求确认。很多人用httpbin.org，我也用了很长一段时间。

后来发现有些代理莫名其妙连不上httpbin，排查下来是代理服务商那边可能屏蔽了某些常用测试站点。后来自己在服务器上搭了个最简单的验证端点，就返回一个200和一个JSON，彻底避免了这个问题。

# 最简单的验证端点 (FastAPI) from fastapi import FastAPI app = FastAPI() @app.get("/ping") async def ping(): return {"status": "ok"}

七、实际效果

切换前后的对比（基于我们系统的实际数据，做了约数处理）：

这里要说明一下，代理有效率本身不会因为换异步库就提高——代理还是那些代理，能不能用是由代理本身决定的。异步验证的价值在于更快地发现失效代理、更快地剔除慢代理，从而提升整体的采集效率。

八、总结

回头看整个过程，有几个体会：

1. 异步不是银弹，但代理管理这个场景确实适合异步。代理验证是典型的IO密集型任务，几千个代理挨个发请求的场景，异步的价值特别明显。

2. httpx是当下做代理管理的一个好选择。API和requests一致，学习成本低；原生异步支持；SOCKS5代理原生可用；HTTP/2支持更好。

3. 关注延迟，不只是存活。能连通的代理不等于好代理。加一个延迟过滤，虽然代理池看起来"瘦了"，但爬虫整体效率反而是提升的。

4. 验证端点自己搭一个。别依赖第三方服务，稳定性不可控。

以上是我在代理池管理上的一些经验。如果你也在做类似的系统，欢迎评论区交流。