本文介绍一种基于 asyncio.Queue 的生产者-消费者模式，实现对异步输入流的并发处理与即时产出，支持无序但完整的结果流，显著提升 slow_task 类型 I/O 密集型任务的吞吐量。
本文介绍一种基于 `asyncio.queue` 的生产者-消费者模式，实现对异步输入流的并发处理与即时产出，支持无序但完整的结果流，显著提升 `slow_task` 类型 i/o 密集型任务的吞吐量。
在原始代码中，my_gen() 以串行方式依次 await slow_task(item)，导致整体耗时线性增长（约 12 × 0.5s = 6s + 等待开销），且无法利用异步并发优势。要真正实现「边生产、边消费、边产出」，需解耦数据生成、并行处理与结果收集三个阶段——这正是 asyncio.Queue 擅长的场景。
核心思路如下：
- 生产者协程：从 input_gen() 异步迭代，将每个 item 推入 input_queue；
- 多个消费者协程：并行运行（如 4 个 slow_task 实例），持续从 input_queue 取值、执行耗时操作，并将结果立即写入 output_queue；
- 结果聚合：待所有输入处理完毕后，清空 output_queue 获取全部结果（顺序随机，但不丢失任何项）。
以下是完整可运行的优化实现：
import asyncio
import random
from collections.abc import AsyncGenerator
async def input_gen() -> AsyncGenerator[str, None]:
"""模拟异步数据源：逐字符产出，带固定延迟"""
for char in "abc123xyz789":
await asyncio.sleep(0.1)
yield char
async def slow_task(input_queue: asyncio.Queue, output_queue: asyncio.Queue) -> None:
"""消费者工作单元：无限循环取值、处理、回传结果"""
while True:
try:
item = await input_queue.get()
# 模拟波动的 I/O 延迟（更贴近真实场景）
await asyncio.sleep(random.uniform(0.2, 0.7))
result = f"{item}_loaded"
output_queue.put_nowait(result)
input_queue.task_done() # 标记该任务完成
except asyncio.CancelledError:
break # 优雅退出
async def main() -> list[str]:
input_queue = asyncio.Queue(maxsize=0) # 无界队列，避免阻塞生产者
output_queue = asyncio.Queue(maxsize=0)
# 启动 4 个并发消费者任务
consumer_tasks = [
asyncio.create_task(slow_task(input_queue, output_queue))
for _ in range(4)
]
# 生产者：异步迭代并填充 input_queue
async for item in input_gen():
input_queue.put_nowait(item)
# 等待所有已入队任务执行完毕（注意：不等待新任务，只等当前队列清空）
await input_queue.join()
# 取消仍在运行的消费者（它们处于 while True 循环中）
for task in consumer_tasks:
task.cancel()
await asyncio.gather(*consumer_tasks, return_exceptions=True)
# 收集所有产出结果
results = []
while not output_queue.empty():
results.append(await output_queue.get())
output_queue.task_done()
return results
if __name__ == "__main__":
results = asyncio.run(main())
print(f"\n✅ 共获取 {len(results)} 个结果：{results}")
✅ 关键设计说明：
- 使用 input_queue.join() 替代 await asyncio.sleep() 等待，确保精确同步：仅当所有已入队 item 均被 task_done() 标记后才继续；
- 消费者采用 while True + try/except CancelledError 结构，保证 asyncio.gather(..., return_exceptions=True) 能安全终止；
- put_nowait() 避免队列满时协程挂起，配合 maxsize=0（无界）提升吞吐；
- 输出顺序天然无序，符合「尽快返回可用结果」的设计目标；若需保序，可在结果中附加索引并后续排序（但会牺牲实时性）。
⚠️ 注意事项：
- 不要直接在 async for 中 await 多个 slow_task —— 这仍为逻辑串行；
- 避免在消费者中使用 await output_queue.put(...)（除非有背压需求），put_nowait() + task_done() 组合更高效；
- 实际部署时，消费者数量（如 range(4)）应根据 I/O 并发能力与系统资源调优，通常设为 CPU 核心数的 2–4 倍。
通过该模式，原本约 6 秒的串行流程可压缩至约 1–2 秒（取决于并发度与任务延迟分布），真正释放 asyncio 的并发潜力。
