Delivery #

Dalam perjalanan sebuah pesan di broker RabbitMQ (message lifecycle), setelah pesan berhasil melewati tahap penerbitan, perutean, dan pengantrean (Queueing), sampailah ia pada gerbang akhir yang menghubungkan broker dengan logika bisnis aplikasi kita: fase Delivery (Pengiriman).

Fase Pengiriman adalah tahap ketika pesan yang berstatus siap dikonsumsi (Ready) diambil dari antrean penyimpanan dan disalurkan melalui saluran TCP/AMQP ke aplikasi konsumen (consumer). Meskipun terdengar sederhana, proses pengiriman ini melibatkan mekanisme negosiasi yang rumit antara broker dan konsumen. Agar sistem berjalan dengan efisien dan andal, RabbitMQ mengandalkan model pengiriman berbasis dorongan (push-based), kontrol aliran data menggunakan batas pra-ambil (prefetch QoS), pelacakan status konfirmasi (acknowledgement), serta penanganan bendera pengiriman ulang (redelivered flags). Artikel ini akan membahas tuntas mekanisme internal pengiriman pesan di RabbitMQ, perbandingan performa antara model push dan pull, kalkulasi matematika untuk menentukan setelan prefetch yang optimal, serta implementasi kode Go untuk menangani konsumsi pesan secara tangguh.

Posisi Pengiriman dalam Siklus Hidup Pesan #

Fase Pengiriman bertindak sebagai jembatan aktif antara fase antrean penyimpanan (Queueing) dan fase penyelesaian konfirmasi (Acknowledgement).

flowchart TD
    Queue["Antrean (Pesan Status: Ready)"] --> Select{"Evaluasi Konsumen & Kuota"}
    Select -->|Memenuhi Syarat| Push["Push Pesan (basic.deliver)"]
    Select -->|Kuota Penuh / Tidak Ada Consumer| Queue
    
    Push --> Track["Catat ke Daftar Unacked"]
    Track --> Client["Consumer Memproses Pesan"]
    
    Client --> Success{"Sukses Pemrosesan?"}
    Success -->|Ya| Ack["Kirim ACK (Pesan Dihapus)"]
    Success -->|Tidak("Crash / NACK")| Requeue["Requeue (Kembali ke Antrean dengan Redelivered=true)"]

Ketika pesan berada dalam fase pengiriman, statusnya bertransisi dari Ready menjadi Unacknowledged. Selama status ini aktif, broker tidak akan pernah menghapus payload pesan dari media penyimpanannya sampai aplikasi konsumen memberikan respon konfirmasi yang jelas.

Mekanisme Push-Model vs Pull-Model: basic.consume vs basic.get #

RabbitMQ menyediakan dua metode protokol AMQP 0-9-1 bagi aplikasi konsumen untuk memperoleh pesan dari antrean: model dorong (push) dan model tarik (pull). Memahami perbedaan fundamental kedua model ini sangat menentukan performa pemrosesan data kita.

1. Model Push (basic.consume) — Direkomendasikan #

Model dorong adalah metode default dan sangat direkomendasikan di RabbitMQ. Pada model ini, aplikasi konsumen mendaftarkan diri (subscribe) ke antrean tertentu melalui perintah basic.consume.

  • Cara Kerja: Sekali konsumen terdaftar, broker RabbitMQ akan bertindak secara aktif membanjiri (push) pesan ke soket jaringan konsumen segera setelah pesan tersedia di antrean, tanpa menunggu konsumen meminta data.
  • Kelebihan: Latensi pengiriman sangat rendah (mendekati nol milidetik setelah pesan diantrekan) dan penggunaan resource jaringan sangat efisien karena tidak ada pemanggilan overhead round-trip yang tidak perlu.
  • Kontrol: Kecepatan dorongan ini dikontrol secara ketat menggunakan setelan Prefetch Limit (QoS) agar konsumen tidak kewalahan menerima data.

