直接结论:用 asyncio.taskgroup 并发发请求 + aiohttp 替代 requests,qps 能翻 3–4 倍,且内存更稳;因其支持任一失败即自动取消其余任务,避免资源泄漏,而 asyncio.gather 无此能力,且必须配合异步客户端与信号量控制并发。

直接结论:用 asyncio.TaskGroup 并发发请求 + aiohttp 替代 requests,QPS 能翻 3–4 倍,且内存更稳。
为什么不用 asyncio.gather 而要用 asyncio.TaskGroup
在 API 测试中,你常要并发调多个端点(比如 /user、/order、/profile),并统一等全部完成。用 asyncio.gather 看似简单,但一旦某个请求失败(如 502 或超时),其余还在跑的请求不会自动取消——测试进程卡住、资源泄漏、超时风险陡增。
asyncio.TaskGroup 则天然支持“任一失败即全 cancel”:
async with asyncio.TaskGroup() as tg:
tg.create_task(client.get("/user"))
tg.create_task(client.get("/order"))
tg.create_task(client.get("/profile"))
- 任意一个
create_task抛出未捕获异常(如aiohttp.ClientConnectorError),其他任务立刻被 cancel - 无需手动写
try/finally清理连接或重试状态 - 错误堆栈直接指向具体哪个 URL 失败,调试更快
必须换掉 requests,否则异步白搭
很多人写 async def test_api(): requests.get(...),结果发现比同步还慢——因为 requests 是阻塞式 HTTP 客户端,它会把整个事件循环卡死在 select() 等待上。
正确做法只有一条:所有 I/O 必须用异步原生库:
- HTTP 客户端:用
aiohttp或httpx.AsyncClient(注意不是httpx.Client) - 数据库:用
asyncpg(PostgreSQL)、aiomysql(MySQL),别用psycopg2或pymysql - Redis:用
redis-py的Redis.from_url(..., encoding="utf-8", decode_responses=True)不行,得用aioredis或新版redis.asyncio.Redis
漏掉任何一个,就等于在高速公路上开拖拉机——协程再多也跑不快。
并发数不是越多越好,得用 asyncio.Semaphore 控制
测试环境里,直接 create_task 1000 次请求,大概率触发目标服务限流、本地文件描述符耗尽(OSError: [Errno 24] Too many open files),甚至压垮测试机自身网络栈。
用信号量硬限并发数是最简单有效的手段:
sem = asyncio.Semaphore(20) # 同时最多 20 个连接 <p>async def fetch_one(url): async with sem: # 这里会阻塞直到有空位 return await client.get(url)</p>
- 20 是经验值,适合大多数内网测试环境;公网压测建议从 5 开始逐步加
- 别用
time.sleep()或asyncio.sleep()代替限流——它们不释放连接,只是假装在等 - 如果测试目标是网关类服务,还需配合
client.timeout设置(如aiohttp.ClientTimeout(total=5)),避免单个慢请求拖垮整批
asyncio.run() 前必须设策略,否则白优化
Python 3.12 默认事件循环已提速,但如果你在 asyncio.run() 之后才调 asyncio.set_event_loop_policy(),那策略根本没生效——日志里也看不出错,就是性能上不去。
正确顺序只有这一种:
import asyncio import aiohttp <h1>必须在 run 之前设置</h1><p>asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy())</p><p>async def main(): async with aiohttp.ClientSession() as client:</p><h1>... 测试逻辑</h1><p>if <strong>name</strong> == "<strong>main</strong>": asyncio.run(main())</p>
- 不要在
main()函数里设策略——太晚了 - 除非你明确需要
uvloop(比如 UDP 高频收发),否则 Python 3.12 原生策略和uvloop性能差不到 10%,还少一个依赖 - 如果用 pytest-asyncio,确保 pytest 配置里启用了
asyncio_mode = "auto",否则async def test_*根本不走事件循环
真正卡性能的地方往往不是并发模型本身,而是连接复用没开、超时设得太宽、或者某处偷偷调了同步函数。跑通 TaskGroup + aiohttp + Semaphore 这三件套后,再逐个检查这些细节,才能稳住高 QPS。
大量免费API接口:立即使用
涵盖生活服务API、金融科技API、企业工商API、等相关的API接口服务。免费API接口可安全、合规地连接上下游,为数据API应用能力赋能!











