Use Quorum #

Dalam perancangan arsitektur sistem berbasis pesan di tingkat produksi, menjamin keandalan pesan (message reliability) dan ketersediaan tinggi (high availability) adalah prioritas utama. Ketika kita menggunakan antrean RabbitMQ, salah satu keputusan paling mendasar yang harus kita ambil adalah menentukan jenis antrean (queue type) yang tepat untuk menyimpan muatan data transaksi kita. Selama bertahun-tahun, pilihan default jatuh pada Classic Queue yang dikonfigurasi sebagai durable dengan pesan yang dikirim secara persistent. Pendekatan ini memang terbukti cepat dan mudah dioperasikan. Namun, untuk beban kerja bisnis modern yang kritikal—di mana kehilangan satu pesan saja dapat menyebabkan kerugian finansial atau ketidaksesuaian data yang serius—praktik terbaik modern RabbitMQ adalah selalu gunakan Quorum Queue.

Quorum Queue dirancang khusus untuk mengatasi keterbatasan inheren dari Classic Queue dalam lingkungan klaster multi-node terdistribusi. Banyak tim developer yang salah mengasumsikan bahwa dengan mencentang opsi durable dan persistent, data mereka sudah sepenuhnya aman dari segala bentuk kegagalan. Durability dan persistensi saja sebenarnya hanya melindungi data dari insiden pemadaman listrik atau restart broker biasa pada satu node tunggal. Kedua parameter tersebut tidak memberikan perlindungan sama sekali jika node tempat antrean itu berada mengalami kerusakan perangkat keras permanen (hardware failure) atau kehilangan koneksi dari klaster secara total. Untuk mendapatkan ketahanan klaster yang sesungguhnya, kita membutuhkan replikasi log yang konsisten, dan itulah fondasi utama yang ditawarkan oleh Quorum Queue.

Mengapa Classic Durable Queue Saja Tidak Cukup? #

Untuk memahami mengapa kita harus bermigrasi dari Classic Queue, kita perlu membedah anatomi penyimpanan data pada Classic Queue durable di dalam klaster RabbitMQ. Meskipun kita menjalankan klaster dengan 3 node, Classic Queue secara default bersifat lokal. Antrean tersebut secara fisik hanya bertempat tinggal di satu node tertentu, yang disebut sebagai home node. Node-node lainnya dalam klaster hanya menyimpan pointer metadata yang merujuk ke home node tersebut.

Ketika produsen mempublikasikan pesan ke exchange, broker merutekan pesan tersebut melewati jaringan klaster untuk ditulis ke disk pada home node antrean. Jika home node tersebut tiba-tiba mati secara permanen akibat kegagalan piringan disk (disk corruption) atau kerusakan VM:

1. Kehilangan Ketersediaan (Availability Loss): Antrean tersebut akan langsung berstatus Down dan tidak dapat diakses sama sekali oleh produsen maupun konsumen. Node lain tidak memiliki salinan pesan untuk melanjutkan pemrosesan.
2. Kehilangan Data secara Senyap (Silent Data Loss): Meskipun server berhasil dinyalakan kembali beberapa jam kemudian, jika disk mengalami kerusakan fisik, pesan-pesan penting yang mengendap di dalamnya akan hilang selamanya.
3. Masalah Antrean Cermin (Classic Mirrored Queues Deprecation): RabbitMQ sebelumnya memiliki solusi replikasi bernama Mirrored Queues (aktif menggunakan kebijakan ha-mode). Namun, mirrored queues memiliki masalah sinkronisasi yang sangat rentan memicu blocking saat terjadi pemulihan jaringan, sering kali memicu kehilangan data saat pemimpin antrean mati sebelum sinkronisasi selesai, dan performanya sangat tidak dapat diprediksi. Fitur Mirrored Queues kini telah didepresiasi sepenuhnya dan dihapus dari versi RabbitMQ modern, digantikan seutuhnya oleh Quorum Queue.

Mekanisme Replikasi Quorum Queue & Konsensus Raft #

Quorum Queue dibangun di atas implementasi algoritma konsensus Raft (menggunakan pustaka Erlang bernama ra). Raft adalah protokol konsensus terdistribusi yang dirancang agar mudah dipahami dan menjamin konsistensi data yang sangat kuat di seluruh node klaster.

Di bawah protokol Raft, replikasi data Quorum Queue beroperasi dengan alur kerja yang sangat terstruktur:

