Python asyncio.Queue.shutdown() 如何安全停机：QueueShutDown、join 与 immediate 边界

异步报表服务停机时，最容易留下的不是一个报错，而是几条还在等待的生产任务：消费者已经退出，生产者却继续往 asyncio.Queue 里写。Python 3.13 提供了 Queue.shutdown() 和 QueueShutDown，可以把“停止接单”和“处理完存量任务”拆成两个明确阶段。

正常下线优先使用 queue.shutdown()，让消费者取完存量任务后以 QueueShutDown 收尾；只有明确放弃剩余工作时，才考虑 immediate=True。

先把停机现场固定下来

下面用“报表任务”模拟一个有生产者和消费者的队列。生产者每次放入一个任务，消费者处理完成后必须调用 task_done()。这个配对关系决定了 join() 是否真的代表“工作完成”，不能只看 qsize()。

import asyncio

async def producer(queue):
    for job_id in range(5):
        await queue.put(f"report-{job_id}")
        await asyncio.sleep(0.02)

async def worker(queue):
    while True:
        job = await queue.get()
        try:
            await asyncio.sleep(0.05)
            print("done", job)
        finally:
            queue.task_done()

如果服务只取消 worker，没有通知 producer，停机逻辑就缺了“禁止继续入队”这一步。队列看起来变空了，实际上可能还有生产协程挂在 put() 上。

用 shutdown() 先停止接单，再排空存量

平滑停机的顺序是：先让生产者结束或收到 QueueShutDown，再等待消费者把已经入队的任务处理完，最后让消费者从空队列退出。shutdown() 本身是同步方法，调用后不会等待消费者，因此仍要显式等待 join()。

async def graceful_stop(queue, producer_task, workers):
    queue.shutdown()

    # 生产者若正在 put，会在这里以 QueueShutDown 结束
    await asyncio.gather(producer_task, return_exceptions=True)

    # 只要每次 get() 都对应 task_done()，这里才表示存量任务已完成
    await queue.join()

    # join 完成后，空队列上的 get() 会收到 QueueShutDown
    await asyncio.gather(*workers, return_exceptions=True)

消费者需要把 QueueShutDown 当作正常收尾信号，而不是把它记成业务失败：

async def worker(queue):
    while True:
        try:
            job = await queue.get()
        except asyncio.QueueShutDown:
            return
        try:
            await handle_report(job)
        finally:
            queue.task_done()

这里的关键不是“捕获异常”本身，而是让异常发生在队列已经停止增长之后。否则一个生产者可能在消费者退出前再次提交任务，停机边界会重新变得模糊。

join() 为什么不能替代 shutdown()

join() 只关心未完成任务计数：每次 put() 增加计数，每次 task_done() 减少计数。它不会阻止新的生产者继续提交，也不会告诉消费者什么时候应该退出。

如果消费者在业务异常后直接跳出，却没有在 finally 中调用 task_done()，join() 会一直等待。相反，重复调用会抛出 ValueError。生产环境可以把“取任务、处理、确认”保持在同一个 try/finally 里，减少收尾遗漏。

immediate=True 只用于明确的中止路径

queue.shutdown(immediate=True) 会立即排空队列，阻塞在 get() 的消费者会收到 QueueShutDown。官方文档特别提醒：它可能让 join() 在工作尚未完成时解除，因此不能把它当成普通的“更快优雅停机”。

async def abort_now(queue, workers):
    queue.shutdown(immediate=True)
    await asyncio.gather(*workers, return_exceptions=True)

这个分支适合进程即将被强制终止、任务本身已失去业务价值等场景。若任务需要补偿、重试或写入审计记录，应优先让队列走默认的排空流程，并把未完成任务交给可恢复的外部存储。

兼容 Python 3.12 及更早版本

Queue.shutdown() 是 Python 3.13 新增能力。项目需要兼容旧版本时，不要只在运行时捕获 AttributeError 就假设停机语义完全一致：旧版本没有标准的 QueueShutDown，通常需要使用哨兵对象、取消任务或自定义停止事件，并为重复停止、生产者阻塞和消费者退出分别写测试。

import sys

if sys.version_info >= (3, 13):
    queue.shutdown()
else:
    # 旧版本保留自己的哨兵/取消策略，不混用新异常名
    stop_event.set()

停机前的四个检查点

• 先停止生产入口，再调用 shutdown()，避免新任务和停机动作交错。
• 消费者的 task_done() 放在 finally 中，确保业务异常不会卡死 join()。
• 平滑模式下检查 join() 真正结束后再收消费者，不用“队列为空”代替完成确认。
• 中止模式记录被丢弃任务的数量和原因，不能把 immediate=True 的返回当作成功处理证明。

常见问题

shutdown() 会自动取消所有消费者吗？

不会。它改变队列状态，并让相关的 put()、get() 在规定条件下收到 QueueShutDown；消费者协程仍需自己捕获并退出。

调用 shutdown() 后还能处理已有任务吗？

默认可以。消费者可以继续取出已有项目，直到队列清空；每个项目都要调用 task_done()，这样 join() 才能按正常计数结束。

什么时候应该使用 immediate=True？

只有剩余任务明确不再需要、服务必须立即中止时才使用。它会破坏“join 等于全部工作完成”的常规直觉，调用前应保存必要的丢弃记录。

把停机语义写进测试

一个可维护的测试至少要覆盖三条路径：默认关闭后新 put() 收到 QueueShutDown；已有任务完成后 join() 返回；立即关闭不会被误判为业务任务已经完成。这样升级 Python 或调整消费者数量时，停机行为才有可回归的边界。