Python asyncio.Queue 怎样用 join 等待所有任务完成

Python asyncio.Queue 应在生产者完成本批投递后使用 await queue.join() 等待，并让消费者对每个成功取出的条目恰好调用一次 queue.task_done()。join 关注的是尚未确认完成的条目数，不等于等待队列变空，也不会自动停止消费者。

假设一个批处理服务把字符串交给两个消费者转换，主协程需要等到本批处理结束才能交付结果。升级 Python 时，最容易混淆的是把新的退出 API 与原来的完成记账混在一起。官方文档入口：https://docs.python.org/3.14/library/asyncio-queue.html。下文代码是原创说明示例，未在本文制作过程中执行。

从旧版消费者迁移时，先保留 join 的记账语义

从 Python 3.11、3.12 迁移到较新版本，不需要为了使用 join 重写整个队列协议。真正要分开的是三个问题：条目还在不在容器里、条目有没有完成记账、消费者协程是否退出。

本文完整例子使用 Python 3.11+ 的共同能力，便于保留旧版运行环境。队列只在同一个事件循环里操作；它不是跨线程共享队列。

为什么队列空了，批处理仍可能没结束

消费者调用 get 后，条目已经从容器移出，但转换函数可能仍在等待 I/O。此时 qsize 可以为零，业务处理却没有结束。因此，用循环检查 empty，再睡一小会，不是可靠的完成协议。

正常记账下，每次成功 put 增加一个未完成条目，task_done 消去一个。get 只负责取出，不替代完成确认；task_done 也不会移除仍在容器里的条目。若投递三条、消费者已取走三条但只确认两条，join 仍应等待剩余那一条。这是推导示例，不是运行日志。

另一个容易遗漏的情况是主协程刚创建一个生产者任务，就马上 await join。若生产者还没放入任何条目，join 可以直接返回。有限批次应先确认生产者完成全部投递，再等待记账归零；多个生产者则要先等待它们全部结束，不能只等其中一个。

用 TaskGroup 管理消费者，按条目确认完成

这个例子的业务约定很简单：转换函数返回后才记录结果，任一消费者处理异常都让整个批次失败。不把处理失败偷偷变成成功，也不把未完成工作转交给没有被等待的后台任务。

import asyncio

STOP = object()

async def transform(item: str) -> str:
    # 用可等待的业务函数表示一次处理，不启动脱离管理的后台任务。
    await asyncio.sleep(0)
    return item.upper()

async def consumer(queue: asyncio.Queue, results: list) -> None:
    while True:
        item = await queue.get()
        try:
            if item is STOP:
                return
            value = await transform(item)
            results.append((item, value))
        finally:
            # 只给本次成功取出的条目确认一次；异常仍向 TaskGroup 传播。
            queue.task_done()

async def run_batch(items: list[str]) -> list:
    queue = asyncio.Queue(maxsize=8)
    results = []
    worker_count = 2

    async with asyncio.TaskGroup() as group:
        for _ in range(worker_count):
            group.create_task(consumer(queue, results))

        # 生产者先完成全部投递，避免空队列上的 join 提前返回。
        for item in items:
            await queue.put(item)
        await queue.join()

        # 每个消费者一个哨兵；哨兵也属于入队条目，需要记账。
        for _ in range(worker_count):
            await queue.put(STOP)
        await queue.join()
    # 离开上下文后，消费者协程也全部结束。
    return results

async def main() -> None:
    results = await run_batch(["alpha", "beta", "gamma"])
    # 并发完成顺序不是接口承诺；只展示按输入排序后的业务结果。
    print(sorted(results))

if __name__ == "__main__":
    asyncio.run(main())

这里有两次 join，职责不同。第一次等待业务条目完成记账；随后放入每个消费者一个退出哨兵，第二次等待这些控制条目完成记账。最后退出 TaskGroup 上下文，才是消费者协程全部结束的边界。

