Queue dan Worker Patterns

Lock melindungi data, tetapi memikirkan kunci di setiap titik aksi rawan keliru. Module queue menawarkan jalan yang lebih disiplin: bertukar data antar thread lewat antrian yang sudah dilengkapi kunci internal, sehingga setiap put dan get dijamin aman tanpa satu baris penguncian pun. Pada Chapter ini kita akan mengenal ragam antrian bawaan Python, lalu menyusun pola producer-consumer lengkap dengan mekanisme shutdown yang bersih.

queue.Queue: Antrian Thread-Safe #

queue.Queue adalah antrian FIFO (first in, first out) yang thread-safe: put() menambah item, get() mengambil item paling depan. Python juga menyediakan dua saudaranya dengan disiplin pengambilan berbeda:

python
from queue import Queue, LifoQueue, PriorityQueue

fifo = Queue()
for pesanan in ["pesanan-1", "pesanan-2", "pesanan-3"]:
    fifo.put(pesanan)
print(f"FIFO : {fifo.get()}, {fifo.get()}, {fifo.get()}")

lifo = LifoQueue()
for pesanan in ["pesanan-1", "pesanan-2", "pesanan-3"]:
    lifo.put(pesanan)
print(f"LIFO : {lifo.get()}, {lifo.get()}, {lifo.get()}")

prioritas = PriorityQueue()
prioritas.put((2, "laporan mingguan"))
prioritas.put((1, "insiden produksi"))
prioritas.put((3, "arsip bulanan"))
print("Prioritas:")
while not prioritas.empty():
    level, nama_tugas = prioritas.get()
    print(f"  level {level}: {nama_tugas}")

Output yang Dihasilkan:

text
FIFO : pesanan-1, pesanan-2, pesanan-3
LIFO : pesanan-3, pesanan-2, pesanan-1
Prioritas:
  level 1: insiden produksi
  level 2: laporan mingguan
  level 3: arsip bulanan

Ketiganya menempati relung yang jelas:

Class	Perilaku	Kegunaan khas
`Queue`	FIFO — yang pertama masuk keluar pertama	penyebaran tugas ke worker, buffer aliran data
`LifoQueue`	LIFO — yang terakhir masuk keluar pertama	tumpukan pekerjaan undo/redo, backtracking
`PriorityQueue`	item prioritas terkecil keluar pertama	penjadwalan tugas (level 1 didahulukan)

PriorityQueue membandingkan item secara keseluruhan; itulah sebabnya kita memasang tuple (level, nama_tugas) — level dibandingkan lebih dulu, dan nama hanya menjadi pemecah seri. Pada antrian kosong, get() tidak error melainkan menunggu sampai item tersedia — perilaku blocking yang menjadi fondasi pola worker berikutnya.

Pola Producer-Consumer #

Arsitektur klasik ini memisahkan penghasil data (producer) dari pengolahnya (consumer). Producer cukup menjatuhkan pesan ke antrian; sekelompok consumer mengambil dan memproses. Berikut satu producer dan satu consumer:

python
from queue import Queue
import threading
import time

SENTINEL = None

def produsen(antrian, jumlah):
    for nomor in range(1, jumlah + 1):
        pesan = f"pesan-{nomor}"
        print(f"Produsen mengirim  : {pesan}")
        antrian.put(pesan)
        time.sleep(0.1)
    antrian.put(SENTINEL)

def konsumen(antrian):
    while True:
        pesan = antrian.get()
        if pesan is SENTINEL:
            antrian.task_done()
            break
        print(f"Konsumen memproses : {pesan}")
        antrian.task_done()

antrian_pesanan = Queue()
t_produsen = threading.Thread(target=produsen, args=(antrian_pesanan, 3))
t_konsumen = threading.Thread(target=konsumen, args=(antrian_pesanan,))

t_konsumen.start()
t_produsen.start()
t_produsen.join()
antrian_pesanan.join()
t_konsumen.join()
print("Semua pesan selesai diproses")

Output yang Dihasilkan:

text
Produsen mengirim  : pesan-1
Konsumen memproses : pesan-1
Produsen mengirim  : pesan-2
Konsumen memproses : pesan-2
Produsen mengirim  : pesan-3
Konsumen memproses : pesan-3
Semua pesan selesai diproses

Mari kita telaah bagian-bagian kuncinya:

Baris 7 (SENTINEL = None): nilai penanda khusus yang tidak mungkin berupa pesan sah. Ketika consumer melihat sentinel, consumer tahu tidak akan ada pesan lagi dan boleh berhenti — inilah pola shutdown paling umum untuk antrian.
Baris 9–15 (def produsen): producer mencetak, menjatuhkan pesan ke antrian, lalu beristirahat sejenak. Producer tidak perlu tahu siapa yang memproses pesannya.
Baris 17–25 (def konsumen): consumer berputar di antrian.get() yang blocking — kalau antrian kosong, thread consumer tertidur sampai item datang, tanpa memakan CPU.
Baris 24 (antrian.task_done()): setiap get() harus diimbangi satu task_done() agar pencacah internal antrian turun. Kombinasi put() dan task_done() inilah yang membuat method berikut bisa bekerja.
Baris 33 (antrian_pesanan.join()): join() milik Queue menunggu hingga semua item yang pernah di-put sudah menerima task_done() — termasuk sentinel yang juga kita akui dengan task_done() sebelum break.

Urutan baris “mengirim” dan “memproses” di atas stabil karena producer sengaja beristirahat 0,1 detik; tanpa jeda, urutan selang-selingnya dapat berbeda di setiap eksekusi, meskipun setiap pesan tetap utuh diproses tepat satu kali.

deque Bukan Pengganti Queue #

Class collections.deque sering diperkenalkan sebagai antrian cepat, dan memang operasi append() serta popleft() di deque bersifat atomik. Namun urutan operasi majemuk — misalnya memeriksa len(d) > 0 lalu popleft() di dua langkah terpisah — tidak dilindungi kunci apa pun, sehingga dua thread bisa sama-sama lolos pemeriksaan lalu berebut item yang tersisa.

Gunakan collections.deque untuk struktur data di dalam satu thread; gunakan queue.Queue ketika data melintasi batas thread. Bedanya bukan kecepatan, melainkan jaminan: queue.Queue menangani penguncian, penantian, dan penutupan secara benar sejak dirancang.