本文介绍如何使用 asyncio.Queue 实现真正的并发生产者-消费者流，让多个 slow_task 并行处理异步生成器输入，并在结果就绪时立即产出（无需等待顺序），显著提升吞吐量。

本文介绍如何使用 asyncio.queue 实现真正的并发生产者-消费者流，让多个 `slow_task` 并行处理异步生成器输入，并在结果就绪时立即产出（无需等待顺序），显著提升吞吐量。

在原始实现中，my_gen() 采用串行 async for ... yield await slow_task(item) 模式，导致整体耗时约为 len(items) × 0.5s ≈ 6s（12 项 × 0.5s），且必须严格按输入顺序阻塞等待每个任务完成。这既浪费了 I/O 空闲时间，又违背了异步编程“尽早响应”的设计初衷。

要突破这一瓶颈，核心思路是解耦数据供给（Producer）、并发处理（Worker Pool）和结果收集（Consumer）三个阶段，并借助 asyncio.Queue 实现线程安全的异步通信。以下是一个完整、可运行的生产者-消费者异步生成器流水线实现：

import asyncio
import random
from collections.abc import AsyncGenerator

async def input_gen() -> AsyncGenerator[str, None]:
    """生产者：逐个生成输入项，模拟外部数据源（如 API 流、文件读取等）"""
    for char in "abc123xyz789":
        await asyncio.sleep(0.1)  # 模拟非均匀输入间隔
        yield char

async def slow_task(input_queue: asyncio.Queue, output_queue: asyncio.Queue) -> None:
    """消费者工作协程：持续从 input_queue 取任务，处理后将结果写入 output_queue"""
    while True:
        try:
            # 使用 get() 而非 get_nowait() 保证优雅退出（当 input_queue 关闭时，需配合 sentinel 或 task_done）
            item = await input_queue.get()
            await asyncio.sleep(random.uniform(0.2, 0.7))  # 模拟波动延迟
            result = f"{item}_loaded"
            await output_queue.put(result)  # 使用 put() 更安全（避免队列满时崩溃）
            input_queue.task_done()
        except asyncio.CancelledError:
            break  # 协程被取消时主动退出

async def concurrent_processor(
    input_gen: AsyncGenerator[str, None],
    max_workers: int = 4
) -> AsyncGenerator[str, None]:
    """
    主协调器：启动生产者-消费者流水线，返回一个异步生成器，
    其产出顺序即为各 slow_task 完成的自然顺序（FIFO on completion）
    """
    input_queue = asyncio.Queue(maxsize=max_workers * 2)  # 防止生产过快压垮内存
    output_queue = asyncio.Queue()

    # 启动 N 个并发 worker 协程
    workers = [
        asyncio.create_task(slow_task(input_queue, output_queue))
        for _ in range(max_workers)
    ]

    # 【生产阶段】将 input_gen 的所有项推入 input_queue
    async for item in input_gen:
        await input_queue.put(item)  # 自动背压：若队列满则暂停生产

    # 【等待处理完成】确保所有已入队任务被执行完毕
    await input_queue.join()

    # 【消费阶段】持续从 output_queue 拉取已完成结果（无序但实时）
    # 注意：此处不使用 output_queue.empty() 判定，因其在多协程下不可靠
    # 改用「计数」或「sentinel」更健壮；本例为简洁演示，假设全部产出后才结束
    done_count = 0
    total_items = len("abc123xyz789")
    while done_count <p>✅ <strong>关键优势说明</strong>：</p><div class="aritcle_card flexRow artxards">
											<div class="artcardd flexRow">
												<a class="aritcle_card_img" rel="nofollow" href="/ai/3014" title="快面"><img
														src="https://img.php.cn/upload/ai_manual/001/246/273/178408065276752.png" alt="快面" onerror="this.onerror='';this.src='/static/lhimages/moren/morentu.png'" ></a>
												<div class="aritcle_card_info flexColumn">
													<a rel="nofollow" href="/ai/3014" title="快面" class="overflowclass">快面</a>
													<p class="overflowclass">一款AI办公效率工具，主要用于Kuaimian 是全流程 AI 面试助手：支持 AI 简历优化、AI 模拟面试，并在真实面试与在线笔试过程中提供实时 AI 辅助，适合需要提升相关任务效率的用户。</p>
												</div>
												<a rel="nofollow" href="/ai/3014" title="快面" class="aritcle_card_btn flexRow flexcenter"><b></b><span>下载</span>
												</a>
											</div>
										</div>

• 真正并发：4 个 slow_task 协程并行执行，总耗时接近 max(输入间隔总和, 最长单任务耗时) ≈ 1.2s + 0.7s ≈ 1.9s，远优于串行的 6s；
• 即时产出：首个 slow_task 完成即刻 yield，无需等待其他项；
• 背压控制：input_queue.put() 在满时自动挂起生产者，防止内存溢出；
• 健壮清理：worker 协程支持取消，主流程结束后资源自动释放。

⚠️ 注意事项：

• 若需严格保序（如后续逻辑依赖顺序），应在 output_queue 中附加序号元组 (idx, result)，再由消费者排序；
• 生产者可能早于消费者启动，建议 input_queue 设置合理 maxsize；
• output_queue 的消费逻辑应避免 while not queue.empty()（竞态条件），推荐使用计数、sentinel 或 asyncio.wait_for(queue.get(), timeout=...) 做超时防护。

该模式是构建高性能异步 ETL、实时数据管道、微服务消息编排等场景的基石，兼顾并发性、响应性与可维护性。