Wawancara coding: Bagaimana seharusnya Python 3.13 asyncio.Queue dimatikan?

Perintah dan konteks

Sebuah asyncio.Queue mendistribusikan pekerjaan ke beberapa worker. Selama deployment, produsen harus berhenti menerima pekerjaan baru, item yang ada harus diselesaikan, dan worker kemudian harus keluar. Selama kegagalan fatal, produsen dan konsumen yang terblokir harus segera dibangunkan. Rancang kedua jalur tersebut dengan Python 3.13 Queue.shutdown().

Dokumentasi Python menyatakan bahwa shutdown(immediate=False) default menutup antrean untuk put baru sambil membiarkan konsumen menguras (drain) item yang ada. immediate=True menguras antrean dan dapat melanggar invarian join() yang biasa. QueueShutDown adalah sinyal siklus hidup untuk kedua belah pihak.

Apa yang sedang diuji oleh pewawancara

Kandidat yang kuat memisahkan penghentian produksi dari pembatalan konsumsi, memasangkan setiap get() yang berhasil dengan tepat satu task_done(), dan menjelaskan mengapa shutdown langsung tidak dapat diartikan sebagai pemrosesan yang berhasil. Mereka juga mencakup fallback untuk Python 3.12 dan pembatalan I/O eksternal.

Pertanyaan klarifikasi untuk diajukan terlebih dahulu

Apakah graceful shutdown harus menyelesaikan setiap item yang diterima?
Apakah emergency shutdown boleh membuang pekerjaan dalam antrean, atau apakah kompensasi tahan lama (durable compensation) diperlukan?
Apakah produsen berada dalam satu event loop, atau lintas thread/proses?
Apakah worker melakukan I/O eksternal, percobaan ulang (retry), atau operasi idempoten?
Berapa versi Python minimum di lingkungan produksi?

Kerangka jawaban 30 detik

"Graceful shutdown pertama-tama menghentikan asupan upstream, kemudian memanggil queue.shutdown() dengan mode default. Panggilan put baru menerima QueueShutDown; worker menguras item yang ada dan memanggil task_done dalam finally; koordinator menunggu queue.join() lalu membatalkan worker yang menganggur. Emergency shutdown menggunakan immediate=True, menerima bahwa item dalam antrean dibuang, dan tidak pernah memperlakukan bangunnya join lebih awal sebagai keberhasilan. Produsen dan konsumen sama-sama menangani QueueShutDown. Versi Python yang lebih lama memerlukan sentinel atau wrapper penutup."

Pembahasan mendalam langkah demi langkah

Langkah 1: Tentukan invarian antrean

Antrean terbatas (bounded queue) menerapkan backpressure dengan maxsize. Setiap put yang berhasil meningkatkan hitungan yang belum selesai, dan setiap item yang selesai membutuhkan satu task_done. join() berarti hitungan telah mencapai nol; ini tidak berarti worker telah keluar.

python

queue = asyncio.Queue(maxsize=100)
await queue.put(job)
job = await queue.get()
try:
    await process(job)
finally:
    queue.task_done()

Langkah 2: Tutup produksi secara graceful

Koordinator menghentikan pembacaan upstream, kemudian memanggil shutdown(immediate=False). Operasi put di masa mendatang, termasuk produsen yang terblokir karena kapasitas penuh, menerima QueueShutDown. Item yang ada tetap tersedia sampai antrean kosong, setelah itu get juga memunculkan pengecualian tersebut.

Langkah 3: Buat worker keluar dengan benar

Worker memperlakukan QueueShutDown sebagai jalan keluar siklus hidup normal. Kegagalan bisnis tidak boleh melewati task_done. Gunakan finally untuk melepaskan koneksi, lease, dan file sementara.

python

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

Langkah 4: Kuras dan hentikan worker

Tunggu queue.join() sehingga setiap item yang diterima telah menyelesaikan akuntansinya, lalu batalkan worker yang menganggur pada get. Membatalkan Task tidak menjamin bahwa operasi database atau HTTP berhenti; driver masih memerlukan batas waktu (deadline) atau mekanisme pembatalan.

Langkah 5: Pahami shutdown langsung (immediate shutdown)

shutdown(immediate=True) menguras antrean, membangunkan pemanggil get dan put yang terblokir, dan dapat melepaskan join sebelum pekerjaan berjalan. Gunakan hanya jika membuang item dalam antrean dapat diterima atau kompensasi tahan lama sudah ada, bukan untuk deployment normal.

