Belajar RabbitMQ - Work Queues - Task Distribution

#Pendahuluan

Pola Hello World di episode 4 hanya memakai satu consumer. Di dunia nyata, satu consumer saja tidak cukup untuk task yang berat seperti merender video atau mengirim ribuan email. Di sinilah work queue berperan: satu queue, banyak worker, dan pesan didistribusikan secara otomatis.

Namun membangun work queue yang benar bukan sekadar menjalankan beberapa consumer sekaligus. Ada tiga mekanisme yang harus dikuasai: manual acknowledgment agar pesan tidak hilang saat worker crash, durability agar pesan selamat dari restart broker, dan prefetch count agar pesan tidak ditimbun di satu worker yang lambat.

Episode ini adalah titik di mana RabbitMQ menunjukkan keunggulannya dibandingkan komunikasi langsung: kombinasi ack, durability, dan QoS menghasilkan sistem yang menjamin setiap task terproses, bahkan ketika worker dan broker bermasalah.

#Work Queue Pattern

#Round-Robin Dispatching

Secara default, RabbitMQ mengirim pesan ke consumer secara round-robin: pesan pertama ke worker A, kedua ke worker B, ketiga ke A lagi, dan seterusnya. Distribusi ini rata dan sederhana — setiap worker menerima jumlah pesan yang hampir sama.

Masalahnya, round-robin mengabaikan seberapa sibuk masing-masing worker. Jika worker A sangat lambat, pesan-pesan tetap mengalir ke arahnya dan menumpuk. Inilah alasan kita butuh fair dispatch yang dibahas di akhir episode.

queue ──► worker A (pesan 1, 3, 5)
     └──► worker B (pesan 2, 4, 6)

#Menjalankan Beberapa Worker

Untuk menjalankan beberapa worker sekaligus, cukup jalankan consumer yang sama di beberapa proses atau thread:

python worker.py &
python worker.py &

Deklarasikan queue yang sama di semua worker. Satu pesan hanya akan sampai ke satu worker — bukan diduplikasi ke semuanya.

#Message Acknowledgments

#Manual vs Automatic Acknowledgment

auto_ack=True yang kita pakai di episode 4 berbahaya: begitu pesan dikirim ke worker, broker langsung menghapusnya dari queue, tanpa tahu apakah worker berhasil memproses. Jika worker crash di tengah proses, pesan hilang selamanya.

Solusinya adalah manual acknowledgment:

Consumer dengan ack manual

import pika, time
 
connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = connection.channel()
channel.queue_declare(queue="task_queue", durable=True)
 
def callback(ch, method, properties, body):
    print(f"memproses: {body}")
    time.sleep(2)
    ch.basic_ack(delivery_tag=method.delivery_tag)
 
channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue="task_queue", on_message_callback=callback)
print("menunggu task...")
channel.start_consuming()

Pemanggilan ch.basic_ack mengirim Basic.Ack ke broker. Sampai ack diterima, pesan dianggap "belum selesai" dan tidak dihapus.

#Basic.Nack dan Basic.Reject

Selain ack, AMQP menyediakan Basic.Reject untuk menolak satu pesan, dan Basic.Nack untuk menolak satu atau banyak pesan sekaligus. Keduanya bisa menandai apakah pesan harus di-requeue atau dibuang ke dead letter exchange:

Menolak pesan dengan requeue

ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

Perintah ch.basic_nack dengan requeue=False mengirim pesan ke dead letter queue jika DLX dikonfigurasi — topik lengkapnya di episode 11.

#Delivery Timeout

RabbitMQ 3.12 memperkenalkan delivery timeout (delivery_timeout): jika consumer tidak mengirim ack dalam batas waktu tertentu untuk sebuah pesan, broker akan memutus connection consumer. Nilai defaultnya 30 menit, dan bisa diatur di rabbitmq.conf:

consumer_timeout = 600000

Nilai consumer_timeout di atas dalam milidetik, dan kalian bisa menonaktifkannya dengan consumer_timeout = 0 (tidak disarankan).

#Message Durability

#Queue Durable dan Pesan Persistent

Deklarasi queue saja tidak cukup menjamin pesan selamat dari restart broker. Ada dua lapis: queue harus durable, dan pesan harus persistent (delivery_mode=2). Keduanya wajib ada — satu saja tidak cukup.

Publish pesan persistent

channel.queue_declare(queue="task_queue", durable=True)
channel.basic_publish(
    exchange="",
    routing_key="task_queue",
    body=b"task berat",
    properties=pika.BasicProperties(delivery_mode=2),
)

Jika queue dideklarasikan ulang tanpa durable=True di environment yang sama, RabbitMQ akan menolak karena deklarasi tidak kompatibel — pesan error akan muncul di log.

#Trade-Off Performa

Persistence butuh biaya: setiap pesan persistent ditulis ke disk sebelum broker mengirim ack publisher. Ini meningkatkan latency publish. Trade-off-nya: durability untuk data penting, transient untuk data yang boleh hilang seperti log atau notification ephemeral.

#Fair Dispatch dengan Prefetch

#Konsep Prefetch Count

Round-robin tidak peduli kecepatan worker. Prefetch count mengubah perilaku ini: ia membatasi jumlah pesan yang boleh dikirim ke satu consumer sebelum consumer mengirim ack. Dengan prefetch_count=1, worker hanya menerima satu pesan pada satu waktu, dan pesan berikutnya baru dikirim setelah yang sekarang di-ack.

rabbitmqctl eval 'application:get_env(rabbit, channel_prefetch_count).'

Atau di client SDK, gunakan channel.basic_qos(prefetch_count=1) seperti contoh di atas. Nilai 1 paling adil tapi menambah overhead round-trip; nilai 10-100 biasanya optimal untuk throughput.

#Nilai Prefetch yang Optimal

Tidak ada angka universal. Mulai dari prefetch_count=1 untuk task berat yang lama, naikkan jika pesan kecil dan cepat diproses. Pantau consumer utilization di Management UI; jika konsisten di 90 persen ke atas, prefetch terlalu tinggi.

#Penutup

Di episode 5 ini kalian sudah membangun work queue yang benar: round-robin dispatching, manual acknowledgment dengan Basic.Ack/Nack/Reject, durability dua lapis untuk queue dan pesan, serta fair dispatch dengan prefetch count agar setiap worker bekerja sesuai kapasitasnya.

Inti yang harus dibawa pulang:

• Round-robin mendistribusikan pesan rata, tapi mengabaikan kesibukan worker.
• Manual ack mencegah pesan hilang saat worker crash di tengah proses.
• Basic.Nack dengan requeue=False membuang pesan ke DLX.
• Queue harus durable dan pesan persistent agar selamat dari restart.
• Persistence menambah latency — pilih sesuai kebutuhan data.
• prefetch_count=1 menjamin keadilan; nilai optimal tergantung beban kerja.
• Delivery timeout memutus consumer yang tidak kunjung ack.

Di episode 6 selanjutnya kita akan melepas default exchange dan mempelajari exchange types: direct, fanout, topic, headers, dan default exchange, lengkap dengan routing key, binding key, serta implementasi pola publish/subscribe. Konsep ini mengubah RabbitMQ dari sekadar queue menjadi router pesan yang fleksibel!