Python queue.Queue.shutdown 怎样让生产者消费者有序结束

在线程生产者—消费者程序里，真正麻烦的往往不是把任务放进 queue.Queue，而是程序准备停止时，怎样让生产者不再塞入新任务、消费者处理完已经入队的任务，并且不会永远卡在 get() 上。Python 3.13 起，Queue.shutdown() 给出了一个明确的关闭协议。

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

默认调用 shutdown() 表示“停止增长、继续排空”；只有不需要保留剩余任务时，才考虑 shutdown(immediate=True)。消费者应捕获 queue.ShutDown，而不是用一个普通任务值猜测队列是否结束。

Queue.shutdown() 解决的是什么问题

传统写法常用哨兵对象表示结束：生产者放入一个特殊值，消费者取到后退出。这种方式在多个消费者、多个生产者或队列已经接近满载时，需要额外约定哨兵数量、投递时机和阻塞处理。shutdown() 把“队列不再接收新任务”变成了队列本身的状态。

默认参数是 immediate=False。调用后，新的 put() 会失败，已经阻塞在 put() 的生产者会被唤醒并收到 ShutDown；队列中已有的任务仍可以被 get() 取出。等队列变空，后续消费者再调用 get() 时也会收到 ShutDown。

默认 shutdown() 为什么适合有序收尾

下面的例子把关闭动作放在主线程中。代码中的生产者只负责提交任务，消费者在每个成功取出的任务上调用 task_done()，主线程用 join() 等待已提交任务完成。

import queue
import threading
import time

tasks = queue.Queue(maxsize=2)

def worker():
    while True:
        try:
            item = tasks.get()
        except queue.ShutDown:
            # 队列已经关闭且没有剩余任务，消费者可以退出。
            break
        try:
            print(f"处理 {item}")
            time.sleep(0.02)
        finally:
            # 每次 get() 都必须对应一次 task_done()。
            tasks.task_done()

def producer():
    for item in range(5):
        try:
            # 队列关闭或阻塞中的 put() 被唤醒时会抛出 ShutDown。
            tasks.put(item)
        except queue.ShutDown:
            break

threading.Thread(target=worker).start()
threading.Thread(target=producer).start()

# 真实程序通常在收到停止信号后执行这一句，阻止队列继续增长。
tasks.shutdown()
tasks.join()

这段示例的关键顺序是：先让生产者停止投递，再关闭队列，最后等待任务计数归零。实际程序中生产者可能仍在另一个线程里运行，因此应由协调代码保证它不会在关闭之后继续创建新的业务任务；即使它已经阻塞在 put()，关闭操作也会让它走到 ShutDown 分支。

生产者和消费者都要处理 ShutDown

ShutDown 不是只给消费者准备的异常。生产者可能在队列已满时阻塞于 put()，调用 shutdown() 后它会被唤醒。生产者捕获异常后应停止生成或转发任务，而不是立即重试同一个队列。

消费者侧也不要只捕获 queue.Empty。Empty 表示当前没有任务，队列仍可能在未来接收任务；ShutDown 表示队列已经进入关闭协议。两者的业务动作不同：前者通常继续等待，后者通常退出工作循环。

immediate=True 为什么要谨慎

tasks.shutdown(immediate=True) 会立即排空队列，并唤醒阻塞的消费者。它适合进程必须快速终止、队列中的任务已经没有继续处理价值的场景，但它改变了默认的“每个入队任务最终对应 task_done()”节奏。

官方文档特别提醒：立即关闭可能让 join() 在任务尚未真正处理完成时解除阻塞，从而破坏通常的不变量。因此不要把它当作普通优雅退出的简写。若任务涉及写文件、提交订单、发送消息或释放外部资源，应优先使用默认模式，并把业务取消设计在任务处理函数内部。

上线前检查版本和关闭顺序

1. 确认运行环境至少是 Python 3.13，因为 Queue.shutdown() 和 queue.ShutDown 是 3.13 增加的能力。
2. 先停止生产端的业务循环，再调用队列的默认 shutdown()，避免关闭动作和新任务生成互相竞速。
3. 每个成功的 get() 都在处理结束后调用一次 task_done()，异常路径也不要漏掉。
4. 让消费者捕获 ShutDown 并退出；不要用 qsize() 判断“马上结束”，它只是近似值。
5. 只有明确接受丢弃剩余任务和 join() 语义变化时，才使用 immediate=True。

几个容易混淆的问题

调用 shutdown() 后，队列里的任务会马上消失吗？

默认不会。immediate=False 会保留已经放入队列的任务，让消费者继续取出；只有立即模式才会清空剩余任务。

为什么消费者不能只靠 qsize() 判断结束？

qsize() 只是近似观察，不能保证下一次 get() 不阻塞，也不能表达队列是否已经关闭。关闭协议应通过 ShutDown 表达。

没有升级到 Python 3.13 怎么办？

旧版本没有这组 API，不能直接照搬示例。可以继续使用哨兵、事件或条件变量实现退出协议，但要明确多个消费者的哨兵数量和阻塞线程的唤醒规则。

把 shutdown() 理解成“关闭队列的接收能力”，把 join() 理解成“等待任务计数归零”，两者配合起来就能把生产者、消费者和主线程的收尾职责分开。默认模式负责有序排空，立即模式负责快速中止，选择哪一个应由任务是否允许丢弃来决定。