2. Model Tarik (basic.get) — Anti-Pattern untuk Produksi #

Model tarik adalah metode polling manual di mana konsumen secara aktif meminta satu pesan tunggal ke broker menggunakan perintah basic.get.

  • Cara Kerja: Konsumen mengirimkan request basic.get ke broker. Jika ada pesan di antrean, broker mengirimkannya; jika tidak, broker mengirimkan respon kosong (no-message). Konsumen harus terus-menerus mengirim request ini di dalam loop tak terbatas.
  • Kelemahan: Setiap panggilan basic.get memicu biaya round-trip jaringan penuh (kirim request, buka transaksi, cari data di antrean, kirim respon, tutup transaksi). Hal ini menyebabkan penurunan throughput hingga 100 kali lipat lebih lambat dibandingkan basic.consume dan membebani utilisasi CPU broker secara drastis.
  • Kapan Digunakan: Hanya boleh digunakan untuk debugging cepat, pengujian manual, atau pemrosesan batch periodik di mana kita hanya ingin memeriksa satu pesan tanpa memelihara koneksi konsumen aktif.

Batas Pra-Ambil (Prefetch QoS) dan Rumus Perhitungannya #

Karena RabbitMQ secara default menggunakan model dorong (basic.consume), jika ada 100.000 pesan di antrean dan kita menyalakan satu konsumen baru, broker akan mencoba mengirimkan seluruh 100.000 pesan tersebut ke memori aplikasi konsumen secepat mungkin melalui soket TCP.

Hal ini dapat menyebabkan aplikasi konsumen mengalami kehabisan memori RAM (OutOfMemory Crash) atau membuat konsumen lain tidak kebagian pesan (consumer starvation). Untuk mencegah hal ini, kita wajib menerapkan Prefetch Limit (Quality of Service - QoS).

Bagaimana Prefetch Bekerja #

Prefetch adalah batas kuota pesan yang belum dikonfirmasi (unacknowledged) yang diizinkan untuk dikirimkan ke satu channel konsumen. Jika kita menyetel prefetch = 100, broker hanya akan mengirimkan maksimal 100 pesan. Broker akan menangguhkan pengiriman pesan ke-101 sampai konsumen mengirimkan setidaknya satu sinyal basic.ack untuk membebaskan kuota.

Rumus Menghitung Prefetch Optimal #

Menyetel prefetch terlalu rendah (misalnya 1) akan membatasi throughput karena konsumen harus menunggu ACK selesai dikirim ke broker sebelum menerima pesan berikutnya. Sebaliknya, menyetel prefetch terlalu tinggi (misalnya 10.000) dapat menyebabkan pemborosan RAM dan distribusi beban tidak merata.

Kita dapat menghitung nilai prefetch optimal menggunakan rumus matematika sederhana berikut:

$$\text{Prefetch} = \frac{\text{Consumer Round-Trip Time (RTT)} + \text{Message Processing Time}}{\text{Message Processing Time}}$$

Atau yang lebih praktis untuk aplikasi paralel:

$$\text{Prefetch} \approx \text{Throughput Target (pesan/detik)} \times \text{RTT Jaringan (detik)}$$

Sebagai contoh:

  • Jika rata-rata waktu pemrosesan pesan oleh kode aplikasi kita adalah 50 milidetik (0.05 detik).
  • Latensi jaringan (RTT) antara server aplikasi dan broker RabbitMQ adalah 10 milidetik (0.01 detik).
  • Maka, Prefetch optimal per konsumen thread adalah:

$$\text{Prefetch} = \frac{10\text{ms} + 50\text{ms}}{50\text{ms}} = 1.2 \approx 2$$

Jika konsumen kita menggunakan sistem multi-threading (goroutine) yang dapat memproses 50 pesan secara paralel:

$$\text{Prefetch} = 50 \times 1.2 = 60$$

