Publishing #

Dalam siklus hidup pesan (message lifecycle) di ekosistem RabbitMQ, perjalanan sebuah data tidak dimulai dari saat ia berada di dalam antrean (queue). Perjalanan tersebut dimulai jauh sebelum itu, tepat pada satu momen krusial di sisi produsen: saat pesan dikirimkan melalui operasi Publishing.

Meskipun terlihat seperti operasi kirim-data sederhana di tingkat kode aplikasi, proses Publishing melibatkan serangkaian interaksi protokol tingkat rendah, segmentasi paket biner, validasi aturan rute oleh broker, dan mekanisme koordinasi ketahanan data. Memahami fase Publishing secara mendalam sangat penting untuk mencegah terjadinya kehilangan data secara tidak sengaja (silent drop), menangani hambatan lalu lintas (backpressure), serta merancang jaminan pengiriman yang andal sejak pesan pertama kali dilahirkan. Artikel ini akan membedah secara granular anatomi Publishing, pembagian frame AMQP, penanganan pesan tak ter-rute via mandatory flag, asinkronisasi Publisher Confirms, dan implementasi kode Go kelas produksi.

Anatomi Frame AMQP dan Ingress Pipeline #

RabbitMQ didasarkan pada protokol AMQP (Advanced Message Queuing Protocol) 0-9-1. Saat produsen memanggil fungsi publikasi pesan (seperti basic.publish), pesan tidak dikirimkan sebagai satu aliran biner mentah yang besar. Protokol AMQP membagi pesan menjadi beberapa unit paket biner terstruktur yang disebut Frame.

flowchart TD
    Producer["Produsen (Publish)"] --> Frame1["Frame 1: Method Frame (basic.publish)"]
    Producer --> Frame2["Frame 2: Content Header Frame (Properties & Size)"]
    Producer --> Frame3["Frame 3: Content Body Frame (Payload 0..N)"]
    
    subgraph Connection["Koneksi TCP / AMQP Channel"]
        Frame1 --> Transport["Socket Send Buffer"]
        Frame2 --> Transport
        Frame3 --> Transport
    end

    Transport --> Broker["Broker Ingress Pipeline"]

Jenis-Jenis Frame AMQP #

Untuk setiap pesan yang dipublikasikan, minimal terdapat tiga jenis frame yang dikirimkan secara berurutan melalui socket TCP:

Method Frame: Frame pembuka yang membawa instruksi perintah protokol. Untuk publikasi, frame ini berisi kode kelas dan metode AMQP (misalnya, Kelas basic = 60, Metode publish = 40), nama exchange tujuan, dan routing key.
Content Header Frame: Frame kedua yang membawa metadata pesan. Frame ini mendefinisikan ukuran total payload pesan (body size) serta atribut properti basic pesan seperti contentType (misalnya, application/json), deliveryMode (persistent atau transient), headers kustom, timestamp, dan messageId.
Content Body Frame: Frame ketiga dan seterusnya yang membawa payload pesan biner aktual. Jika ukuran payload pesan melebihi batas ukuran frame yang disepakati, payload akan dipecah menjadi beberapa Content Body Frame berturut-turut.

Negosiasi Ukuran Frame (`frame_max`) #

Selama fase jabat tangan koneksi (connection handshake) antara aplikasi klien dan broker, kedua belah pihak melakukan negosiasi parameter frame_max. Parameter ini menentukan ukuran maksimum (dalam byte) dari satu frame AMQP yang diizinkan untuk melewati koneksi tersebut.

Batas Default: Secara bawaan, RabbitMQ menyetel batas frame_max sebesar 131.072 byte (128 KB).
Segmentasi Payload: Jika kita mempublikasikan pesan berukuran 500 KB dengan frame_max default 128 KB, pustaka klien akan secara otomatis memecah pesan tersebut menjadi 1 Method Frame, 1 Content Header Frame, dan 4 Content Body Frame terpisah saat dikirim melalui socket.
Bahaya Menurunkan frame_max: Menyetel frame_max terlalu kecil (misalnya di bawah 4 KB) akan meningkatkan overhead header TCP/AMQP secara signifikan karena broker harus menyusun ulang potongan frame yang sangat banyak di sisi memori RAM. Sebaliknya, menyetel terlalu besar dapat menyebabkan konsumsi buffer memori instan melonjak saat menangani ribuan koneksi paralel.

Mekanisme Mandatory Flag & Return #

Setelah seluruh frame penyusun pesan tiba di pintu masuk broker (ingress pipeline), broker akan membaca Method Frame untuk mencari nama exchange tujuan dan mengevaluasi binding yang terdaftar. Salah satu risiko terbesar pada tahap ini adalah ketika pesan dikirim ke exchange yang ada, namun tidak ada satu pun antrean (queue) yang terikat ke exchange tersebut dengan kriteria routing key yang cocok. Kondisi ini disebut Unroutable Message.

