Python queue.ShutDown怎样让消费者感知队列关闭

Python 3.13及以上版本可以由管理线程调用Queue.shutdown()，让消费者在get()处捕获queue.ShutDown并退出。默认关闭会保留已经入队的任务，消费者取完后才收到关闭异常；不要把它与表示暂时取不到数据的queue.Empty混在一起。

写线程队列的停止逻辑时，我更愿意把“没有活干”和“不再接活”分开表达。下面用一个文档编号处理器说明这种取舍：队列承接编号，工作线程处理任务，只有管理线程负责关闭。示例是教学设计，文中的状态说明不是实际运行日志。

官方文档：https://docs.python.org/3/library/queue.html

先约定停机时要不要做完队列里的任务

我的默认选择是正常排空：停止接收新任务，处理已接受的任务，再回收消费者线程。这个约定适合允许等待的批处理。如果业务明确允许放弃尚未领取的工作，才考虑立即关闭。两种模式都不能代替业务任务本身的取消设计。

shutdown()和ShutDown从Python 3.13加入。这里讨论的是标准库queue.Queue在线程之间的协作；不能把相同方法名直接套到多进程队列、异步队列或SimpleQueue上。部署环境仍在3.12或更早时，应继续使用自己定义并完整实现的停止协议。

队列关闭之后不能再增长，后续put()会抛出关闭异常，原先因容量不足而等待的生产者也会被唤醒并收到该异常。这一点让我倾向于把关闭权放在管理层：普通任务处理函数只消费任务，不能因为自己处理完一项就关闭全局队列。

把ShutDown捕获放在get的边界

消费者应先尝试领取，再分别处理任务。领取失败意味着没有拿到任务，因此不能给这次失败调用补一次task_done()。如果把整个循环都放进同一个宽泛异常捕获里，很容易把正常停止误记成任务失败，或者把真正的处理异常吞掉。

import queue

def consume(tasks, handle):
    while True:
        try:
            item = tasks.get()
        except queue.ShutDown:
            # 关闭是队列生命周期事件，不是业务任务失败。
            return
        try:
            handle(item)
        finally:
            # 只有成功领取的任务才确认；业务失败记录由调用方设计。
            tasks.task_done()

这个片段只展示捕获位置；如果handle抛出异常，它仍会退出当前线程。生产使用时还需要像下一节那样明确处理失败，避免消费者全部提前退出后，剩余任务无人领取。正常关闭且队列为空时，阻塞在领取位置的消费者能够获知关闭；不必额外塞入一个与业务编号混用的特殊值。

用完整例子连接排空、任务确认和线程回收

下面把处理失败收集到线程安全队列里，消费者遇到单项错误后继续服务。为便于解释，任务只有三个文档编号，处理函数只检查编号。实际项目可以替换处理函数，但需要另外定义持久化失败记录和重试策略。

import logging
import queue
import threading

logging.basicConfig(level=logging.INFO, format="%(threadName)s %(message)s")
tasks = queue.Queue(maxsize=16)
failures = queue.SimpleQueue()

def process(document_id):
    # 教学任务：实际业务在此处理文档，并自行管理外部资源超时。
    if not document_id.startswith("doc-"):
        raise ValueError("文档编号格式不正确")

def worker():
    while True:
        try:
            document_id = tasks.get()
        except queue.ShutDown:
            logging.info("消费者获知队列关闭")
            return
        try:
            process(document_id)
        except Exception as exc:
            # 单项错误单独留存，不把它误当作关闭信号。
            failures.put((document_id, type(exc).__name__))
            logging.error("任务处理失败，错误类型=%s", type(exc).__name__)
        finally:
            # 这里确认的是本次处理结束，业务是否成功由失败记录判断。
            tasks.task_done()

workers = [threading.Thread(target=worker, name=f"consumer-{i}") for i in range(2)]
for thread in workers:
    thread.start()

try:
    for document_id in ("doc-alpha", "doc-beta", "doc-gamma"):
        tasks.put(document_id)
finally:
    # 这个教学例子只有一个生产者，由管理线程统一封闭入口。
    tasks.shutdown()
    tasks.join()
    for thread in workers:
        thread.join()

failed_items = []
while True:
    try:
        failed_items.append(failures.get_nowait())
    except queue.Empty:
        break
if failed_items:
    raise RuntimeError(f"存在{len(failed_items)}项处理失败，请进入业务补偿")

我会把tasks.join()读作“等待任务确认计数归零”，把thread.join()读作“等待线程退出”。这两个等待对象不同，不能删掉其中一个就假定另一个也完成了。上面没有打印固定顺序的预期日志，因为两个消费者的领取和记录可以交错。

同样，task_done()没有检查数据库提交，也没有证明文件已经保存。例子在业务异常后仍然确认任务，是为了让本次消费尝试结束，并由失败集合决定最终批次结果。如果需求是直到成功才算完成，就应引入明确的重试次数和失败归档，不能简单省略确认，否则管理线程可能一直等待。

关闭权限和业务失败日志要分开设计

在这个方案里，管理线程是生命周期的唯一拥有者。多个生产者存在时，可以先禁止外部提交，再等待生产者结束，最后正常关闭队列；如果选择先关闭来打断阻塞生产者，则生产者必须捕获ShutDown，把尚未成功提交的任务标记为拒绝或交回调用方。

我不建议遇到关闭异常后无条件循环重投。同一个已关闭实例不会因为重试而重新接收任务，这只会让停止路径变成错误日志风暴。需要恢复接收时，应由上层重新建立一组队列和工作线程，并明确旧任务归属。

日志至少区分“入口已关闭”“任务处理失败”“消费者已退出”和“批次存在待补偿任务”。编号、异常类型和批次号足以帮助定位的场景，不要把整份文档或凭据写入日志。这个日志建议来自示例的职责划分，并不意味着标准库会自动生成这些记录。

immediate=True为什么不能当成任务成功的捷径

立即关闭会清空队列中尚未领取的项目，并按被清除的数量减少未完成任务计数。因此join()可能在这些工作根本没被处理的情况下返回。已经被消费者拿走的任务不会因此被强制中断，它们仍要按自己的处理逻辑结束并完成确认。

对我来说，立即关闭适合“明确放弃排队工作”的故障处置；正常部署停机更适合默认排空。任务处理函数若卡在网络或文件操作上，队列关闭不能解除那段阻塞，必须在相关操作上设置业务允许的超时或协作取消。

上线接入时，我会关注的几个问题

为什么不先判断empty再退出？队列快照无法为下一次操作提供并发保证，而且空队列仍可能继续收到任务。停止应由生命周期协议决定，不能靠某一刻的长度推断。

收到ShutDown后还要task_done吗？没有领取到任务就不要确认；成功领取的项目则在处理结束时确认一次。多次确认可能触发计数错误。

消费者已经退出，join还不返回怎么办？检查是否有成功领取后漏确认的项目，以及是否因处理异常导致所有工作线程提前退出。上面完整示例把单项业务失败与关闭事件分开，正是为了让这两类问题可追踪。

这套写法适合谁？如果项目已使用Python 3.13以上的线程Queue，且停机目标是把已接受工作做完，我会优先采用默认shutdown。需要跨进程持久消息、任务可靠投递或强制取消正在执行的操作时，还需要独立的系统设计。