Dengan setelan prefetch sebesar 60, channel kita akan selalu memiliki cadangan pesan di memori lokal soket untuk menjaga utilitas CPU 100% tanpa jeda tunggu jaringan, namun tidak berlebihan sehingga membebani RAM.

Distribusi Beban Kerja: Round-Robin vs Prefetch Scheduling #

Ketika sebuah antrean memiliki beberapa konsumen terhubung secara bersamaan, RabbitMQ akan membagikan pesan di antara mereka. Mekanisme pembagian ini dikelola melalui dua strategi:

1. Round-Robin Murni (Tanpa Prefetch QoS) #

Jika kita mendaftarkan beberapa konsumen tanpa menentukan batas prefetch (atau prefetch = 0 yang berarti tak terbatas):

  • Logika: RabbitMQ akan membagi pesan secara merata bergiliran (Round-Robin). Konsumen A mendapat pesan 1, konsumen B mendapat pesan 2, konsumen A mendapat pesan 3, dan seterusnya.
  • Masalah: Jika pesan 1 membutuhkan waktu proses 1 jam (misalnya memproses video berat) sedangkan pesan 2 selesai dalam 1 detik, konsumen A akan memikul backlog besar di memori lokalnya sementara konsumen B menganggur setelah 1 detik. Ini disebut fenomena starvation dan distribusi beban yang tidak seimbang.

2. Fair Dispatch (Prefetch Scheduling) #

Ketika kita menetapkan batas prefetch (misalnya prefetch = 1 atau prefetch = 10):

  • Logika: RabbitMQ tidak lagi mengirim pesan secara membabi buta. Broker akan terus memantau jumlah pesan Unacknowledged di setiap konsumen.
  • Dampak: Jika konsumen A sedang sibuk memproses pesan berat dan batas prefetch-nya telah tercapai, broker akan melompati konsumen A dan langsung mengarahkan pesan-pesan berikutnya ke konsumen B yang masih memiliki kuota prefetch kosong. Ini menghasilkan pembagian beban kerja yang dinamis dan adil (fair dispatch).

Bendera Redelivered dan Garansi Pengiriman At-Least-Once #

RabbitMQ menganut prinsip At-Least-Once Delivery (Pesan dijamin terkirim minimal satu kali). Untuk mewujudkan garansi ini, broker harus siap mengirim ulang pesan yang gagal dikonfirmasi.

Bagaimana Redelivery Terjadi? #

Ketika pesan dikirim ke konsumen, broker melacak masa aktif koneksi TCP konsumen tersebut. Jika:

  1. Koneksi TCP konsumen terputus secara tiba-tiba (misalnya aplikasi crash, server mati, atau jaringan putus).
  2. Konsumen secara eksplisit menolak pesan dengan instruksi requeue (basic.nack atau basic.reject dengan requeue = true).

Broker akan secara otomatis mengembalikan pesan tersebut ke status Ready di dalam antrean tujuan.

Properti Redelivered = true #

Saat broker mengirimkan kembali pesan yang di-requeue tersebut ke konsumen baru (atau konsumen yang sama setelah terhubung kembali), broker akan menyematkan flag khusus di header AMQP: redelivered: true.

flowchart TD
    A["Pesan Pertama Kali"] -->|"basic.deliver"| B["redelivered: false"]
    B -->|"Koneksi Putus"| C["Pesan Dikirim Ulang"]
    C -->|"basic.deliver"| D["redelivered: true"]

Pentingnya Idempotensi di Tingkat Aplikasi #

