Python asyncio.Queue shutdown 后等待者会收到什么

平时写Python异步协程逻辑，用asyncio.Queue做跨协程生产消费通信的开发者，经常会碰到队列主动调用shutdown方法后，还在队列上挂起等待的协程直接抛出异常的情况，此时所有还在队列上等待put、get操作的协程，都会直接抛出asyncio.exceptions.QueueShutDown异常，不会再继续阻塞等待。

对调用过shutdown的asyncio.Queue执行任何put、get操作，或是还有协程正挂起在这两个方法上等待，都会直接抛出QueueShutDown异常，不会再返回原有业务数据。

我第一次把生产者和消费者一起停掉时，最容易误判的是“调用了 shutdown()，所有等待中的协程都会立刻拿到同一种结果”。实际要看等待者是谁，以及是否传入了 immediate=True。在 Python 3.13 及更高版本里，默认的 shutdown(False) 会停止继续入队，但允许消费者把已有项目处理完；只有队列排空后，阻塞在 get() 上的协程才会收到 asyncio.QueueShutDown。立即关闭则会把队列清空，阻塞的 getter 直接收到这个异常。

先把三类等待者分开看

shutdown() 不是取消所有任务的快捷方式，它改变的是队列的生命周期状态。生产者通常卡在 put()，消费者卡在 get()，协调方则可能等待 join()。三者的完成条件不同，不能只看协程是否从 await 返回。

默认关闭会保留已入队任务

温和关闭适合“停止接收新任务，但不丢掉已接收任务”的场景。调用后，新的 put() 立即失败；已经阻塞的生产者也会被唤醒。消费者仍可以通过 get() 取出队列里剩下的项目，处理完成后照常调用 task_done()。当最后一个项目取走，之后再调用 get() 才会收到 QueueShutDown。

import asyncio

async def worker(queue: asyncio.Queue):
    while True:
        try:
            # 先消费已有项目，空队列且已关闭时才会抛异常
            item = await queue.get()
            try:
                await handle(item)
            finally:
                # 只有真正取到的项目才对应一次完成确认
                queue.task_done()
        except asyncio.QueueShutDown:
            # 默认关闭下，这里表示队列已经排空，可以退出消费者
            return

这里的关键不是捕获异常本身，而是把 task_done() 放在已经成功 get() 的项目范围内。这样 join() 才能准确等待处理计数归零。

立即关闭会牺牲剩余任务的完成语义

传入 immediate=True 后，队列会被排空，阻塞在 get() 上的消费者会因队列已经为空而收到 QueueShutDown。这适合进程即将退出、剩余任务已经不值得继续处理的场景，不适合需要保证消息不丢失的优雅停机。

async def stop_workers(queue: asyncio.Queue, immediate: bool):
    # 关闭后不再接受新项目，阻塞中的 put() 会收到 QueueShutDown
    queue.shutdown(immediate=immediate)
    if not immediate:
        # 温和关闭仍等待已取项目完成，保持 join 的正常含义
        await queue.join()
    else:
        # 立即关闭只代表队列已终止，不代表剩余项目已处理完成
        return

尤其要注意：立即关闭可能让 join() 在剩余工作尚未完成时解除。因此停机代码应把“队列已终止”和“业务任务已完成”记录成两个不同状态。

我会这样安排生产者和消费者收尾

需要保留任务时，先调用默认的 shutdown()，等待消费者自然排空，再等待 join()；生产者统一捕获 QueueShutDown，把它当成停止投递信号，而不是故障报警。只有明确接受丢弃剩余项目时，才使用 immediate=True，并在日志里单独记录被放弃的数量或业务补偿动作。

• 运行环境低于 Python 3.13 时，不能直接假设存在 Queue.shutdown() 和 QueueShutDown。
• 消费者拿到项目后，即使业务处理抛错，也要决定是否调用 task_done()，避免 join() 永久等待。
• 不要用 immediate=True 的 join() 返回值证明所有项目都成功处理。

相关问题

shutdown 后还能 put 新项目吗？

不能。新的 put() 和已经阻塞的 put() 都会以 QueueShutDown 结束。

默认 shutdown 会马上让所有 get 失败吗？

不会。只要队列里还有项目，消费者仍能取走它们；队列变空后，后续或阻塞中的 get() 才结束为关闭异常。

什么时候可以用 immediate=True？

当剩余项目可以丢弃或由其他补偿机制接管时可以使用；对需要逐条确认的任务，应优先选择默认关闭。