Python queue.Queue 怎样结束生产消费循环

Python 的 queue.Queue 生产消费循环，不能用“队列现在为空”来判断系统已经结束。正确的关闭协议要把三件事分开：生产者不再提交新任务、消费者把已经提交的任务处理完、消费者线程退出。Python 3.12 及更早版本通常用 sentinel（停止标记）完成这件事；Python 3.13 及更高版本可以使用 Queue.shutdown()，让队列本身进入关闭状态。

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

如果目标是“正常收尾”，优先使用 graceful 语义：生产者先停止，消费者继续取完已有任务，所有已取任务都调用 task_done()，控制线程再等待 join() 并回收工作线程。只有明确要放弃排队任务时，才考虑立即关闭。

先把三个结束条件拆开

Queue.put() 会增加未完成任务计数，消费者调用 task_done() 才会减少这个计数；join() 等待的是计数归零，而不是某一瞬间的队列长度为零。消费者刚取走最后一个元素时，任务可能仍在处理，队列也可能马上被另一个生产者重新填入。

因此，empty() 和 qsize() 只能作为近似观察值，不能作为并发关闭的同步协议。一个线程看到空队列，并不代表另一个生产者不会马上提交任务，也不代表消费者已经完成最后一项工作。

Python 3.12 及更早版本：每个消费者一个 sentinel

旧版本没有 Queue.shutdown()，最稳妥的兼容方式是把一个特殊对象放进队列，消费者取到它就退出。停止标记必须使用独立对象，而不是容易与正常业务数据冲突的字符串；如果有多个消费者，通常要放入同样数量的停止标记。

import queue
import threading

SENTINEL = object()

def worker(tasks: queue.Queue) -> None:
    while True:
        item = tasks.get()
        try:
            if item is SENTINEL:
                # 停止标记也占用一个未完成任务计数，必须确认后再退出。
                return
            process_item(item)
        finally:
            # 正常任务和停止标记都只对应一次 task_done()。
            tasks.task_done()

def process_item(item: object) -> None:
    # 这里替换成实际业务；示例不依赖外部服务或真实数据。
    print(f"处理任务: {item}")

tasks = queue.Queue()
workers = [
    threading.Thread(target=worker, args=(tasks,))
    for _ in range(3)
]
for thread in workers:
    thread.start()

for item in range(10):
    # 所有真实任务先于停止标记提交，保持 FIFO 的收尾顺序。
    tasks.put(item)

for _ in workers:
    # 每个消费者都需要拿到一个停止标记。
    tasks.put(SENTINEL)

# join 等待真实任务和 sentinel 都完成 task_done()。
tasks.join()
for thread in workers:
    # join 之后线程应已从 worker 返回，这里完成线程资源回收。
    thread.join()

这段代码里最容易漏掉的是停止分支的 task_done()。如果拿到 sentinel 后直接 return，队列的未完成任务计数会永远多一个，控制线程就会卡在 tasks.join()。把 task_done() 放在 finally 中，可以同时覆盖正常任务和退出标记。

停止标记应在所有生产者都停止投递后再加入。如果生产者仍在工作，先加入 sentinel 可能让消费者提前退出，后续真实任务就没人处理。高并发场景还要明确“生产者全部结束”的协调点，例如等待生产线程的 thread.join() 后，再统一投递 sentinel。

Python 3.13 及更高版本：用 Queue.shutdown 关闭队列

从 Python 3.13 开始，Queue.shutdown(immediate=False) 可以表达“队列不再接受新任务，但允许消费者把已经入队的任务取完”。之后的 put() 会抛出 queue.ShutDown；当队列排空后，阻塞中的 get() 也会因关闭而结束。这样，退出信号不再伪装成业务队列中的一项数据。

import queue
import threading

def worker(tasks: queue.Queue) -> None:
    while True:
        try:
            item = tasks.get()
        except queue.ShutDown:
            # graceful shutdown 且队列已经排空，消费者可以退出。
            return
        try:
            process_item(item)
        finally:
            # get 成功一次，就必须对应一次 task_done()。
            tasks.task_done()

tasks = queue.Queue()
workers = [threading.Thread(target=worker, args=(tasks,)) for _ in range(3)]
for thread in workers:
    thread.start()