Perilaku broker dalam menangani Unroutable Message dikendalikan oleh argumen mandatory flag pada perintah publish.

1. Perilaku `mandatory = false` (Default) #

Jika kita mempublikasikan pesan dengan menyetel mandatory = false (atau tidak mengonfigurasinya):

Broker akan mengevaluasi aturan rute.
Jika tidak ada antrean yang cocok, broker akan membuang pesan tersebut secara instan tanpa memberikan notifikasi kesalahan apa pun ke produsen.
Ini adalah pemicu utama hilangnya pesan secara senyap (silent drop) yang sering kali baru disadari setelah sistem berjalan di produksi.

2. Perilaku `mandatory = true` #

Jika kita menyetel mandatory = true saat melakukan publish:

Broker berkewajiban untuk memastikan pesan berhasil masuk ke minimal satu antrean.
Jika hasil evaluasi rute menunjukkan tidak ada antrean yang cocok, broker akan mengirimkan kembali pesan tersebut ke produsen melalui metode asinkron basic.return.
Frame basic.return ini membawa data pesan lengkap beserta kode status kesalahan (misalnya, 312 NO_ROUTE) dan alasan penolakan.
Aplikasi produsen harus mendaftarkan listener khusus untuk menangkap pesan yang memantul (returned messages) ini dan melakukan penanganan darurat (seperti mencatat log kesalahan atau memasukkannya ke antrean lokal).

flowchart LR
    A["mandatory=true"] --> B["Publish"] --> C["Exchange"] -->|"No Match Queue"| D["basic.return"] --> E["Produsen"]

Publisher Confirms & Flow Control (Backpressure) #

Menerbitkan pesan ke broker tidak menjamin pesan tersebut telah aman tersimpan di media penyimpanan persisten. Jaringan TCP dapat terputus sepihak, atau broker bisa mengalami crash sesaat setelah menerima data dari socket namun sebelum data sempat ditulis ke disk. Untuk menjamin ketahanan transmisi, kita harus mengaktifkan Publisher Confirms.

Cara Kerja Publisher Confirms #

Publisher Confirms (dikenal juga sebagai Confirm Mode) adalah ekstensi protokol AMQP 0-9-1 yang mengamankan pengiriman pesan dari produsen ke broker secara asinkron.

Aktivasi: Produsen mengirimkan instruksi confirm.select ke broker untuk mengaktifkan confirm mode pada channel tersebut.
Monotonic Sequence Number: Setiap kali pesan dipublikasikan melalui channel yang telah aktif confirm mode-nya, broker memberikan nomor urut biner (sequence number) mulai dari angka 1.
Confirm ACK: Broker mengirimkan frame basic.ack kembali ke produsen berisi sequence number pesan setelah:
- Pesan transient berhasil masuk ke seluruh antrean tujuan di RAM.
- Pesan persistent berhasil tertulis dan di-fsync ke disk fisik pada antrean durable.
- Pesan pada Quorum Queue berhasil mencapai konsensus mayoritas replika klaster.
Confirm NACK: Jika broker mengalami kegagalan internal (misalnya, disk penuh atau database Mnesia terkunci), broker akan mengirimkan frame basic.nack untuk memberi tahu produsen bahwa pesan gagal diamankan.

Backpressure dan Flow Control #

Ketika volume penerbitan pesan sangat tinggi, broker dapat kehabisan kapasitas memori RAM atau disk. Untuk mencegah crash, RabbitMQ menerapkan mekanisme pembatasan laju pengiriman (backpressure):

Memory Alarm: Jika memori RAM broker melewati watermark, broker akan memblokir (block) koneksi produsen yang aktif melakukan publish.
TCP Window Throttling: Broker berhenti membaca data dari socket TCP produsen. Ini menyebabkan socket buffer di OS produsen penuh, yang kemudian secara otomatis memperlambat laju pengiriman di sisi aplikasi produsen (TCP backpressure). Klien akan mendeteksi status koneksi sebagai blocked.

Implementasi Kode Go: Publisher Confirms & Return Handling #

Berikut adalah contoh kode Go lengkap menggunakan pustaka github.com/rabbitmq/amqp091-go untuk mengaktifkan Confirm Mode, menyetel mandatory = true, menangani pesan yang memantul (basic.return), serta mendengarkan konfirmasi ACK/NACK secara asinkron.

package main

