Python asyncio.Queue.shutdown 怎样让生产者和消费者安全收尾

当异步消费者需要停机时，真正麻烦的不是让任务退出，而是让生产者、队列和消费者在同一个边界上收尾。Python 3.13 起，asyncio.Queue.shutdown() 提供了明确的关闭语义：默认的 shutdown(False) 停止继续入队，但保留已经入队的工作；shutdown(True) 则立即清空队列，适合紧急终止，不适合普通优雅退出。

官方地址：https://docs.python.org/3/library/asyncio-queue.html

下面用一个生产者和多个消费者的场景说明如何处理 QueueShutDown、task_done() 与 join()。配图均为静态说明图，不是运行截图或运行证据。

为什么要先关生产者再排空队列

shutdown(False) 的核心含义是“队列不再增长”。之后新的 put() 和 put_nowait() 会抛出 QueueShutDown，正在等待空位的生产者也会被唤醒并收到同样的异常；消费者仍然可以通过 get() 取走关闭前已经放入的任务。

因此生产协程不要把关闭当成普通写入错误吞掉，而应把它当作退出信号。消费者则要继续处理队列里的存量任务，直到队列已经关闭且为空时，get() 抛出 QueueShutDown。

生产者和消费者怎样共同处理 QueueShutDown

一个可复用的收尾形态如下。示例中的中文注释说明了关闭边界、异常处理和任务计数的目的：

import asyncio


async def producer(queue):
    try:
        for item in range(10):
            # 生产协程只负责投递；队列关闭后让异常成为退出信号。
            await queue.put(item)
    except asyncio.QueueShutDown:
        # 关闭可能唤醒正在等待空位的生产者，这里正常结束生产。
        return


async def worker(name, queue):
    while True:
        try:
            # 队列未关闭且为空时，消费者会在这里等待新任务。
            item = await queue.get()
        except asyncio.QueueShutDown:
            # 关闭且已经没有存量任务，当前消费者安全退出。
            return

        try:
            # 这里替换成真实业务处理，不要把 task_done 放到 get 之前。
            await asyncio.sleep(0)
        finally:
            # 每次成功 get 都必须对应一次 task_done，join 才能正确结束。
            queue.task_done()


async def graceful_close():
    queue = asyncio.Queue(maxsize=100)
    producer_task = asyncio.create_task(producer(queue))
    workers = [asyncio.create_task(worker(f"worker-{i}", queue)) for i in range(3)]

    await producer_task
    # 停止继续增长，但允许消费者处理关闭前已经入队的任务。
    queue.shutdown()
    await queue.join()
    # 关闭且排空后，等待消费者捕获 QueueShutDown 并退出。
    await asyncio.gather(*workers)


asyncio.run(graceful_close())

这个顺序里有两个容易漏掉的点。第一，生产者结束后才调用 shutdown()，否则仍在生产的协程会收到关闭异常；第二，消费者必须在 finally 中调用 task_done()，否则某个任务中途出错时，join() 可能永久等待。若业务错误需要重试，应先明确任务是否仍算当前这次取出的工作，再决定何时调用 task_done()。

join、task_done 与 immediate 怎么选

join() 等待的是未完成任务计数归零，不是简单等待队列长度变成零。每次成功 put() 会增加计数，每次对应的 task_done() 会减少计数，所以正常收尾应该让 shutdown(False)、消费者处理和 join() 配合工作。

shutdown(True) 会把队列排空，并可能让等待中的 join() 提前解除，即使那些任务并没有真正完成。它适用于进程即将退出、任务已经没有继续处理价值等紧急路径；不要把它当成默认的优雅停机方式。

常见误区与兼容边界

如果把 shutdown(False) 写在生产者仍会继续调用 put() 的位置，生产者会在关闭边界收到 QueueShutDown；这不是异常故障，而是它应该停止工作的信号。反过来，如果消费者只捕获 QueueEmpty，就无法识别关闭后的正常退出。

还要注意，asyncio.Queue 面向异步代码，并不是线程安全队列；它的操作也没有内置 timeout 参数，需要超时控制时应在外层使用 asyncio.wait_for()。而 Queue.shutdown() 与 QueueShutDown 是 Python 3.13 新增能力，旧版本不能只改一个方法名就获得同样语义。

相关问题

• 为什么调用 join 后消费者还没有退出？ 因为 join 只等待未完成任务计数归零；要唤醒空队列中的消费者，还需要关闭队列并让它捕获 QueueShutDown。
• 什么时候可以用 immediate=True？ 只有明确接受丢弃存量任务或破坏 join 完成语义时，才把它作为紧急终止路径。
• task_done 调用多了会怎样？ 如果调用次数超过已经入队的任务数，队列会抛出 ValueError，因此应保证每次成功 get 恰好对应一次 task_done。

把队列收尾拆成“停止生产、处理存量、等待计数归零、让消费者退出”四个边界，就能看清 shutdown(False) 为什么适合普通停机，而 shutdown(True) 只适合紧急场景。