for item in range(10):
    # 生产阶段结束前，仍允许正常 put。
    tasks.put(item)

# 先保证没有生产者会继续 put，再进入默认的优雅关闭。
tasks.shutdown()
tasks.join()
for thread in workers:
    # join 等待线程观察到 ShutDown 并返回。
    thread.join()

这里的关键不是把 shutdown() 当作“立刻杀死线程”，而是把它当作队列边界：关闭后不能增长，已有任务仍按正常方式消费。消费者仍要在处理任务后调用 task_done()，因为 graceful shutdown 并不会替代任务完成确认。

如果工程需要同时兼容 3.12 和 3.13，可以在能力探测后选择 sentinel 分支，或者把队列关闭协议封装成一个小适配层。不要在旧版本中直接引用不存在的 queue.ShutDown，也不要只捕获关闭异常而跳过已经成功取出的任务的 task_done()。

高并发场景的关闭顺序：先收生产，再关队列

在压力较高的批处理服务里，真正的瓶颈往往不是消费者循环，而是“谁有权宣布不会再有新任务”。可以把关闭协议分成一个控制层和一个工作层：控制层等待所有生产者完成，工作层只负责取任务、处理任务和确认任务完成。

推荐的协议顺序如下：

1. 停止或等待所有生产者，不再允许新的 put()。
2. 旧版本向队列放入与消费者数量匹配的 sentinel；3.13+ 调用默认的 shutdown()。
3. 消费者处理每个成功取出的任务，并在 finally 中调用一次 task_done()。
4. 控制线程调用 queue.join()，等待未完成任务计数归零。
5. 最后调用工作线程的 join()，确认消费者本身已经退出。

这套顺序把“任务全部完成”和“线程已经退出”分开了。前者由队列的未完成任务计数表达，后者由线程对象的 join() 表达；混用两种 join，排查卡死时就会很困难。

为什么不建议用 immediate=True 做常规收尾

shutdown(immediate=True) 是硬关闭语义：队列会被排空，阻塞中的消费者会收到 ShutDown。官方文档特别提醒，立即关闭可能让 join() 在任务尚未真正完成时解除阻塞，破坏“每个入队任务都完成确认”的通常不变量。

因此它更适合取消、超时或进程即将终止的场景，而不是普通批次的成功收尾。使用硬关闭时，业务层需要另外记录哪些任务已取出、哪些任务被放弃、是否可重试；不能把 join() 返回简单解释成“所有任务都处理成功”。

卡死与提前退出的排查表

小结：把关闭协议写成接口约定

如果需要兼容旧 Python，采用“生产者停止后、每个消费者一个 sentinel、sentinel 也调用 task_done”的方案；如果最低版本是 Python 3.13，则可以采用 shutdown() 加 queue.ShutDown，让队列状态本身表达关闭。无论选择哪种方式，task_done() 都只对应成功的一次 get()，queue.join() 等的是任务完成，线程 join() 等的是线程退出。

真正稳定的生产消费系统，不是让线程“尽快停掉”，而是让停止投递、完成已有任务、允许消费者退出这三个状态可观察、可复盘。把这三个状态写进控制层的关闭协议，才能避免偶发卡死和静默丢任务。

一个 sentinel 够所有消费者使用吗？

通常不够。若每个消费者都需要独立退出，最简单的是放入与消费者数量相同的 sentinel。也可以设计“取到 sentinel 后重新放回”的协议，但那会让停止标记的计数和确认规则更复杂，必须单独证明不会重复确认或让 queue.join() 永远等待。

调用 shutdown 后还需要 task_done 吗？

需要。shutdown 只改变队列是否继续接受任务以及消费者何时因关闭而结束；已经成功 get() 的任务仍要在处理结束后调用一次 task_done()，这样 graceful 模式下的 queue.join() 才有明确的完成语义。

可以用 empty() 判断消费者该退出吗？

不建议。empty() 是并发观察值，不是生产者和消费者之间的关闭协议。只要存在并发生产者，就可能在判断之后再次入队；即使没有新任务，队列为空也不代表最后一个任务已经处理完成。

参考文档：https://docs.python.org/3/library/queue.html、https://docs.python.org/3.13/library/queue.html、https://docs.python.org/3.11/library/queue.html。示例中的任务、标识和数据均为原创演示内容。