flowchart TD
    Producer["Aplikasi Produsen"] -->|"1. Publish Pesan"| Leader["Raft Leader (Home Node)"]
    Leader -->|"2. Tulis ke Disk WAL Lokal"| LeaderWAL[("Leader Disk WAL")]
    
    Leader -->|"3. Replikasi Log Raft"| Follower1["Raft Follower Node A"]
    Leader -->|"3. Replikasi Log Raft"| Follower2["Raft Follower Node B"]
    
    Follower1 -->|"4. Tulis ke Disk WAL"| Follower1WAL[("Follower A Disk WAL")]
    Follower2 -->|"4. Tulis ke Disk WAL"| Follower2WAL[("Follower B Disk WAL")]
    
    Follower1WAL -->|"5. Kirim ACK Tulis"| Leader
    Follower2WAL -->|"5. Kirim ACK Tulis"| Leader
    
    Leader -->|"6. Komit ke State Machine & Kirim Confirm"| Producer

    style Leader stroke:#0288d1,stroke-width:2px
    style Follower1 stroke:#7b1fa2,stroke-width:2px
    style Follower2 stroke:#7b1fa2,stroke-width:2px

1. Peran Leader dan Followers #

Setiap Quorum Queue terdiri dari satu instans Leader (yang menangani semua operasi penulisan dari produsen dan pembacaan dari konsumen) serta beberapa instans Follower yang disebarkan secara merata di seluruh node klaster lainnya.

2. Aturan Mayoritas (Write Quorum) #

Ketika produsen mengirimkan pesan, node Leader tidak langsung mengirimkan konfirmasi sukses (publisher confirm) ke produsen. Leader pertama-tama menulis pesan tersebut ke Write-Ahead Log (WAL) disk lokalnya, kemudian menyebarkan log replikasi tersebut ke seluruh node Follower secara paralel.

Pesan dianggap telah aman berkomitmen (committed state) jika dan hanya jika mayoritas dari total replika klaster telah berhasil menulis log tersebut ke disk fisik masing-masing dan mengirimkan sinyal persetujuan kembali ke Leader. Rumus mayoritas quorum didefinisikan sebagai: $$\lfloor N/2 \rfloor + 1$$ Di mana $N$ adalah jumlah replika antrean yang dikonfigurasi. Untuk klaster 3 node, mayoritas adalah 2 node (Leader + 1 Follower). Begitu mayoritas tercapai, Leader akan mengirimkan sinyal konfirmasi (Publisher Confirm) ke produsen dan membuat pesan tersebut siap dikonsumsi oleh aplikasi konsumen.

3. Failover Tanpa Hambatan (Automatic Leader Election) #

Jika node Leader mengalami crash, node-node Follower yang tersisa akan mendeteksi ketiadaan sinyal detak jantung (heartbeat timeout) dari Leader. Mereka akan segera mengadakan pemilihan suara (leader election) secara otomatis untuk memilih Leader baru dari salah satu Follower yang memiliki salinan log paling mutakhir (synchronized replica). Proses ini berjalan dalam hitungan milidetik, tanpa adanya data yang hilang, dan tanpa membutuhkan intervensi manual dari tim operasional.

Konfigurasi Ukuran Klaster Node yang Tepat #

Saat merancang klaster RabbitMQ yang menggunakan Quorum Queue, jumlah node yang kita gunakan adalah faktor penentu ketersediaan sistem. Kita wajib menggunakan jumlah node ganjil, dengan rekomendasi minimal adalah 3 node atau 5 node di lingkungan produksi.

Mengapa Harus Jumlah Ganjil? #

Aturan matematika konsensus Raft menuntut mayoritas suara untuk mengambil keputusan. Mari kita bandingkan konfigurasi klaster genap vs ganjil saat terjadi kegagalan node:

• Kasus Klaster 2 Node (Genap - Sangat Tidak Direkomendasikan):
  • Rumus Quorum: $\lfloor 2/2 \rfloor + 1 = 2$ node.
  • Artinya, kedua node harus aktif agar pesan dapat berkomitmen. Jika 1 node mati, node yang tersisa (50% dari klaster) tidak dapat membentuk mayoritas suara (>50%). Akibatnya, seluruh proses penulisan pesan pada Quorum Queue akan membeku total. Klaster 2 node memiliki toleransi kegagalan sebesar nol node.
• Kasus Klaster 3 Node (Ganjil - Standar Produksi):
  • Rumus Quorum: $\lfloor 3/2 \rfloor + 1 = 2$ node.
  • Jika 1 node mati, masih ada 2 node yang aktif. 2 dari 3 node adalah mayoritas (66%). Klaster tetap berfungsi penuh dan dapat memilih Leader baru. Toleransi kegagalan klaster 3 node adalah 1 node.
• Kasus Klaster 5 Node (Ganjil - Skala Besar):
  • Rumus Quorum: $\lfloor 5/2 \rfloor + 1 = 3$ node.
  • Jika 2 node mati bersamaan, masih ada 3 node yang aktif (60%). Klaster tetap stabil. Toleransi kegagalan klaster 5 node adalah 2 node.

Analisis Trade-off Performa (Biaya Sebuah Keandalan) #

