Belajar RabbitMQ - Flow Control & Backpressure

#Pendahuluan

Bayangkan sebuah toko dengan pintu masuk yang tidak pernah ditutup: jika pengunjung datang ribuan sekaligus, toko pasti jebol. Broker pesan menghadapi masalah yang sama — jika publisher membanjiri broker tanpa henti, memori dan disk akan habis dan seluruh node bisa crash.

RabbitMQ menyelesaikan ini dengan flow control: serangkaian mekanisme yang memperlambat publisher saat broker kewalahan. Alih-alih crash, broker memblokir connection atau menahan kredit antar proses internal. Consumer tidak perlu tahu apa-apa, tapi publisher wajib memahami mekanisme ini agar aplikasinya tidak macet.

Episode ini mengupas lapisan-lapisan flow control: dari TCP backpressure paling bawah, credit-based flow control di dalam node, sampai memory alarm yang memblokir seluruh connection. Lalu kita bangun publisher confirms — cara resmi publisher mengetahui bahwa pesannya benar-benar diterima broker — dan strategi praktis menangani backpressure.

#Flow Control Mechanisms

#TCP Backpressure dan Credit-Based Flow Control

Ketika consumer lambat, queue menumpuk, dan memori node mulai penuh. RabbitMQ berhenti membaca data dari TCP socket connection yang mem-publish, sehingga receive buffer client penuh dan TCP akan menahan pengiriman. Ini disebut TCP backpressure — mekanisme paling dasar.

Di dalam node, proses Erlang berkomunikasi lewat credit-based flow control: setiap proses harus memiliki kredit sebelum mengirim pesan ke proses lain. Jika proses penerima sibuk, kredit menipis dan pengirim otomatis melambat. Mekanisme ini membuat RabbitMQ tidak mudah banjir secara internal.

#Connection Blocked State dan Memory Alarm

Ketika kondisi kritis tercapai, RabbitMQ memblokir seluruh connection yang mem-publish dan mengirimkan notifikasi connection.blocked kepada client:

Connection.blocked reason=memory_limit
Connection.unblocked

Client SDK modern menerjemahkan ini ke callback. Di pika:

Callback blocked dan unblocked

def on_blocked(connection, reason):
    print(f"publisher diblokir: {reason}")
 
def on_unblocked(connection):
    print("publisher dibuka kembali")
 
connection.add_on_connection_blocked_callback(on_blocked)
connection.add_on_connection_unblocked_callback(on_unblocked)

Callback add_on_connection_blocked_callback memberitahu aplikasi saat broker memblokir publish karena memory alarm. Aplikasi yang baik menjeda publish dan menahan buffer secara internal.

#Publisher Confirms

#Mengaktifkan Confirm Mode

Publisher confirms adalah fitur yang memberitahu publisher bahwa pesan sudah diterima broker (dan, jika persistent, ditulis ke disk). Tanpa confirms, publisher tidak pernah yakin pesannya tidak hilang.

Aktifkan confirm mode

channel.confirm_delivery()

Setelah confirm_delivery() diaktifkan, setiap publish akan menerima konfirmasi atau nack.

#Confirms Sinkron dan Asinkron

Confirms bisa ditunggu sinkron per pesan:

Confirm sinkron dengan timeout

try:
    channel.basic_publish(exchange="", routing_key="q", body=b"data")
    channel.confirm_delivery()  # hanya contoh; aktivasi sekali
except pika.exceptions.UnroutableError:
    print("pesan gagal di-routing")

Lebih efisien: batch confirms — publish banyak pesan, lalu tunggu satu konfirmasi di akhir batch. Untuk throughput tertinggi, gunakan asynchronous confirms dengan callback yang melacak setiap delivery tag.

#Menangani Nack

Jika broker mengirim nack untuk sebuah pesan, berarti pesan ditolak (misalnya queue penuh dengan reject-publish). Publisher harus menyimpan pesan tersebut dan mengirim ulang atau masuk antrean internal. Jangan pernah diam saja — pesan yang di-nack dianggap hilang dari sisi publisher.

#Handling Backpressure

#Strategi Publisher Throttling

Saat connection.blocked terjadi, jangan terus memaksa publish. Strategi yang benar:

• Jeda publish dan tahan pesan di buffer internal sampai unblocked.
• Terapkan batas ukuran buffer agar memori aplikasi tidak ikut meledak.
• Kurangi batch size dan turunkan intensitas publish.

#Batas Queue dan Pemantauan Disk

Backpressure di broker biasanya dipicu queue yang membesar tak terkendali. Pasang x-max-length atau x-overflow untuk membatasi, dan pantau metrik disk serta memori secara proaktif:

rabbitmqctl status | grep -A 2 -i memory
rabbitmqctl status | grep -A 2 -i disk

Perintah rabbitmqctl status menampilkan penggunaan memori dan disk terkini — keduanya adalah alarm utama yang memicu blocked state.

#Penutup

Di episode 14 ini kalian sudah memahami seluruh lapisan flow control RabbitMQ: TCP backpressure, credit-based flow control, memory alarm, dan connection blocked state; mengaktifkan publisher confirms sinkron maupun asinkron; serta menyusun strategi throttling publisher dan pemantauan disk.

Inti yang harus dibawa pulang:

• TCP backpressure dan credit control mencegah banjir internal.
• Memory alarm memblokir connection dan mengirim connection.blocked.
• Aplikasi wajib menjeda publish saat connection diblokir.
• Publisher confirms memberi kepastian pesan diterima broker.
• Batch confirms meningkatkan throughput; async confirms paling tinggi.
• Nack berarti pesan ditolak — tangani, jangan abaikan.
• Pasang queue limit dan pantau memori serta disk secara proaktif.

Di episode 15 selanjutnya kita akan mengelola user management dan authentication — membuat user lewat Management UI dan rabbitmqctl, menetapkan user tags, menerapkan kebijakan password, serta membandingkan backend authentication: internal, LDAP, OAuth 2.0, sertifikat x509, dan HTTP backend. Keamanan RabbitMQ dimulai dari siapa yang boleh login!