本文详解 httpx 异步批量请求中“前150个成功、后续频繁超时”的典型问题，指出根本原因在于无限制并发导致服务端限流或客户端资源耗尽，并提供基于动态任务批处理的稳定并发控制方案。
本文详解 httpx 异步批量请求中“前150个成功、后续频繁超时”的典型问题，指出根本原因在于无限制并发导致服务端限流或客户端资源耗尽，并提供基于动态任务批处理的稳定并发控制方案。
在使用 httpx.AsyncClient 进行大规模异步 HTTP 请求（如 223 个不同域名 URL）时，若直接通过 asyncio.gather(*tasks) 启动全部任务，极易触发看似“随机”的超时现象——尤其在第 150+ 个请求后集中失败，而单独重试又可成功。这并非代码逻辑错误，而是典型的并发失控问题。
根本原因有三：
- ? 服务端反爬/限流：即使目标 URL 域名不同，部分站点可能共享同一 CDN 或后端集群（如 Cloudflare、AWS ALB），对单 IP 的并发连接数进行硬性限制；
- ? 客户端资源瓶颈：操作系统默认文件描述符限制（ulimit -n）、DNS 解析并发阻塞（尤其使用系统默认 DNS 时）、TCP 连接池耗尽等；
- ? 超时配置未适配高并发场景：httpx 默认连接/读取超时（通常 5s）在大量请求排队时极易被击穿。
✅ 正确解法：主动限流 + 动态批处理，而非依赖 gather 全量并发。以下为生产就绪的推荐实现：
import asyncio
import httpx
async def fetch(url: str, client: httpx.AsyncClient) -> httpx.Response:
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
}
# 显式设置超时，避免继承 client 默认值
timeout = httpx.Timeout(15.0, connect=10.0, read=15.0)
return await client.get(url, follow_redirects=True, headers=headers, timeout=timeout)
async def process_url(
url: str,
skip_empty_results: bool,
depth: int,
pattern_list: list,
dataset: dict,
max_retries: int,
client: httpx.AsyncClient,
):
# 实际业务逻辑（此处简化）
try:
response = await fetch(url, client)
# ... 处理响应
return {"url": url, "status": response.status_code, "success": True}
except Exception as e:
return {"url": url, "error": str(e), "success": False}
async def main():
urls_list = [site.strip() for site in start_urls.replace(',', ' ').split() if site.strip()]
pattern_list = second_input or []
batch_size = 30 # 推荐 20–50，根据目标站点容忍度调整
results = []
async with httpx.AsyncClient(
limits=httpx.Limits(max_connections=50, max_keepalive_connections=20),
timeout=httpx.Timeout(10.0),
) as client:
tasks = [
process_url(url, skip_empty_results, depth, pattern_list, dataset, 5, client)
for url in urls_list
]
# 动态批处理：维持约 batch_size 个活跃任务
pending = set()
while tasks or pending:
# 补充新任务至目标并发数
while tasks and len(pending) < batch_size:
pending.add(asyncio.create_task(tasks.pop(0)))
if not pending:
break
# 等待任意任务完成（非全量等待）
done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
for task in done:
try:
result = task.result()
results.append(result)
except Exception as e:
results.append({"error": f"Task failed: {e}"})
return results
# 启动执行
# asyncio.run(main())? 关键优化点说明：
- ✅ 使用 asyncio.wait(..., return_when=FIRST_COMPLETED) 实现“流水线式”并发，避免 gather 的全量阻塞；
- ✅ httpx.AsyncClient 显式配置 limits（连接池上限）和 timeout，防止底层连接耗尽；
- ✅ 批大小 batch_size=30 是经验安全值（远低于 223），可根据实测逐步调优（建议从 10 起测）；
- ✅ 每个 fetch 调用显式传入独立 timeout，避免因个别慢请求拖垮整批；
- ⚠️ 切勿为每个 URL 创建独立 AsyncClient —— 这会引发 TCP 端口耗尽和 TLS 握手开销爆炸。
? 进阶建议：
- 对已知高延迟站点（如含复杂 JS 渲染的页面），可单独设置更长 timeout；
- 添加指数退避重试（配合 tenacity 库）；
- 记录 response.elapsed.total_seconds() 并监控 P95 延迟，动态调整 batch_size；
- 若需极致稳定性，可引入 asyncio.Semaphore 替代手动 pending 管理，语义更清晰。
通过将“全量并发”重构为“可控流水线”，即可彻底解决此类“越往后越容易超时”的顽疾，兼顾效率与鲁棒性。