Memilih Quorum Queue berarti kita memprioritaskan konsistensi dan keandalan data di atas aspek lainnya. Pilihan ini membawa konsekuensi operasional (trade-offs) yang harus kita kelola secara bijaksana:

1. Latensi Konfirmasi (Publisher Confirm Latency) Lebih Tinggi #

Karena setiap pesan harus melewati proses serialisasi jaringan ke node pengikut dan menunggu operasi penulisan disk WAL di mayoritas node selesai, waktu respon konfirmasi ke produsen akan meningkat dibanding Classic Queue. Classic Queue hanya memerlukan satu kali penulisan disk lokal sebelum membalas konfirmasi. Kita harus mengoptimalkan produsen untuk menggunakan Asynchronous Confirms (menghindari mode sinkronisasi satu-per-satu) agar throughput tetap terjaga tinggi.

2. Amplifikasi Penulisan Disk (Disk Write Amplification) #

Setiap pesan Quorum Queue ditulis ke disk beberapa kali di node yang berbeda. Ini berarti utilisasi disk I/O klaster akan meningkat secara signifikan. Pastikan node RabbitMQ produksi menggunakan media penyimpanan berkecepatan tinggi seperti SSD NVMe dengan kapasitas IOPS yang memadai, serta memantau metrik latency I/O secara berkala.

3. Penggunaan Memori RAM dan CPU yang Lebih Besar #

Menjalankan mesin konsensus Raft per antrean membutuhkan alokasi memori heap Erlang yang lebih besar untuk memelihara state mesin, mendeteksi detak jantung koneksi Raft, dan mengelola buffer log di memori. Jangan gunakan Quorum Queue untuk ribuan antrean dinamis yang berumur pendek; Quorum Queue didesain untuk topologi statis berumur panjang dengan volume pesan tinggi.

Implementasi Quorum Queue pada Aplikasi Go #

Saat mengimplementasikan Quorum Queue pada aplikasi Go, kita cukup menyisipkan argumen kustom "x-queue-type": "quorum" saat melakukan deklarasi antrean. Selain itu, sangat direkomendasikan untuk menyertakan argumen "x-delivery-limit" untuk membatasi jumlah percobaan ulang otomatis oleh broker jika terjadi kegagalan pemrosesan pesan yang berulang (poison message protection).

Berikut adalah kode Go lengkap untuk mendeklarasikan dan memproses pesan menggunakan Quorum Queue secara aman:

package main

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

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

const (
	amqpURI      = "amqp://admin:securepassword@rabbitmq-cluster:5672/"
	queueName    = "queue.order.process"
	exchangeName = "exchange.order.direct"
	routingKey   = "order.created"
)

type OrderConsumer struct {
	conn    *amqp.Connection
	channel *amqp.Channel
	close   chan *amqp.Error
}

// InitQueue membangun koneksi dan mendeklarasikan Quorum Queue secara aman
func (oc *OrderConsumer) InitQueue() error {
	var err error
	log.Println("Menghubungkan ke cluster RabbitMQ...")
	oc.conn, err = amqp.Dial(amqpURI)
	if err != nil {
		return err
	}

	oc.channel, err = oc.conn.Channel()
	if err != nil {
		oc.conn.Close()
		return err
	}

	// 1. Deklarasikan Exchange Utama
	log.Printf("Mendeklarasikan exchange '%s'...\n", exchangeName)
	err = oc.channel.ExchangeDeclare(
		exchangeName,
		"direct",
		true,  // durable
		false, // auto-deleted
		false, // internal
		false, // no-wait
		nil,   // arguments
	)
	if err != nil {
		oc.channel.Close()
		oc.conn.Close()
		return err
	}

	// 2. Deklarasikan Quorum Queue dengan konfigurasi parameter ketahanan tingkat tinggi
	log.Printf("Mendeklarasikan Quorum Queue '%s'...\n", queueName)
	queueArgs := amqp.Table{
		"x-queue-type":     "quorum", // Mengaktifkan tipe Quorum Queue berbasis Raft
		"x-delivery-limit": 5,        // Maksimal percobaan konsumsi sebelum otomatis dibuang/ke DLQ
	}

	_, err = oc.channel.QueueDeclare(
		queueName,
		true,  // durable (wajib disetel ke true untuk quorum queue)
		false, // auto-delete (wajib disetel ke false)
		false, // exclusive (wajib disetel ke false)
		false, // no-wait
		queueArgs,
	)
	if err != nil {
		oc.channel.Close()
		oc.conn.Close()
		return err
	}

	// 3. Ikat (Bind) Quorum Queue ke Exchange
	log.Println("Mengikat queue ke exchange...")
	err = oc.channel.QueueBind(
		queueName,
		routingKey,
		exchangeName,
		false,
		nil,
	)
	if err != nil {
		oc.channel.Close()
		oc.conn.Close()
		return err
	}

	// Batasi prefetch limit QoS agar consumer memproses pesan secara terkendali
	err = oc.channel.Qos(10, 0, false)
	if err != nil {
		oc.channel.Close()
		oc.conn.Close()
		return err
	}

	oc.close = make(chan *amqp.Error)
	oc.channel.NotifyClose(oc.close)

	return nil
}