import (
	"context"
	"log"
	"time"

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

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

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

	// 3. Mengaktifkan Confirm Mode pada Channel
	err = ch.Confirm(false) // false artinya non-noWait
	if err != nil {
		log.Fatalf("Gagal mengaktifkan Publisher Confirms: %s", err)
	}

	// 4. Mendaftarkan Listener untuk Publisher Confirms (ACK/NACK)
	confirmChan := ch.NotifyPublish(make(chan amqp.Confirmation, 100))

	// 5. Mendaftarkan Listener untuk Returned Messages (basic.return)
	// Channel ini akan menerima data jika mandatory=true dan pesan gagal dirutekan
	returnChan := ch.NotifyReturn(make(chan amqp.Return, 100))

	// Goroutine untuk menangani pesan yang memantul (Unroutable Messages)
	go func() {
		for r := range returnChan {
			log.Printf("⚠ PESAN MEMANTUL (basic.return)! Kode: %d, Alasan: %s, RoutingKey: %s, Payload: %s",
				r.ReplyCode, r.ReplyText, r.RoutingKey, string(r.Body))
			// Tindakan darurat: masukkan ke database lokal atau kirim ke DLQ manual
		}
	}()

	// Goroutine untuk menangani konfirmasi asinkron dari broker
	go func() {
		for c := range confirmChan {
			if c.Ack {
				log.Printf("✓ Pesan dengan Sequence Number %d berhasil diamankan oleh broker.", c.DeliveryTag)
			} else {
				log.Printf("✗ Pesan dengan Sequence Number %d GAGAL diamankan (NACK)!", c.DeliveryTag)
			}
		}
	}()

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	// Skenario A: Mengirim pesan ke exchange yang ada tetapi dengan routing key yang salah
	// (Memicu basic.return karena mandatory = true)
	payloadA := []byte(`{"event":"order_created","id":"123"}`)
	err = ch.PublishWithContext(ctx,
		"amq.topic",       // Menggunakan exchange default AMQP topic
		"wrong.route.key", // Routing key salah (tidak ada queue yang terikat)
		true,              // mandatory = true: Mencegah silent drop!
		false,             // immediate (deprecated di AMQP 3.0+)
		amqp.Publishing{
			DeliveryMode: amqp.Persistent,
			ContentType:  "application/json",
			Body:         payloadA,
		},
	)
	if err != nil {
		log.Fatalf("Gagal mengirim pesan A: %s", err)
	}

	// Memberikan waktu bagi proses asinkron untuk berjalan sebelum aplikasi keluar
	time.Sleep(3 * time.Second)
}

Metrik Monitoring & TCP Driver Pipeline #

Untuk menjamin keandalan sistem Publishing di lingkungan produksi yang memiliki beban kerja tinggi, kita harus memahami metrik monitoring penting dan bagaimana driver TCP Erlang memproses frame pesan.

Alur Socket Driver Erlang #

Di tingkat internal runtime BEAM, koneksi TCP dikelola oleh proses rabbit_reader. Ketika produsen memublikasikan pesan:

Frame Parsing: Proses rabbit_reader membaca biner dari socket TCP dan mem-parsing Method Frame untuk mengidentifikasi perintah basic.publish.
Stateful Assembly: Proses pembaca kemudian beralih ke state menunggu Content Header Frame untuk mengekstrak informasi ukuran body (body_size) dan properti dasar.
Body Collection: Terakhir, ia membaca serangkaian Content Body Frame hingga akumulasi ukuran biner bodi cocok dengan body_size yang dilaporkan di header.

Jika koneksi TCP mengalami kegagalan atau terputus di tengah jalan sebelum seluruh Content Body Frame selesai dikirim, proses rabbit_reader akan langsung mendeteksi kondisi tidak lengkap ini, membuang potongan frame yang sempat terkumpul di memori RAM, dan tidak akan meneruskan pesan setengah jadi tersebut ke exchange. Ini memberikan jaminan integritas data transaksi.

Metrik Kunci untuk Fase Publishing #

Kita wajib memantau metrik-metrik berikut pada dasbor Prometheus/Grafana untuk mendeteksi degradasi performa Publishing sejak dini:

rabbitmq_messages_published_total: Total pesan yang dipublikasikan. Metrik ini membantu mengukur throughput ingress sistem kita.
rabbitmq_messages_returned_total: Jumlah pesan yang memantul (unroutable) akibat kegagalan rute binding. Jika metrik ini melonjak, itu menandakan adanya ketidakcocokan routing key di sisi produsen atau rusaknya topologi binding di broker.
rabbitmq_connection_blocked: Indikator boolean yang bernilai 1 ketika produsen diblokir oleh broker akibat alarm memori atau disk alarm terpicu. Ini adalah sinyal kritis bahwa broker sedang mengalami overload.
rabbitmq_channel_confirm_pending: Jumlah pesan yang telah dipublikasikan namun belum mendapatkan ACK/NACK konfirmasi dari broker. Jika angka ini terus merangkak naik, itu berarti media penyimpanan disk broker mulai mengalami kelambatan menulis data (I/O bottleneck).

