Consumer Acknowledgements #

Dalam ekosistem broker RabbitMQ, jaminan pengiriman sisi hulu saja tidaklah cukup. Setelah pesan berhasil diterbitkan oleh produsen dan disimpan oleh broker, bagian akhir dari jaminan pengiriman berada sepenuhnya di sisi hilir: Consumer Acknowledgements (Konfirmasi Konsumen).

Consumer Acknowledgements adalah protokol komunikasi aktif dari aplikasi konsumen kembali ke broker untuk menyatakan status penyelesaian tugas. Melalui mekanisme ini, RabbitMQ dapat mengetahui kapan sebuah pesan aman untuk dihapus dari media penyimpanan permanen, kapan harus dikirimkan kembali ke konsumen lain karena kegagalan koneksi, atau kapan harus diisolasi ke saluran lain. Kesalahan dalam mengonfigurasi dan mengelola konfirmasi konsumen dapat menyebabkan bencana operasional di produksi, mulai dari hilangnya pesan secara diam-diam, kemacetan antrean secara total (queue stall), hingga kebocoran memori RAM yang melumpuhkan broker. Artikel ini akan membahas tuntas semantik konfirmasi konsumen, hubungan erat ACK dengan kuota Prefetch QoS, bahaya kebocoran memori akibat pesan tak terkonfirmasi, urutan transaksi yang aman, serta implementasi Go untuk menangani konfirmasi secara tangguh.

Posisi Konfirmasi Konsumen dalam Siklus Hidup Pesan #

Konfirmasi konsumen adalah operasi penutup yang menentukan akhir perjalanan pesan di dalam antrean broker.

flowchart TD
    Queue["Antrean (Pesan Status: Ready)"] -->|"basic.deliver"| Consumer["Aplikasi Konsumen"]
    Consumer -->|"Pesan Berubah Status: Unacknowledged"| Logic["Eksekusi Logika Bisnis & DB Commit"]
    
    Logic --> Result{"Sukses?"}
    
    Result -->|Ya| Ack["Kirim basic.ack"]
    Result -->|Tidak (Error Sementara)| NackRequeue["Kirim basic.nack (requeue=true)"]
    Result -->|Tidak (Error Permanen)| NackDLX["Kirim basic.nack (requeue=false)"]
    
    Ack --> Delete["Pesan Dihapus dari RAM/Disk Antrean"]
    NackRequeue --> QueueRequeue["Pesan Di-requeue (Kembali ke Antrean)"]
    NackDLX --> DLX["Pesan Dialihkan ke Dead Letter Exchange (DLX)"]

Ketika broker mengirimkan pesan ke konsumen, pesan tersebut bertransisi menjadi status Unacknowledged. Di titik ini, broker melacak kepemilikan pesan berdasarkan koneksi TCP konsumen. Broker tidak akan melepaskan pesan tersebut sampai menerima sinyal balik ACK/NACK, atau mendeteksi terputusnya koneksi TCP.

Semantik Konfirmasi: Manual ACK, NACK, dan REJECT #

Protokol AMQP 0-9-1 menyediakan instruksi konfirmasi yang spesifik bagi konsumen untuk mengomunikasikan nasib akhir pesan ke broker:

1. `basic.ack` (Positive Acknowledgement) #

Perintah ini dikirimkan ketika aplikasi konsumen telah berhasil menyelesaikan pemrosesan pesan dan commit transaksi database bisnis tanpa kendala.

Implikasi: Broker RabbitMQ langsung menghapus pesan tersebut dari antrean (RAM dan disk), serta memulihkan kuota prefetch konsumen untuk menerima pesan berikutnya.

2. `basic.nack` (Negative Acknowledgement) #

Perintah penolakan yang diperkenalkan oleh RabbitMQ untuk menandakan kegagalan proses. Berbeda dengan reject biasa, basic.nack mendukung penolakan pesan secara berkelompok (batch rejection).