func (oc *OrderConsumer) StartConsume(ctx context.Context) {
	deliveries, err := oc.channel.Consume(
		queueName,
		"order-worker-1",
		false, // auto-ack disetel ke false agar pesan aman jika consumer crash
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		log.Printf("Gagal memulai consume loop: %v\n", err)
		return
	}

	log.Println("Worker aktif. Menunggu transaksi masuk...")

	for {
		select {
		case <-ctx.Done():
			log.Println("Menghentikan worker secara graceful...")
			return
		case errClosed := <-oc.close:
			if errClosed != nil {
				log.Printf("Channel koneksi ditutup secara sepihak: %v. Reconnecting...\n", errClosed)
				oc.Reconnect(ctx)
				return
			}
		case msg, ok := <-deliveries:
			if !ok {
				return
			}
			oc.handleOrder(msg)
		}
	}
}

func (oc *OrderConsumer) Reconnect(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return
		default:
			log.Println("Mencoba menghubungkan kembali ke cluster dalam 5 detik...")
			time.Sleep(5 * time.Second)

			err := oc.InitQueue()
			if err == nil {
				log.Println("✓ Berhasil terhubung kembali.")
				go oc.StartConsume(ctx)
				return
			}
			log.Printf("Gagal menghubungkan kembali: %v\n", err)
		}
	}
}

func (oc *OrderConsumer) handleOrder(msg amqp.Delivery) {
	// Evaluasi header x-delivery-count untuk memantau status kegagalan pesan
	deliveryCount, exists := msg.Headers["x-delivery-count"]
	if exists {
		log.Printf("[METRIK] Pesan ini sudah dicoba sebanyak %v kali\n", deliveryCount)
	}

	log.Printf("[TRANSAKSI] Memproses order: ID=%s, Payload=%s\n", msg.CorrelationId, string(msg.Body))
	
	// Simulasikan logika penyimpanan ke database transaksi
	err := oc.saveToDatabase(msg.Body)
	if err != nil {
		log.Printf("[GAGAL] Gagal memproses order ID %s: %v. Mengirimkan NACK...\n", msg.CorrelationId, err)
		
		// Kirim NACK dengan requeue = false agar diteruskan ke DLQ jika melebihi delivery limit
		msg.Nack(false, false)
		return
	}

	// Kirim ACK jika proses transaksi berhasil
	msg.Ack(false)
	log.Printf("[SUKSES] ACK dikirim untuk order ID %s\n", msg.CorrelationId)
}

func (oc *OrderConsumer) saveToDatabase(payload []byte) error {
	// Simulasikan I/O database latency
	time.Sleep(50 * time.Millisecond)
	return nil
}

func (oc *OrderConsumer) Close() {
	if oc.channel != nil {
		oc.channel.Close()
	}
	if oc.conn != nil {
		oc.conn.Close()
	}
}

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

	oc := &OrderConsumer{}
	err := oc.InitQueue()
	if err != nil {
		log.Fatalf("Gagal inisialisasi awal ke cluster RabbitMQ: %v\n", err)
	}
	defer oc.Close()

	go oc.StartConsume(ctx)

	stop := make(chan os.Signal, 1)
	signal.Notify(stop, os.Interrupt, syscall.SIGTERM)

	<-stop
	log.Println("Menerima sinyal shutdown...")
	cancel()
	time.Sleep(1 * time.Second)
}

Perbandingan Fitur: Classic Queue vs Quorum Queue #

Berikut adalah tabel matriks perbandingan mendalam untuk membantu kita membedakan kemampuan Classic Queue durable dengan Quorum Queue:

Ringkasan #

• Durability vs Replikasi — Durable dan persistent hanya melindungi data dari restart server tunggal, sementara Quorum Queue melindungi data dari kegagalan fisik hardware atau crash node.
• Gunakan Raft Konsensus — Quorum Queue mereplikasi log penulisan pesan ke mayoritas node klaster sebelum membalas ACK ke produsen, memastikan integritas data tetap konsisten.
• Gunakan Jumlah Node Ganjil — Selalu jalankan klaster produksi dengan 3 atau 5 node untuk memungkinkan pembentukan suara mayoritas yang stabil saat terjadi failover.
• Pahami Trade-off Performa — Keandalan Quorum Queue menuntut biaya performa berupa confirm latency yang lebih tinggi serta beban I/O disk yang lebih intensif dibanding classic queue.

← Sebelumnya: Design First   Berikutnya: Message Size →