Anti-Pattern vs Solusi Kelas Produksi #

Dalam implementasi arsitektur pesan, hindari kesalahan desain Publishing berikut:

1. Menyisipkan `mandatory = false` untuk Pesan Transaksional Kritis #

Mempublikasikan pesan pembayaran atau pembuatan akun pengguna tanpa menyetel flag mandatory = true.

Mengapa ini salah? #

Jika terjadi kesalahan konfigurasi topologi antrean di broker (misalnya antrean terhapus atau binding terlepas karena ketidaksengajaan operasional), broker akan membuang pesan transaksi tersebut secara diam-diam tanpa ada galat (error) di sisi produsen. Produsen berasumsi transaksi aman, padahal data telah lenyap dari sistem.

Solusi: Selalu setel mandatory = true untuk seluruh data transaksional penting. Kombinasikan dengan pendaftaran listener NotifyReturn di sisi aplikasi produsen untuk menangani pengiriman ulang atau pencatatan log peringatan jika pesan memantul.

2. Mengirim Pesan Secara Sinkron Satu per Satu (Synchronous Blocking Publish) #

Aplikasi produsen mempublikasikan pesan, lalu memblokir thread eksekusi aplikasi untuk menunggu konfirmasi ACK dari broker sebelum mengirimkan pesan berikutnya secara berurutan.

Mengapa ini salah? #

Pola ini menghancurkan performa pengiriman pesan. Produsen hanya akan mampu mengirimkan beberapa puluh pesan per detik karena harus menunggu latensi jaringan RTT dan proses fsync disk biner broker selesai untuk setiap pesan tunggal.

Solusi: Gunakan pola pengiriman asinkron (Asynchronous Publisher Confirms). Kirimkan pesan secara terus-menerus tanpa memblokir thread, dan proses ACK/NACK yang masuk secara asinkron di goroutine/thread terpisah dengan mencocokkan nomor urut sequence biner (DeliveryTag).

Ringkasan #

Segmentasi Frame AMQP — Setiap pesan dipecah menjadi tiga jenis frame biner saat melewati socket TCP: Method Frame (instruksi), Content Header Frame (metadata/properties), dan Content Body Frame (payload biner).
Negosiasi frame_max — Klien dan broker menegosiasikan ukuran frame maksimum (frame_max) saat handshake koneksi (default 128 KB). Setelan yang terlalu kecil meningkatkan overhead CPU broker.
Cegah Silent Drop via Mandatory — Setel flag mandatory = true untuk memaksa broker mengembalikan unroutable messages melalui frame basic.return alih-alih membuangnya secara senyap.
Publisher Confirms Asinkron — Aktifkan confirm mode pada channel untuk menerima konfirmasi ACK/NACK asinkron dari broker setelah data berhasil ditulis ke disk atau konsensus Quorum tercapai.
Backpressure Flow Control — Jika broker kehabisan kapasitas memori RAM atau disk, broker akan menghentikan pembacaan socket TCP klien (status blocked) untuk memperlambat produsen.
Gunakan Sequence Number — Pelacakan konfirmasi pada Publisher Confirms dilakukan menggunakan nomor urut sequence DeliveryTag yang menaik secara monoton per channel.

← Sebelumnya: Quorum Berikutnya: Routing & Binding →

Publishing #

Anatomi Frame AMQP dan Ingress Pipeline #

Jenis-Jenis Frame AMQP #

Negosiasi Ukuran Frame (frame_max) #

Mekanisme Mandatory Flag & Return #

1. Perilaku mandatory = false (Default) #

2. Perilaku mandatory = true #

Publisher Confirms & Flow Control (Backpressure) #

Cara Kerja Publisher Confirms #

Backpressure dan Flow Control #

Implementasi Kode Go: Publisher Confirms & Return Handling #

Metrik Monitoring & TCP Driver Pipeline #

Alur Socket Driver Erlang #

Metrik Kunci untuk Fase Publishing #

Anti-Pattern vs Solusi Kelas Produksi #

1. Menyisipkan mandatory = false untuk Pesan Transaksional Kritis #

Mengapa ini salah? #

2. Mengirim Pesan Secara Sinkron Satu per Satu (Synchronous Blocking Publish) #

Mengapa ini salah? #

Ringkasan #

Negosiasi Ukuran Frame (`frame_max`) #

1. Perilaku `mandatory = false` (Default) #

2. Perilaku `mandatory = true` #

1. Menyisipkan `mandatory = false` untuk Pesan Transaksional Kritis #