Opsi requeue = true: Mengembalikan pesan ke antrean asal di posisi terdepan agar dicoba lagi. Cocok untuk error sementara (koneksi database drop).
Opsi requeue = false: Menghapus pesan dari antrean asal. Jika antrean memiliki Dead Letter Exchange (DLX), pesan akan dialihkan ke DLX tersebut; jika tidak, pesan dibuang selamanya. Cocok untuk error permanen (format JSON rusak).

3. `basic.reject` (Single Rejection) #

Instruksi penolakan standar AMQP 0-9-1. Perilaku sama dengan basic.nack, namun hanya terbatas untuk memproses satu pesan tunggal per panggilan jaringan, tanpa dukungan batching.

Pelacakan State dan Pengelolaan Memori Erlang BEAM #

Di tingkat internal broker RabbitMQ, pengelolaan status Unacknowledged dilakukan menggunakan struktur data memori cepat Erlang yang sangat efisien.

1. Tabel ETS (Erlang Term Storage) per Channel #

Setiap channel konsumen (rabbit_channel) memelihara tabel memori internal ETS untuk mencatat daftar sequence tag dari pesan-pesan in-flight yang sedang diproses oleh konsumen. Tabel ini menyimpan relasi antara delivery_tag lokal, ID fisik pesan di antrean, serta sisa alokasi kuota prefetch.

Pencarian Cepat: Ketika konsumen mengirimkan basic.ack(102), broker melakukan lookup ETS berlatensi mikrodetik untuk menemukan rujukan pesan asli di antrean.
Pembersihan: Setelah entri ditemukan, data di tabel ETS langsung dibersihkan dan ruang memori segera dibebaskan.

2. Pengosongan Memori Off-Heap (`refc` Binaries) #

Payload bodi pesan yang berukuran besar dialokasikan oleh Erlang VM di luar area heap antrean (off-heap storage) menggunakan sistem penghitung referensi (reference counting).

Ketika pesan didorong ke konsumen, counter referensi disetel bertambah.
Selama pesan berstatus Unacknowledged, counter referensi tidak boleh nol, memaksa sistem operasi terus mempertahankan byte data pesan tersebut di RAM fisik server.
Hanya setelah konsumen mengirimkan ACK (yang menurunkan counter referensi menjadi nol), Erlang Garbage Collector (GC) dapat secara asinkron menghapus byte pesan tersebut dari memori fisik RAM server. Kegagalan mengirimkan ACK secara otomatis memblokir proses GC ini, memicu kebocoran memori RAM broker secara perlahan namun pasti.

Korelasi ACK dengan Prefetch QoS dan Kebocoran Memori #

Salah satu area yang paling sering memicu kegagalan operasional di produksi adalah interaksi antara Consumer Acknowledgements dengan Prefetch QoS (Quality of Service).

Bagaimana Prefetch Membatasi Pesan Unacknowledged #

Prefetch QoS membatasi jumlah maksimal pesan berstatus Unacknowledged yang diizinkan mengalir ke satu channel konsumen. Jika kita menetapkan prefetch = 100, broker hanya akan mengirimkan maksimal 100 pesan. Pengiriman pesan ke-101 akan ditangguhkan sampai konsumen mengirimkan minimal satu ACK untuk membebaskan kuota.

Bahaya Lupa Mengirimkan ACK (Memory Leak & Queue Stall) #

Jika dalam penulisan kode aplikasi konsumen kita melakukan kesalahan (misalnya lupa menuliskan pemanggilan ACK pada salah satu cabang penanganan error if/else, atau goroutine pemroses mengalami deadlock sebelum sempat memanggil ACK):