Keberadaan properti redelivered: true adalah sinyal peringatan dini bagi aplikasi konsumen kita. Ketika aplikasi kita menerima pesan dengan flag ini aktif:

  • Aplikasi harus berasumsi bahwa pesan ini mungkin sudah pernah diproses sebagian atau sepenuhnya oleh konsumen sebelumnya sebelum crash terjadi.
  • Untuk mencegah eksekusi ganda yang merusak data (misalnya memotong saldo pengguna dua kali), aplikasi kita wajib menerapkan prinsip Idempotensi.
  • Solusi Idempotensi: Gunakan basis data kunci unik (seperti Redis atau indeks database relasional) untuk mencatat UUID unik pesan yang sukses diproses. Sebelum mengeksekusi logika bisnis, periksa apakah UUID pesan tersebut sudah ada di daftar riwayat sukses.

Implementasi Kode Go: Membuat Konsumen dengan QoS Prefetch yang Graceful #

Berikut adalah contoh implementasi lengkap dalam bahasa Go. Kode ini mendeklarasikan konsumen yang tangguh dengan menetapkan batas prefetch QoS, menggunakan manual ACK, memeriksa flag Redelivered, serta menerapkan penanganan sinyal sistem untuk graceful shutdown agar pesan yang sedang diproses tidak terputus di tengah jalan.

package main

