Python asyncio.Queue 关闭后怎样结束消费者

如果消费者一直停在 await queue.get()，不要再用“队列为空”猜测它应该退出。Python 3.13 起，asyncio.Queue 提供了 shutdown()，可以明确告诉生产者停止入队，并让消费者在处理完已有任务后收到 QueueShutDown。旧版本没有这个 API，仍需要哨兵值或取消任务。

推荐把正常收尾写成“停止生产 → 软关闭队列 → 等待 join() → 等待消费者任务退出”。默认的 shutdown() 会保留已入队任务；只有放弃剩余任务时才考虑 immediate=True。

先确认 Queue 的版本边界

Queue.shutdown() 和 asyncio.QueueShutDown 是 Python 3.13 新增的能力。软关闭后，新的 put() 会抛出 QueueShutDown，已经在队列里的项目仍可通过 get() 取出；队列排空后，消费者下一次 get() 才会收到关闭异常。Python 3.12 及更早版本不能直接照搬这段异常处理。

用软关闭让消费者自然结束

下面的示例把生产和消费拆成独立协程。生产者结束后调用 queue.shutdown()，主协程等待 queue.join() 确认每个项目都对应过一次 task_done()，再等待消费者任务结束。代码里的 finally 很重要：即使处理某个项目时报错，也不能让未完成任务计数永久卡住。

import asyncio

async def worker(queue: asyncio.Queue) -> None:
    while True:
        try:
            item = await queue.get()
        except asyncio.QueueShutDown:
            # 队列已经排空，消费者收到明确的结束信号
            return

        try:
            # 这里替换为真正的业务处理逻辑
            await asyncio.sleep(0)
            print(f"processed: {item}")
        finally:
            # 每次 get() 都必须配对 task_done()，否则 join() 无法结束
            queue.task_done()

async def main() -> None:
    queue = asyncio.Queue(maxsize=100)
    consumer = asyncio.create_task(worker(queue))

    try:
        for item in range(5):
            # 生产阶段结束前允许 put；关闭后新的 put 会失败
            await queue.put(item)
    finally:
        # 默认 immediate=False，保留队列里已经接收的项目
        queue.shutdown()

    await queue.join()
    await consumer

asyncio.run(main())

这里的顺序不是装饰性的：如果先等待消费者退出，消费者会继续等待一个永远不会到来的项目；如果不等待 join() 就直接取消消费者，队列可能还有未处理任务。生产者与关闭动作之间若存在多个生产协程，应由统一协调者在确认它们都结束后再调用 shutdown()。

为什么不应随意使用 immediate=True

queue.shutdown(immediate=True) 会立即排空队列，并唤醒等待中的消费者；它适合进程已经进入放弃剩余工作的故障路径，不适合普通的优雅停机。官方文档特别提醒，这种模式可能让 join() 在任务还没有真正处理完成时解除阻塞，破坏“每个入队项目都已完成”的常规约束。

如果处理逻辑需要保留审计、重试或补偿信息，应使用默认软关闭。只有在明确接受丢弃剩余项目、并且上层另有恢复机制时，才使用立即关闭；同时把未处理数量和关闭原因写进应用日志。

旧版本怎样保留同样的收尾语义

在 Python 3.12 及更早版本，可以放入一个唯一哨兵对象。消费者拿到它后退出循环，但不要把哨兵当作普通业务项目处理。消费者数量大于一个时，通常要放入同样数量的哨兵，或者由第一个退出的消费者把哨兵再次放回队列；选择哪一种取决于队列是否还会被其它生产者使用。

SENTINEL = object()

async def legacy_worker(queue: asyncio.Queue) -> None:
    while True:
        item = await queue.get()
        try:
            if item is SENTINEL:
                # 旧版本没有 QueueShutDown，只能用约定值通知退出
                return
            await handle(item)
        finally:
            # 哨兵同样要完成 task_done()，否则 join() 会一直等待
            queue.task_done()

迁移到 3.13+ 时，不要只把哨兵替换成异常名。还要检查生产者是否可能在关闭后继续 put()、是否有协程仍持有队列引用，以及取消消费者时是否会漏掉 task_done()。这些才是“消费者结束了但主任务仍不退出”的常见来源。

发布前的回归清单

• 软关闭后，新的 put() 是否被捕获并转成明确的停止状态。
• 每个成功 get() 是否严格对应一次 task_done()。
• 是否先等待 join()，再等待消费者任务，而不是反过来取消。
• 是否只在允许丢弃剩余任务时使用 immediate=True。
• 目标环境低于 Python 3.13 时，是否保留哨兵或取消任务的兼容分支。

这样改造后，消费者的结束不再依赖“队列暂时为空”这种不稳定判断，而是由版本明确支持的关闭信号决定；正常停机能排空任务，紧急停机也有清晰的丢弃边界。

相关问题

调用 shutdown() 后还能继续 put() 吗？不能。新的 put() 和等待中的生产者都会收到 QueueShutDown。

为什么 queue.join() 一直不返回？通常是某个消费者取到项目后没有调用 task_done()，或者异常路径绕过了清理代码。

多个消费者需要几个哨兵？旧版本通常按消费者数量投放哨兵，避免仍在等待的消费者永远收不到结束信号。