Kemacetan Antrean (Queue Stall): Konsumen akan menerima 100 pesan pertama dan memprosesnya. Namun, karena tidak ada ACK yang dikirimkan kembali ke broker, kuota prefetch konsumen terisi penuh selamanya. Broker RabbitMQ akan berhenti mengirimkan pesan baru ke konsumen tersebut. Aplikasi konsumen akan tampak membeku (stalled) tanpa aktivitas baru, sementara pesan baru di antrean terus menumpuk.
Kebocoran Memori RAM Broker (RAM Memory Leak): RabbitMQ harus terus menahan payload bodi pesan dari 100 pesan Unacknowledged tersebut di dalam memori RAM-nya untuk berjaga-jaga jika konsumen terputus. Jika ada ribuan channel konsumen mengalami masalah lupa ACK ini secara massal, memori RAM broker akan membengkak hingga melampaui batas high memory watermark, memicu alarm memori yang membekukan seluruh aktivitas klaster RabbitMQ (flow control).

Oleh karena itu, memantau metrik jumlah pesan Unacknowledged melalui dashboard Prometheus/Grafana adalah hal yang wajib dilakukan di produksi.

Urutan Eksekusi yang Aman: Komit Database Baru Kirim ACK #

Dalam merancang konsumen transaksional, urutan penulisan kode antara eksekusi database utama dengan pengiriman ACK adalah penentu keselamatan data.

1. Urutan Fatal: ACK Sebelum Commit Database (Anti-Pattern) #

Beberapa developer menulis kode dengan mengirimkan ACK terlebih dahulu untuk membebaskan antrean, baru kemudian memproses transaksi database lokal yang lambat.

flowchart LR
    A["Menerima Pesan"] --> B["Kirim ACK ke Broker"] --> C["Eksekusi Database"] --> D["Commit Gagal! (Data Hilang)"]

Risiko: Jika tepat setelah ACK dikirim (dan broker menghapus pesan dari antreannya), server database lokal kita mengalami crash atau mati listrik saat mencoba melakukan commit data bisnis, pesan tersebut sudah hilang secara fisik dari broker dan tidak dapat dipulihkan kembali. Ini menyebabkan hilangnya data secara permanen.

2. Urutan Aman: Commit Database Baru Kirim ACK (Solusi) #

Urutan penulisan kode yang benar dan aman wajib memprioritaskan database lokal terlebih dahulu:

flowchart LR
    A["Menerima Pesan"] --> B["Eksekusi Database"] --> C["Commit DB Sukses"] --> D["Kirim ACK ke Broker"]

Keuntungan: Jika server database gagal melakukan commit transaksi atau aplikasi konsumen crash sebelum langkah akhir, transaksi database di-rollback secara otomatis dan broker RabbitMQ (yang belum menerima ACK) akan mendeteksi hilangnya koneksi konsumen untuk mengembalikan pesan ke antrean (requeue). Data aman dari risiko kehilangan.

Strategi Pemulihan Channel dan Konsumsi Ulang #

Ketika terjadi gangguan jaringan yang memutus koneksi TCP sejenak, semua pesan Unacknowledged di channel yang terputus tersebut otomatis dikembalikan oleh broker ke antrean asal.

Langkah Rekoveri Konsumen yang Disiplin: #

Deteksi Penutupan: Aplikasi konsumen kita harus mendengarkan notifikasi penutupan channel (NotifyClose).
Buka Koneksi Baru: Lakukan negosiasi ulang koneksi TCP dan buka channel baru.
Reset Prefetch QoS: Setel ulang batas prefetch QoS pada channel baru tersebut.
Daftar Ulang Konsumen: Panggil kembali perintah basic.consume untuk mendaftarkan ulang konsumen kita.
Pembersihan Status Lokal: Kosongkan atau selaraskan kembali cache pelacakan pesan internal di memori konsumen agar tidak bentrok dengan delivery_tag baru yang akan dimulai kembali dari angka 1 oleh broker.

Implementasi Kode Go: Konsumen Transaksional dengan Manual ACK dan Graceful Shutdown #

