6.2 asyncio.gather

并发执行所有协程，返回结果列表

import asyncio
 
async def work(i):
    await asyncio.sleep(0.1)
    return f"task-{i}"
 
async def main():
    results = await asyncio.gather(work(1), work(2), work(3))
    print(results)
 
asyncio.run(main())   # 三个一起跑，总耗时 ≈ 0.1s 而不是 0.3s

为什么推理服务都爱它

推理服务要并发起几十、几百个请求。gather 把这些 IO / 调度类工作并发起来，让 GPU 始终有事可做。

注意

await gather 会等所有任务完成；如果某个任务抛异常，默认会立刻抛出来（其他任务会被取消）。

示范

import asyncio
 
async def fake_infer(prompt):
    await asyncio.sleep(0.05)
    return f"[fake] {prompt}"
 
async def main():
    prompts = ["a", "b", "c", "d"]
    results = await asyncio.gather(*(fake_infer(p) for p in prompts))
    print(results)
 
asyncio.run(main())

✍️ 练习

接着上一节：

把 main 改成用 asyncio.gather 并发 fake_infer
prompts 扩大到 5 个：["a", "b", "c", "d", "e"]
每个 fake_infer 调用时 latency_ms=100
打印结果列表，并 print(f"elapsed ~ {elapsed*1000:.0f}ms")

期望：总耗时 ~ 100ms（并发）而不是 500ms（串行）。

💡 思路提示

gather(*coros) 接受一堆协程对象，用 * 解包
生成器表达式 (fake_infer(p) for p in prompts) 可以直接喂给 *
asyncio.get_event_loop().time() 已弃用，用 time.perf_counter() 计时更稳

✅ 参考解法

import asyncio
import time
 
async def fake_infer(prompt, latency_ms=100):
    await asyncio.sleep(latency_ms / 1000)
    return f"[fake] {prompt}"
 
async def main():
    prompts = ["a", "b", "c", "d", "e"]
    start = time.perf_counter()
    results = await asyncio.gather(*(fake_infer(p) for p in prompts))
    elapsed = time.perf_counter() - start
    print(results)
    print(f"elapsed ~ {elapsed*1000:.0f}ms")
 
asyncio.run(main())

🔍 进阶思考

如果某个 fake_infer 抛异常，gather 默认立刻传播，所有协程被取消。生产里常用 return_exceptions=True 拿到所有结果（包括异常）。

results = await asyncio.gather(*tasks, return_exceptions=True)
for r in results:
    if isinstance(r, Exception):
        log(r)
    else:
        use(r)

6.2 asyncio.gather：并发执行多个协程