本文介绍如何在 Python 中真正并发地消费多个异步生成器（async def ... yield），不等待任一生成器完成即可立即收集所有已产出的值，实现低延迟、高吞吐的异步流式扁平化。
本文介绍如何在 python 中真正并发地消费多个异步生成器（`async def ... yield`），不等待任一生成器完成即可立即收集所有已产出的值，实现低延迟、高吞吐的异步流式扁平化。
在异步爬虫、数据流处理或微服务聚合等场景中，我们常需并发请求多个资源，并对每个响应流式解析（如逐行解析 HTML 或分块读取 API 响应）。此时，若每个请求封装为一个异步生成器（async def fetch_page(url) -> AsyncGenerator[str, None]），目标是：一旦任意生成器 yield 出一个值，就立刻获取它、加入结果列表，无需等待其他生成器结束或按顺序轮询。
但直接使用 async for 串行遍历会阻塞——例如 for i in range(10): async for x in test(i): ... 实际是顺序执行，总耗时 ≈ 10 × 单个生成器耗时；而 asyncio.gather 或 as_completed 则无法直接作用于异步生成器对象（它们不是可 await 的协程），导致 TypeError。
✅ 正确解法的核心思想是：将每个异步生成器的消费逻辑（即 async for 循环）封装为独立的异步任务（Task），再并发调度这些任务。这样，10 个生成器的 yield 操作完全并行，首个元素在约 0.5 秒后即可产出，整体耗时仅略高于单个生成器的最长生命周期。
以下为推荐实现（Python 3.11+，使用 asyncio.TaskGroup）：
import asyncio
import random
async def test(n):
# 模拟网络延迟：每个生成器启动时固定延迟 0.5s
await asyncio.sleep(0.5)
for i in range(10):
yield 10 * n + i
# 模拟元素间随机延迟（如网络抖动）
await asyncio.sleep(random.random() / 10)
async def run_test():
results = [] # 共享结果列表
async with asyncio.TaskGroup() as tg:
for i in range(10):
# 为每个生成器创建独立消费任务
async def consume_generator(idx):
async for item in test(idx):
results.append(item)
tg.create_task(consume_generator(i))
return results
# 运行并验证
if __name__ == "__main__":
output = asyncio.run(run_test())
print(f"共收集 {len(output)} 个元素，首 10 个: {output[:10]}")
# 示例输出（每次顺序不同）: [0, 10, 20, ..., 90, 1, 11, 21, ...]
? 关键要点说明：
-
TaskGroup确保所有子任务并发执行且自动等待全部完成，比asyncio.gather更安全（自动传播异常、资源清理更可靠）； - 内部
consume_generator是闭包函数，可安全访问外层results列表，避免线程/协程安全问题（协程天然单线程，list.append()是原子操作）； - 每个
test(i)的async for在独立任务中运行，彼此不阻塞——test(0)产出第 1 个值时，test(1)可能刚启动或正在产第 3 个值。
⚠️ 注意事项与替代方案：
- 若需兼容 Python asyncio.gather(*tasks) 替代
TaskGroup，但需手动捕获异常（gather(return_exceptions=True)）； - 切勿尝试
await test(i)——异步生成器本身不可 await，必须用async for或anext()驱动； - 如需严格控制内存（如生成器项极大），可改用
asyncio.Queue实现生产者-消费者模式，避免全量list.append()； - 若需保持原始生成器的“批次”语义（如按页分组），应在
consume_generator中收集子列表再extend，而非直接append。
此方案完美匹配真实场景：10 个网页并发抓取，每个页面解析出若干字符串，你希望第一个页面的第一个字符串在 0.5 秒后立即可用，而非等待全部 10 页解析完毕——这正是异步生成器并发消费的价值所在。