Berikut adalah contoh implementasi lengkap dalam bahasa Go untuk konsumen yang tangguh. Kode ini mendaftarkan konsumen dengan manual ACK, menyetel batas prefetch QoS, mengevaluasi tipe kesalahan pemrosesan untuk menentukan pilihan requeue, serta menangani sinyal sistem untuk graceful shutdown guna memastikan pesan yang sedang diproses selesai di-ACK dengan aman sebelum koneksi diputus.

package main

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

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

const (
	amqpURL   = "amqp://guest:guest@localhost:5672/"
	queueName = "payment-transactions-queue"
)

type PaymentPayload struct {
	PaymentID string  `json:"payment_id"`
	UserID    string  `json:"user_id"`
	Amount    float64 `json:"amount"`
}

func main() {
	// 1. Membuka Koneksi TCP ke RabbitMQ
	conn, err := amqp.Dial(amqpURL)
	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 batasi kuota maksimal 30 pesan unacknowledged per channel
	err = ch.Qos(
		30,    // prefetch count
		0,     // prefetch size
		false, // global
	)
	if err != nil {
		log.Fatalf("Gagal menyetel QoS Prefetch: %s", err)
	}

	// 3. Deklarasi Antrean Durable
	_, err = ch.QueueDeclare(
		queueName,
		true,  // durable
		false, // auto-delete
		false, // exclusive
		false, // no-wait
		nil,   // arguments
	)
	if err != nil {
		log.Fatalf("Gagal mendeklarasikan antrean: %s", err)
	}

	// 4. Mendaftarkan Konsumen (autoAck = false)
	msgs, err := ch.Consume(
		queueName,
		"",    // consumer tag
		false, // autoAck: false (manual ACK untuk garansi At-Least-Once!)
		false, // exclusive
		false, // no-local
		false, // no-wait
		nil,   // arguments
	)
	if err != nil {
		log.Fatalf("Gagal mendaftarkan konsumen: %s", err)
	}

	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	log.Println("[*] Konsumen aktif. Menunggu pesan...")

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

				// Pemrosesan pesan transaksional
				processPaymentMessage(d)
			}
		}
	}()

	// Menunggu sinyal terminasi OS untuk Graceful Shutdown
	<-sigChan
	log.Println("[*] Menerima sinyal terminasi. Memulai graceful shutdown...")
	
	// Batalkan konteks agar goroutine berhenti menerima pesan baru
	cancel()
	
	// Memberikan waktu toleransi bagi pesan yang sedang berjalan untuk mengirimkan ACK terakhir mereka
	time.Sleep(1500 * time.Millisecond)
	log.Println("[*] Shutdown aman diselesaikan.")
}

func processPaymentMessage(d amqp.Delivery) {
	// Definisikan defer recovery untuk menangani panic tak terduga dalam kode pemroses
	defer func() {
		if r := recover(); r != nil {
			log.Printf("[PANIC] Terjadi kepanikan saat memproses pesan: %v. Mengirimkan NACK requeue...", r)
			// Kirim NACK requeue = true agar pesan dicoba lagi di instance lain
			d.Nack(false, true)
		}
	}()

	var payment PaymentPayload
	
	// 5. Evaluasi Format Payload (Kesalahan Permanen)
	err := json.Unmarshal(d.Body, &payment)
	if err != nil {
		log.Printf("[ERROR] Payload rusak untuk ID Pesan %s: %s. Menolak pesan permanen...", d.MessageId, err)
		// Kirim NACK dengan requeue = false agar pesan dibuang ke Dead Letter Queue (DLQ)
		d.Nack(false, false)
		return
	}

	// 6. Evaluasi Validitas Logika Bisnis (Kesalahan Permanen)
	if payment.Amount <= 0 {
		log.Printf("[ERROR] Nilai pembayaran tidak valid (Amount: %.2f) untuk ID %s. Mengabaikan pesan...", payment.Amount, payment.PaymentID)
		// Kirim NACK requeue = false agar tidak menyumbat antrean utama
		d.Nack(false, false)
		return
	}

	// 7. Eksekusi Logika Transaksi Database Utama
	dbSuccess := savePaymentToDatabase(payment)
	if !dbSuccess {
		log.Printf("[WARN] Kegagalan database sementara saat memproses %s. Melakukan requeue...", payment.PaymentID)
		// Kirim NACK dengan requeue = true agar dicoba lagi nanti setelah database pulih
		d.Nack(false, true)
		return
	}

	// 8. Mengirimkan Manual ACK setelah SUKSES komit database
	err = d.Ack(false)
	if err != nil {
		log.Printf("[ERROR] Gagal mengirimkan ACK ke broker: %s", err)
	} else {
		log.Printf("✓ Transaksi %s sukses terproses dan di-ACK.", payment.PaymentID)
	}
}

