用aiohttp.ClientSession+asyncio.gather并发请求可将100个接口测试从8秒降至1.1秒,但裸用易触发ConnectionRefusedError等错误;必须加return_exceptions=True捕获异常,并配TCPConnector(limit=0, limit_per_host=0, keepalive_timeout=1, use_dns_cache=False)确保本地测试稳定。

直接上结论:用 aiohttp.ClientSession + asyncio.gather 并发发请求,100 个接口测试用例从串行 8 秒压到并行 1.1 秒;但裸写会触发 ConnectionRefusedError、ClientConnectorError 或本地端口耗尽——关键不是“怎么快”,而是“怎么稳着快”。
为什么并发测试常卡在第一个失败请求上?
异步请求失败后不显式捕获,整个 asyncio.gather 就会中断,后续用例全被跳过。同步 requests 至少能 try/except 单个请求,而 aiohttp 的异常类型更细、更隐蔽:
-
aiohttp.ClientConnectorError:目标服务还没起来、端口没开、Docker 容器未就绪 -
aiohttp.ServerDisconnectedError:FastAPI/Flask 测试服务器主动断连(尤其 debug 模式下 reload 频繁) -
asyncio.TimeoutError:测试环境网络抖动,或 mock server 响应延迟超预期
必须加 return_exceptions=True,让失败请求返回异常对象而非抛出,再统一处理:
results = await asyncio.gather(*tasks, return_exceptions=True)
for i, r in enumerate(results):
if isinstance(r, Exception):
print(f"用例 {i} 失败:{type(r).__name__}")
else:
assert r.status == 200
测试场景下 TCPConnector 怎么配才不翻车?
本地跑接口测试,通常只压自己本机的 Flask/FastAPI 服务,没有反爬、没有限流,但容易因连接复用不当导致“假失败”:
- 不设
limit_per_host=0:默认 100 限制,但单机测试常只打http://localhost:8000,所有请求挤一个 host,实际并发数被锁死 - keepalive_timeout 过长:测试用例反复启停服务时,旧连接还挂着,新请求可能复用已失效的 socket
- DNS 缓存无意义:本地测
localhost或127.0.0.1,关掉 DNS 缓存省事
推荐配置:
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac,无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启:辅助功能 + 屏幕录制权限
connector = aiohttp.TCPConnector(
limit=0, # 不限总连接数
limit_per_host=0, # 关键:解除单 host 限制
keepalive_timeout=1, # 短保活,适配频繁重启
enable_cleanup_closed=True, # 及时回收关闭的连接
use_dns_cache=False # localhost 不需要 DNS 缓存
)
如何避免测试用例之间互相污染?
测试框架(如 pytest-asyncio)里多个 test_* 函数共用一个 event loop,若前一个测试没 clean 掉 session 或 connector,后一个可能拿到半残状态:
- 不要全局复用
ClientSession:每个测试函数内建 session,用完自动 close - 别在
setup_method里初始化 session:pytest 的 async fixture 更可控 - CookieJar 默认共享:如果测试含登录态,需显式传
cookie_jar=aiohttp.CookieJar(unsafe=True)避免跨用例 session 混淆
一个安全的 pytest fixture 示例:
@pytest.fixture
async def aiohttp_client():
connector = aiohttp.TCPConnector(limit=0, limit_per_host=0)
timeout = aiohttp.ClientTimeout(total=5)
async with aiohttp.ClientSession(
timeout=timeout,
connector=connector,
cookie_jar=aiohttp.CookieJar(unsafe=True)
) as session:
yield session
真正难的不是并发本身,是测试环境下那些“本不该出问题却总出”的细节:服务启动时机、连接池残留、DNS 缓存错位、异常传播路径——这些地方一松懈,100 个用例就变成 10 个能跑通。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!