get 放在 try 外面，是为了不在“尚未成功取得条目”的取消或异常路径里调用 task_done。finally 则覆盖拿到条目后的正常返回和处理异常。task_done 没有 await，不应写成 await queue.task_done()。

有界队列的 maxsize=8 是示例容量，不是性能结论。先创建消费者再投递，可避免主协程填满队列后没有消费者取走条目的死等。返回结果不承诺并发完成顺序；若业务必须保留输入顺序，应随条目传递序号并按序整理，而不是假定两个消费者总按取出顺序完成。

Python 3.13 以后怎样替换退出哨兵

迁移到 3.13+ 后，如果选择 shutdown 协议，可以在全部生产者结束后调用 queue.shutdown()，不再追加哨兵。消费者在队列关闭且清空后会从 get 收到 asyncio.QueueShutDown；此异常应作为正常退出信号，在获取条目的位置处理。

这不是只把哨兵那行删掉就算迁移完成：需要同时改变消费者的取件异常处理，并继续保证所有已取出的业务条目都有一次完成确认。shutdown 之后不能继续 put；仍在等待 put 的生产者也会受到关闭影响，所以必须明确谁负责停止生产。

优雅关闭保留已经排队的条目供消费者处理。shutdown(immediate=True) 则可能丢弃尚未处理的队列条目并调整计数；因此 join 解除等待不再能按通常约定解释成“所有条目都处理过了”。强制退出应记录未处理或中止状态，而不是对外宣布成功。

若部署中仍有 3.12 节点，就保留哨兵版本或明确划分不同实现，不能直接调用不存在的 shutdown。优先保持一套清楚的退出协议，不要同时发送哨兵、关闭队列，又无条件取消所有消费者。

异常、取消与业务成功要分别判断

finally 中确认条目，表示本次取件已经终结，不代表它转换成功。本文采用失败即终止的批次策略：transform 抛出普通异常时，TaskGroup 会取消组内其他消费者，并使等待中的主协程退出正常路径；异常会从上下文传播。调用方必须将其视为失败，不能在异常处理里仍打印“全部成功”。

如果改成允许部分失败，就要为每条任务单独记录成功、失败和重试决定。要重新入队时，新一次 put 产生新一份记账，旧取件仍只能确认一次。无限重试会使批次迟迟不能完成，重试上限和失败出口需要由业务协议规定，join 不会替你做这些决定。

取消也是独立结果。某个消费者取件后被取消，finally 可能仍会确认该条目，但业务可能没有完成。不要吞掉 CancelledError 后继续当成功运行。若给等待设置超时，超时只限制等待时长，不会自动给剩余条目生成成功结果，也不是持久消息系统的可靠投递机制。

迁移时可由项目自己的测试覆盖：空批次、处理函数抛错、最后一条较慢、消费者取消以及关闭时尚有排队条目。这里列的是后续工程检查点，本文没有运行测试或生成测试报告。

迁移前值得确认的几个问题

join 卡住时，先检查什么？ 看是否漏掉了某条成功取件的 task_done，以及消费者是否已经因异常退出。不要为了让等待结束，随意补很多次 task_done；调用过多会抛 ValueError。

join 返回就能退出程序吗？ 它不负责停止等待 get 的消费者。哨兵、优雅关闭或明确取消策略之后，仍要等待消费者协程结束。本文例子用 TaskGroup 完成这一层管理。

队列处理完就能保证数据已持久化吗？ 不能。若 transform 只是发起了写入却没有等待提交，提前 task_done 就把完成边界放错了。应把实际需要的业务提交纳入被等待的处理函数，并自行设计幂等、重试和持久化策略。

迁移时保留四个约束即可：生产者先收口，每次成功取件一次确认，消费者退出另行管理，业务成功另有结果记录。join 是一个很有用的批次记账等待点，但不是同时解决这四件事的总开关。

参考：Python 3.14 队列、Python 3.12 队列、TaskGroup 与取消。