func savePaymentToDatabase(payment PaymentPayload) bool {
	// Simulasi fungsi database
	return true
}

Anti-Pattern vs Solusi Praktis di Lingkungan Produksi #

Hindari kesalahan fatal dalam pengelolaan konfirmasi penerimaan pesan berikut:

Anti-Pattern: Mengirim ACK Menggunakan Goroutine Asinkron Tanpa Sinkronisasi Channel Scope #

Menerima pesan di loop utama konsumen, langsung melemparkan tugas pemrosesan dan pemanggilan fungsi d.Ack(false) ke dalam goroutine baru yang dideklarasikan secara asinkron lewat perintah go process(d), tanpa membatasi jumlah goroutine paralel yang menyala.

Mengapa ini salah? #

Pola ini memicu dua bahaya besar:

Channel Scope Exception: Pustaka driver RabbitMQ tidak dirancang untuk aman dari thread (thread-safe) jika beberapa goroutine mencoba memanggil operasi ACK/NACK pada channel yang sama secara bersamaan secara acak. Hal ini dapat memicu error protokol AMQP 406 PRECONDITION_FAILED yang menutup koneksi channel secara paksa.
Kehilangan Kontrol Backpressure: Karena goroutine baru dibuat tanpa batas, kita secara tidak sadar melewati pembatasan kuota prefetch QoS. Konsumen akan terus menyedot pesan dari soket dan menimbunnya di memori RAM internal aplikasi, memicu crash OutOfMemory.

Solusi: Batasi konkurensi goroutine konsumen kita menggunakan Worker Pool Pattern yang terukur. Pastikan pengiriman sinyal ACK/NACK dikembalikan ke goroutine pengelola channel utama secara berurutan, atau gunakan sinkronisasi mutex/channel internal untuk mengamankan panggilan ACK pada objek channel AMQP.

Ringkasan #

Kontrol Keandalan Konsumen — Consumer Acknowledgements menentukan kapan pesan aman dihapus dari antrean broker (basic.ack) atau harus dikirim ulang (basic.nack).
Mekanisme manual ACK — Merupakan best practice wajib untuk sistem produksi transaksional. Menyetel Auto-Ack (autoAck = true) sangat rentan memicu hilangnya pesan saat aplikasi crash.
Dampak Lupa ACK — Mengakibatkan kebocoran memori RAM pada broker RabbitMQ dan menghabiskan kuota prefetch QoS yang membekukan aliran data antrean (queue stall).
Aturan Urutan Transaksi — Pastikan kode aplikasi selalu menyelesaikan logika bisnis dan commit transaksi database lokal terlebih dahulu, baru kemudian mengirimkan instruksi ACK ke broker.
Penolakan dengan Requeue — Gunakan requeue = true untuk error infrastruktur sementara, dan gunakan requeue = false (mengalihkan ke DLX) untuk error format data permanen.
Konkurensi Goroutine Aman — Hindari pemanggilan ACK secara acak lintas goroutine tanpa sinkronisasi channel untuk mencegah exception PRECONDITION_FAILED yang memutus koneksi.

← Sebelumnya: Publisher Confirms Berikutnya: Requeue vs Drop →