import (
	"context"
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

func main() {
	// 1. Membuka Koneksi TCP ke RabbitMQ Broker
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	if err != nil {
		log.Fatalf("Gagal terhubung ke RabbitMQ: %s", err)
	}
	defer conn.Close()

	ch, err := conn.Channel()
	if err != nil {
		log.Fatalf("Gagal membuka channel: %s", err)
	}
	defer ch.Close()

	// 2. Mengonfigurasi Prefetch QoS (Batas Pra-ambil)
	// Kita membatasi maksimal 20 pesan unacknowledged pada channel ini
	err = ch.Qos(
		20,    // prefetch count: batas kuota pesan
		0,     // prefetch size: ukuran dalam byte (0 berarti tidak dibatasi)
		false, // global: false berarti batas berlaku per konsumen aktif di channel ini
	)
	if err != nil {
		log.Fatalf("Gagal menyetel QoS Prefetch: %s", err)
	}

	queueName := "transaction-logs-queue"

	// 3. Mendaftarkan Konsumen (Push Model - basic.consume)
	msgs, err := ch.Consume(
		queueName,
		"",    // consumer tag (dibiarkan kosong agar dibuat otomatis oleh broker)
		false, // autoAck: Wajib false untuk menjamin manual ACK demi keamanan data!
		false, // exclusive
		false, // noLocal
		false, // noWait
		nil,   // arguments
	)
	if err != nil {
		log.Fatalf("Gagal mendaftarkan konsumen: %s", err)
	}

	// Channel untuk menangkap sinyal terminasi OS (Ctrl+C, kill)
	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

	// Context untuk mengontrol siklus hidup goroutine pemroses
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	log.Println("[*] Menunggu pesan. Untuk keluar, tekan Ctrl+C")

	go func() {
		for {
			select {
			case <-ctx.Done():
				return
			case d, ok := <-msgs:
				if !ok {
					log.Println("[!] Channel pesan ditutup.")
					return
				}

				// 4. Memeriksa Bendera Redelivered untuk Idempotency
				if d.Redelivered {
					log.Printf("[Peringatan] Pesan dengan ID %s adalah pesan kirim ulang (Redelivered). Menjalankan pengecekan duplikasi...", d.MessageId)
					// Di sini kita seharusnya memverifikasi di database / Redis cache
				}

				log.Printf("Menerima pesan: %s", d.Body)

				// Simulasi pemrosesan logika bisnis (misal simpan ke DB)
				time.Sleep(200 * time.Millisecond)

				// 5. Mengirimkan Manual ACK setelah sukses memproses
				err := d.Ack(false) // multiple: false berarti hanya meng-ACK pesan spesifik ini
				if err != nil {
					log.Printf("Gagal mengirimkan ACK: %s", err)
				} else {
					log.Printf("✓ Sukses memproses pesan dengan ID %s", d.MessageId)
				}
			}
		}
	}()

	// Menunggu sinyal terminasi OS untuk Graceful Shutdown
	<-sigChan
	log.Println("[*] Menerima sinyal keluar. Memulai proses shutdown yang aman...")
	
	// Membatalkan context agar goroutine tidak membaca pesan baru lagi
	cancel()
	
	// Memberikan waktu toleransi bagi pesan yang sedang berjalan untuk menyelesaikan pemrosesannya
	time.Sleep(1 * time.Second)
	log.Println("[*] Shutdown selesai. Semua koneksi ditutup dengan aman.")
}

Anti-Pattern vs Solusi Praktis di Lingkungan Produksi #

Hindari kesalahan fatal dalam konfigurasi pengiriman pesan berikut untuk menjaga stabilitas klaster:

1. Menyetel Auto-ACK (autoAck = true) untuk Mempercepat Throughput #

Mengaktifkan setelan konfirmasi otomatis dengan asumsi bahwa pemrosesan kode aplikasi kita selalu berjalan sukses tanpa error.

Mengapa ini salah? #

Dengan autoAck = true, RabbitMQ menganggap pesan telah selesai diproses dengan sukses tepat ketika pesan berhasil ditulis ke soket jaringan TCP konsumen. Jika aplikasi konsumen kita mengalami crash di tengah jalan (misal kehabisan memori RAM saat memproses payload, atau koneksi database terputus), pesan tersebut akan hilang selamanya dari antrean broker tanpa bisa dipulihkan kembali. Ini adalah penyebab utama hilangnya data secara misterius.

  • Solusi: Selalu gunakan manual konfirmasi (autoAck = false). Biarkan pesan tetap berstatus Unacknowledged di broker sampai logika bisnis aplikasi kita benar-benar mengembalikan respon sukses dan memanggil perintah d.Ack(false) secara eksplisit.

2. Menggunakan Pull-Model (basic.get) di dalam Loop Pemrosesan Utama #

Menulis kode konsumen yang memanggil basic.get secara berulang-ulang di dalam perulangan for tak terbatas dengan alasan agar kode tampak sederhana seperti membaca data dari database relasional.

Mengapa ini salah? #

Siklus polling konstan membebani disk I/O dan CPU broker karena broker harus terus-menerus memproses pembukaan saluran transaksi pendek untuk setiap pesan tunggal. Ini menghancurkan skalabilitas broker dan membuat latensi pengiriman melonjak tinggi.

  • Solusi: Gunakan model langganan dorong (basic.consume). Kombinasikan dengan goroutine pemroses paralel dan batasi throughput menggunakan setelan pra-ambil Qos(prefetchCount) yang terukur.

Ringkasan #

  • Push-Model lebih unggul — Menggunakan model dorong basic.consume jauh lebih efisien dan memiliki latensi lebih rendah daripada model tarik polling manual basic.get.
  • Peran penting Prefetch QoS — Menetapkan batas pra-ambil pesan mencegah satu instansi konsumen kewalahan menampung beban payload di RAM-nya dan memfasilitasi fair dispatch.
  • Kalkulasi Prefetch — Tentukan nilai pra-ambil yang ideal dengan menghitung perbandingan rasio latensi round-trip (RTT) jaringan terhadap waktu proses riil logika bisnis.
  • Distribusi Beban Fair Dispatch — Prefetch QoS membelokkan alokasi pesan dari konsumen yang sedang sibuk (kuota penuh) ke konsumen lain yang menganggur.
  • Evaluasi Bendera Redelivered — Selalu periksa nilai header Redelivered untuk mengaktifkan mekanisme proteksi idempotensi dan deduplikasi data di sisi aplikasi.
  • Keamanan Manual ACK — Penggunaan manual konfirmasi menjaga integritas data agar pesan tidak hilang saat aplikasi konsumen mengalami crash tidak terduga.

← Sebelumnya: Queueing   Berikutnya: Acknowledgement →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact