asyncio.Queue.shutdown: ปิดสายพานงานโดยไม่ส่ง Sentinel

producer-consumer แบบ async มักปิดระบบด้วย sentinel เช่น None ครับ แต่ต้องส่งกี่ตัวเมื่อมี worker หลายคน ถ้า None เป็นข้อมูลที่ถูกต้องจะทำอย่างไร และ producer ที่ยังเขียนอยู่รู้ได้อย่างไรว่าสายพานปิดแล้ว ปัญหาเล็กนี้มักกลายเป็น task ค้างตอน deploy

asyncio.Queue.shutdown ที่เพิ่มตั้งแต่ Python 3.13 ทำสถานะปิดเป็นส่วนหนึ่งของ queue โดยตรง หลัง shutdown จะรับ put ใหม่ไม่ได้ ส่วน graceful mode ยังให้ consumer ดึงงานเดิมจนหมด ก่อน get ครั้งถัดไปยก QueueShutDown

WHY: แยก “หยุดรับ” ออกจาก “ทิ้งงาน”

การหยุดรับงานใหม่ไม่ควรแปลว่าทิ้งงานในคิว เหมือนร้านปิดประตูเวลาเลิกงาน แต่ยังทำออเดอร์ที่รับเงินแล้วให้เสร็จ shutdown() ค่าเริ่มต้นรักษา invariant ของ join: ทุก item ต้องมี task_done ก่อน join จบ

shutdown(immediate=True) ต่างออกไป มันระบายคิวและอาจปลด join แม้งานยังไม่ถูกทำ เอกสารจึงเตือนชัดว่า invariant ปกติถูกละเมิด ใช้เฉพาะ incident ที่ยอมเสียงานได้และมีระบบ replay หรือ reconciliation รองรับ

import asyncio
import unittest


async def consume(queue, seen):
    while True:
        try:
            item = await queue.get()
        except asyncio.QueueShutDown:
            return
        seen.append(item * 2)
        queue.task_done()


class QueueShutdownTests(unittest.IsolatedAsyncioTestCase):
    async def test_graceful_shutdown_drains_existing_work(self):
        queue = asyncio.Queue()
        seen = []
        worker = asyncio.create_task(consume(queue, seen))

        await queue.put(2)
        await queue.put(3)
        queue.shutdown()
        await queue.join()
        await worker

        self.assertEqual(seen, [4, 6])
        with self.assertRaises(asyncio.QueueShutDown):
            await queue.put(4)


if __name__ == "__main__":
    unittest.main()

HOW: วาง Shutdown Sequence

หยุด source ก่อน เช่นยกเลิก HTTP intake หรือ unsubscribe message broker แล้วเรียก queue.shutdown จากนั้น await queue.join และรอ worker ทุกตัวจบ หากกลับลำดับ producer อาจแข่งกับ shutdown แล้วได้ QueueShutDown ต้องกำหนดว่าจะ retry upstream หรือบันทึกเป็น rejected work

consumer ควรจับ QueueShutDown รอบ get ไม่จับ Exception กว้าง เพราะ error จาก process(item) ต้องถูกสังเกตและจัดการต่างหาก ใช้ try/finally เรียก task_done เฉพาะ item ที่ get สำเร็จแล้ว มิฉะนั้น counter ของ join จะไม่สมดุล

Queue ของ asyncio ไม่ thread-safe ถ้า producer มาจาก thread อื่น ให้ส่งผ่าน event-loop API ที่เหมาะสม อย่านำ queue เดียวข้าม thread โดยตรง และถ้าต้องมี timeout ให้ครอบ operation ด้วย asyncio.wait_for เพราะ queue method ไม่มี timeout parameter

Recap

• shutdown ปิดการรับ item ใหม่อย่างชัดเจน
• graceful mode ระบายงานเก่าก่อนยก QueueShutDown
• immediate mode อาจทำให้ join จบทั้งที่งานไม่ครบ
• หยุด producer ก่อนปิด queue
• แยก error ของ lifecycle ออกจาก error ของ business logic

ลองแทน sentinel ใน worker pool หนึ่งตัว แล้วเพิ่ม test ว่า deploy drain งานเดิมครบและปฏิเสธงานใหม่ครับ