题干与适用场景
一个服务使用 asyncio.Queue 分发任务给多个 worker。部署替换时要停止接收新任务，处理完已入队任务后退出；出现致命故障时又要立即唤醒阻塞的生产者和消费者。请使用 Python 3.13 的 Queue.shutdown() 设计两种停机路径。
官方文档说明：默认的 shutdown(immediate=False) 封口队列但允许消费者排空已有项目；immediate=True 会清空队列并打破通常的 join() 完成不变量。QueueShutDown 是生产和消费双方应识别的终止信号。
面试官考察点
看候选人能否区分“停止生产”和“取消消费”，正确维护 put、get、task_done 的计数；能否解释立即关闭为何不能当作任务已完成，并设计兼容 Python 3.12 的回退方案。
回答前需要澄清的问题
- 优雅停机是否必须处理完所有已接受任务？
- 紧急停机时，队列中的任务是否可丢弃，是否需要持久化补偿？
- 生产者来自同一事件循环，还是跨线程/进程？
- worker 是否有外部 I/O、重试和幂等要求？
- 部署环境最低 Python 版本是什么，能否直接使用 3.13 API？
30 秒回答框架
“优雅停机先调用 queue.shutdown()，阻止新的 put，让 worker 继续 get 并在每项完成后 task_done，最后等待 queue.join()，再取消空闲 worker。紧急停机使用 shutdown(immediate=True)，接受剩余任务被丢弃，等待方收到 QueueShutDown；不能把它当作成功处理。生产者和消费者都捕获该异常，清理资源后退出。旧 Python 用封口标志、哨兵或自定义队列，并明确版本差异。”
分步骤深入解答
第一步：定义队列不变量
有界队列用 maxsize 施加背压；每次成功 put 增加未完成计数，每次 worker 完成一个项目调用一次 task_done。join() 只表示计数归零，不表示 worker 已退出。
queue = asyncio.Queue(maxsize=100)
await queue.put(job)
job = await queue.get()
try:
await process(job)
finally:
queue.task_done()第二步：实现优雅封口
停机协调器先停止上游读取，再调用 shutdown(immediate=False)。后续 put（包括正在等待空间的生产者）会收到 QueueShutDown；队列中的项目仍可被取出，直到空队列的 get 也抛出该异常。
第三步：让 worker 正确退出
worker 循环捕获 QueueShutDown 作为正常退出信号；业务异常不能跳过 task_done。用 finally 释放连接、租约和临时文件，避免停机时留下半处理资源。
async def worker(queue):
while True:
try:
job = await queue.get()
except asyncio.QueueShutDown:
return
try:
await process(job)
finally:
queue.task_done()第四步：等待排空并停止 worker
协调器等待 queue.join()，确认所有已接受项目都调用过 task_done，再取消仍在等待新项目的 worker。取消 worker 不等于取消正在执行的下游 I/O，驱动仍需超时或取消支持。
第五步：理解立即关闭
shutdown(immediate=True) 会排空队列、唤醒阻塞的 get 和 put，并让 join 可能在项目尚未处理时解除阻塞。它适合进程即将崩溃或任务已转移到持久化补偿的场景，不适合正常发布。
第六步：处理生产者、消费者和调用方取消
QueueShutDown 表示队列生命周期结束；CancelledError 表示调用方取消。两者都要停止循环，但记录原因不同。不要捕获 BaseException 后吞掉取消，也不要在 task_done 之前返回。
第七步：版本兼容与跨边界
shutdown 与 QueueShutDown 自 Python 3.13 提供。多版本服务可在启动时检测能力，或使用带关闭状态的封装；跨线程应使用线程安全队列，asyncio.Queue 只适用于单一事件循环。
第八步：测试停机语义
测试队列为空、已满、生产者阻塞、消费者阻塞、优雅排空、立即关闭、重复关闭、worker 业务异常、调用方取消和进程超时。断言每个成功取出的项目恰好一次 task_done，并验证立即关闭时未处理项目有明确丢弃或补偿记录。
高质量示范回答
“我把停机分为封口排空和立即终止。优雅路径先停止上游，再调用默认 shutdown()；生产者收到 QueueShutDown，worker 继续处理已有项目并在 finally 调用 task_done，协调器等待 join 后取消空闲 worker。紧急路径用 immediate=True，明确接受队列中项目被丢弃，不能把提前解除的 join 当作成功。部署前检查 Python 版本，并为旧版本保留哨兵或封装回退。”
常见错误
- 只设置一个 stopped 布尔值 → 阻塞的
put/get永远不醒 → 使用 shutdown 或显式唤醒协议。 - 立即关闭后把 join 当成功 → 未处理项目被误报完成 → 记录丢弃并区分终止原因。
- 忘记 task_done → 优雅停机永久卡在 join → 用 finally 保证每个 get 配对一次。
- 先取消 worker 再封口 → 新生产者仍不断入队 → 先停止上游和队列生产。
- 吞掉 QueueShutDown 与 CancelledError → worker 无法可靠退出 → 分别记录并结束循环。
- 在跨线程共享 asyncio.Queue → 事件循环安全性失效 → 使用线程安全队列或消息系统。
追问及应对
追问一：优雅 shutdown 后还能 get 到项目吗？
可以，已有项目可继续被取出；当队列排空后，后续 get 抛出 QueueShutDown。
追问二：为什么 immediate=True 会破坏 join 不变量？
它直接清空队列并调整未完成计数，可能让 join 在工作尚未处理时解除，所以只能用于明确接受丢弃或已有补偿的紧急路径。
追问三：worker 正在处理的项目怎么办？
优雅路径等待它完成；紧急路径取消 worker，并要求下游操作支持超时、取消和幂等，失败项目写入持久化补偿。
追问四：如何兼容 Python 3.12？
封装队列关闭状态，用哨兵唤醒消费者、拒绝新生产，并自行追踪阻塞生产者；升级到 3.13 后再切换原生 API，保持相同的契约测试。
追问五：重复调用 shutdown 是否安全？
实现应把关闭作为幂等状态转换，重复调用不重新处理项目；仍需在目标 Python 版本上测试生产者和消费者的唤醒结果。