Langkah 6: Pisahkan shutdown antrean dari pembatalan pemanggil

QueueShutDown berarti siklus hidup antrean telah berakhir; CancelledError berarti pemanggil membatalkan pekerjaan. Keduanya menghentikan loop, tetapi membutuhkan alasan yang berbeda dalam log dan metrik. Jangan menangkap BaseException lalu menelan pembatalan, dan jangan pernah return sebelum task_done untuk item yang berhasil diambil.

Langkah 7: Tangani versi dan batasan

shutdown dan QueueShutDown ditambahkan pada Python 3.13. Layanan multi-versi dapat mendeteksi dukungan saat startup atau menggunakan wrapper penutup. asyncio.Queue ditujukan untuk satu event loop; pekerjaan lintas thread memerlukan antrean yang thread-safe atau sistem pengiriman pesan.

Langkah 8: Uji semantik shutdown

Uji antrean kosong dan penuh, produsen dan konsumen yang terblokir, pengurasan graceful, pengurasan langsung, shutdown berulang, error pada worker, pembatalan pemanggil, dan deadline proses. Lakukan penegasan (assert) tepat satu task_done per get yang berhasil, dan catat hasil pembuangan atau kompensasi yang jelas untuk item yang hilang akibat shutdown langsung.

Model jawaban berkualitas tinggi

"Saya memisahkan pengurasan graceful dari terminasi darurat. Jalur graceful menghentikan asupan upstream, memanggil shutdown default, membiarkan worker menyelesaikan item yang ada dengan task_done di dalam finally, menunggu join, lalu membatalkan worker yang menganggur. Jalur darurat menggunakan immediate=True, secara eksplisit menerima kehilangan item dalam antrean, dan tidak menyebut bangunnya join lebih awal sebagai keberhasilan. Saya memeriksa versi Python dan mempertahankan fallback sentinel atau wrapper untuk runtime yang lebih lama."

Kesalahan umum

Hanya menyetel boolean berhenti → panggilan put/get yang terblokir tidak pernah bangun → gunakan shutdown atau protokol bangun yang eksplisit.
Memperlakukan shutdown langsung plus join sebagai keberhasilan → pekerjaan yang dibuang dilaporkan selesai → catat pembuangan dan alasannya secara terpisah.
Melupakan task_done → join yang graceful menggantung selamanya → pasangkan setiap get dalam blok finally.
Membatalkan worker sebelum menutup produksi → pekerjaan baru terus berdatangan → hentikan upstream terlebih dahulu.
Menelan QueueShutDown dan CancelledError → worker tidak dapat keluar dengan andal → catat setiap alasan siklus hidup dan lakukan return.
Berbagi asyncio.Queue lintas thread → keamanan event-loop hilang → gunakan antrean yang thread-safe atau sistem perpesanan.

Pertanyaan lanjutan dan respons yang kuat

Lanjutan 1: Bisakah Anda tetap mengambil item setelah graceful shutdown?

Ya. Item yang ada dapat diambil; setelah antrean kosong, panggilan get berikutnya akan memunculkan QueueShutDown.

Lanjutan 2: Mengapa mode langsung melanggar invarian join?

Mode ini menguras antrean dan menyesuaikan akuntansi item yang belum selesai, sehingga join dapat bangun sebelum pekerjaan diproses. Ini hanya cocok untuk jalur yang secara eksplisit menerima kehilangan data atau memiliki mekanisme kompensasi.

Lanjutan 3: Apa yang terjadi pada item yang sedang diproses?

Graceful shutdown menunggunya selesai. Emergency shutdown membatalkan worker; operasi downstream harus mendukung batas waktu (deadline), pembatalan, dan kompensasi yang idempoten.

Lanjutan 4: Bagaimana Anda mendukung Python 3.12?

Bungkus antrean dengan status tertutup, tolak put baru, bangunkan konsumen dengan sentinel, dan lacak produsen yang terblokir. Beralih ke API bawaan setelah peningkatan versi sambil mempertahankan pengujian kontrak yang sama.

Lanjutan 5: Apakah shutdown berulang kali aman dilakukan?

Wrapper harus membuat penutupan bersifat idempoten dan menghindari pemrosesan ulang item. Tetap uji perilaku bangunnya produsen dan konsumen pada runtime Python yang tepat digunakan.