題干與適用場景
一個服務使用 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 版本上測試生產者和消費者的喚醒結果。