Message Size #

Saat merancang aplikasi berbasis arsitektur microservices yang saling berkomunikasi menggunakan RabbitMQ, salah satu aspek penting yang sering diabaikan adalah ukuran dari muatan data (message payload size) yang dikirimkan. Di atas kertas, RabbitMQ memang memiliki fleksibilitas luar biasa untuk menerima pesan dengan ukuran yang bervariasi. Batas default yang ditentukan oleh broker bahkan dapat mencapai 128 MB per pesan. Namun, kemampuan teknis ini sering disalahartikan sebagai restu desain. Banyak tim developer yang tergoda untuk menggunakan RabbitMQ sebagai media transfer file besar, seperti melampirkan berkas PDF dokumen transaksi, gambar profil pengguna, data log biner, atau ekspor laporan CSV berukuran puluhan megabyte langsung ke dalam payload antrean.

Dalam dunia sistem pengiriman pesan (messaging systems), RabbitMQ diposisikan secara eksklusif sebagai transport layer atau pipa koordinasi transient yang cepat, bukan sebagai media penyimpanan data blob. Mengirimkan pesan dengan ukuran yang terlalu besar adalah salah satu antipattern paling fatal yang dapat mengganggu stabilitas operasional klaster. Setiap byte tambahan pada payload pesan membawa konsekuensi performa tersembunyi yang berlipat ganda di sisi broker, mulai dari konsumsi memori RAM Erlang VM yang membengkak, penurunan throughput jaringan klaster secara drastis, hingga peningkatan risiko terkuncinya koneksi produsen akibat tercapainya alarm batas memori (memory watermark alarm).

Mengapa Ukuran Pesan Sangat Berdampak bagi Kinerja Broker? #

Untuk memahami mengapa broker RabbitMQ sangat sensitif terhadap ukuran pesan, kita harus menganalisis bagaimana runtime Erlang BEAM Virtual Machine mengelola alokasi memori untuk setiap pesan yang masuk:

1. Mekanisme Alokasi Memori Erlang (Heap vs Refc Binaries) #

Di dalam Erlang BEAM VM, pesan-pesan yang berukuran sangat kecil (<64 byte) disimpan langsung di dalam heap memori dari proses Erlang yang mewakili antrean tersebut. Namun, untuk pesan yang berukuran lebih besar, Erlang menempatkannya di area memori eksternal yang disebut Refc Binaries (Reference-Counted Binaries) yang bersifat off-heap. Proses Erlang antrean hanya memegang pointer kecil berukuran beberapa byte yang merujuk ke memori fisik asli.

Meskipun mekanisme refc binaries ini sangat efisien karena menghindari penyalinan payload besar antar-proses Erlang, masalah akan muncul ketika pesan-pesan tersebut menumpuk di antrean (backlog). Broker harus terus memelihara referensi biner tersebut di RAM. Ketika proses Garbage Collection (GC) berjalan pada proses Erlang antrean, pelacakan referensi biner yang terlalu banyak dan berukuran besar akan memicu overhead CPU yang intensif dan memperlambat kinerja penjadwalan proses BEAM VM (reduction scheduling latency).

2. Paging Disk yang Mematikan Performa (Disk Paging Swap) #

RabbitMQ menggunakan sistem manajemen memori dinamis. Jika total penggunaan RAM broker melampaui ambang batas default (biasanya 40% dari total RAM fisik server, yang dikenal sebagai vm_memory_high_watermark), RabbitMQ akan panik dan mengaktifkan mekanisme Disk Paging.

Dalam kondisi ini, broker akan memindahkan pesan-pesan aktif dari RAM secara paksa ke dalam media penyimpanan disk untuk membebaskan ruang memori. Jika antrean dipenuhi oleh jutaan pesan kecil, proses paging ini berjalan relatif stabil. Namun, jika antrean menampung pesan-pesan berukuran besar (misalnya 10 MB per pesan), proses penulisan disk I/O akan langsung mengalami kemacetan (disk I/O saturation). Selama proses paging berlangsung, broker akan menolak menerima pesan baru dari produsen dengan cara memblokir koneksi TCP (connection blocking), yang berakibat pada kelumpuhan sistem produsen.

3. Biaya Replikasi Klaster (Quorum Queue Replication Overhead) #

Jika kita menggunakan Quorum Queue, setiap pesan yang dikirim oleh produsen wajib direplikasi melewati jaringan lokal klaster ke node-node Follower sebelum dikonfirmasi. Jika kita mengirim payload berukuran 5 MB pada klaster 3 node, maka:

• Produsen mengirim 5 MB ke Leader.
• Leader mengirim 5 MB ke Follower A.
• Leader mengirim 5 MB ke Follower B.
• Leader menulis 5 MB ke disk WAL lokalnya.
• Follower A dan B menulis 5 MB ke disk WAL mereka masing-masing.

Total lalu lintas jaringan internal yang tercipta untuk satu pesan tersebut adalah 15 MB, dan total penulisan disk klaster adalah 15 MB. Jika throughput sistem kita adalah 100 pesan per detik, maka beban bandwidth jaringan internal klaster akan langsung melonjak ke 1.5 GB/s, sebuah angka yang dengan mudah akan melumpuhkan kartu jaringan gigabit standar dan memicu keterlambatan konsensus Raft (Raft election timeout).

Rekomendasi Ukuran Payload di Lingkungan Produksi #

Berdasarkan pengujian performa skala besar di industri, berikut adalah acuan ukuran pesan yang sehat untuk menjaga stabilitas klaster RabbitMQ:

• Ukuran Optimal (Sangat Direkomendasikan): Di bawah 100 KB (ideal di kisaran 1 KB hingga 10 KB). Pada rentang ukuran ini, RabbitMQ dapat memproses pesan sepenuhnya di memori RAM dengan latensi sub-milidetik dan throughput hingga puluhan ribu pesan per detik.
• Ukuran Maksimum Aman: 1 MB. Pesan di atas 1 MB harus dianggap sebagai pengecualian desain dan memerlukan pengawasan kapasitas infrastruktur yang ketat.
• Zona Bahaya: Di atas 5 MB. Pengiriman pesan di atas ukuran ini secara konstan hampir pasti akan memicu disk paging, lonjakan confirm latency, dan ketidakstabilan klaster saat terjadi lonjakan beban kerja (spikes).

Solusi Desain: Claim Check Pattern (Pola Pointer Pesan) #

Untuk menangani skenario bisnis di mana kita harus memproses data berukuran besar (seperti pemrosesan laporan bulanan CSV, pemrosesan citra medis, atau tanda tangan digital berkas PDF), kita dilarang keras melampirkan data fisik tersebut ke dalam pesan. Solusi arsitektur terbaik untuk masalah ini adalah dengan menerapkan pola Claim Check Pattern (juga dikenal sebagai Reference Message Pattern).

Konsep kerja Claim Check Pattern sangat sederhana namun elegan:

1. Pemisahan Data: Alih-alih mengirimkan data biner besar ke RabbitMQ, aplikasi produsen mengunggah data besar tersebut terlebih dahulu ke media penyimpanan eksternal yang murah dan efisien untuk data blob (seperti AWS S3, Google Cloud Storage, MinIO, atau database cold storage).
2. Kirim Pointer: Setelah unggahan sukses, produsen mendapatkan referensi lokasi berupa URI unik atau ID data (pointer). Produsen kemudian membuat pesan JSON ringan yang hanya berisi pointer tersebut beserta metadata minimal (misalnya ID pengguna atau jenis operasi) dan mempublikasikannya ke RabbitMQ.
3. Klaim Data: Aplikasi konsumen menerima pesan pointer ringan dari antrean RabbitMQ, menggunakan URI atau ID di dalam payload pesan untuk mengunduh data besar asli dari Object Storage, mengeksekusi logika bisnis, dan setelah selesai, mengirimkan ACK ke broker.

flowchart TD
    Producer["Aplikasi Produsen"] -->|"1. Unggah Payload Besar"| ObjectStorage[("S3 / MinIO Object Storage")]
    ObjectStorage -->>|"2. Kembalikan URI Pointer (s3://bucket/file.pdf)"| Producer
    
    Producer -->|"3. Kirim Pesan Pointer Ringan"| RabbitMQ(("RabbitMQ Broker"))
    RabbitMQ -->|"4. Salurkan Pesan Pointer"| Consumer["Aplikasi Konsumen"]
    
    Consumer -->|"5. Unduh Payload Besar via Pointer"| ObjectStorage
    ObjectStorage -->>|"6. Kembalikan File Asli"| Consumer
    
    Consumer -->|"7. Eksekusi Bisnis & Kirim ACK"| RabbitMQ

    style ObjectStorage stroke:#0288d1,stroke-width:2px
    style RabbitMQ stroke:#7b1fa2,stroke-width:2px

Dengan pola ini, RabbitMQ hanya memproses pesan JSON kecil berukuran beberapa ratus byte. Broker tetap berjalan super cepat, memori RAM stabil, replikasi klaster berjalan instan, dan Object Storage menangani beban transfer file besar dengan kapasitas bandwidth-nya yang memang dirancang untuk skenario tersebut.

Implementasi Claim Check Pattern pada Go dan S3 #

Berikut adalah implementasi lengkap Claim Check Pattern menggunakan bahasa pemrograman Go. Skenarionya adalah memproses laporan PDF besar. Kita menggunakan pustaka AWS SDK untuk menyimulasikan unggahan ke S3 Object Storage, dan pustaka AMQP Go untuk mengalirkan pesan pointer ringan:

1. Implementasi Produsen (Publisher) #

Produsen bertugas mengunggah data besar ke penyimpanan, merancang payload pointer, dan mempublikasikannya:

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"math/rand"
	"time"

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

const (
	amqpURI      = "amqp://admin:securepassword@rabbitmq-cluster:5672/"
	exchangeName = "exchange.report.direct"
	routingKey   = "report.generate"
)

// ReportGeneratedEvent mewakili pesan pointer ringan (Claim Check)
type ReportGeneratedEvent struct {
	EventID     string    `json:"event_id"`
	GeneratedAt time.Time `json:"generated_at"`
	ReportID    string    `json:"report_id"`
	StorageURI  string    `json:"storage_uri"` // Pointer merujuk ke Object Storage
	FileSize    int64     `json:"file_size"`
}

// SimulateS3Upload menyimulasikan proses unggah file laporan PDF 15MB ke AWS S3
func SimulateS3Upload(reportID string) (string, error) {
	log.Printf("[S3] Mengunggah berkas PDF laporan bulanan (15 MB) untuk ID: %s...\n", reportID)
	// Menyimulasikan latensi jaringan I/O unggah file ke cloud storage
	time.Sleep(250 * time.Millisecond)
	
	s3URI := fmt.Sprintf("s3://produksi-laporan-bucket/bulanan/pdf/%s.pdf", reportID)
	log.Printf("[S3] Unggahan sukses. Lokasi penyimpanan: %s\n", s3URI)
	return s3URI, nil
}

func main() {
	// Koneksi ke broker RabbitMQ
	conn, err := amqp.Dial(amqpURI)
	if err != nil {
		log.Fatalf("Gagal terhubung ke RabbitMQ: %v\n", err)
	}
	defer conn.Close()

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

	// Pastikan mode publisher confirms aktif untuk keandalan
	err = ch.Confirm(false)
	if err != nil {
		log.Fatalf("Gagal mengaktifkan publisher confirms: %v\n", err)
	}

	confirms := ch.NotifyPublish(make(chan amqp.Confirmation, 1))

	// 1. Simulasikan pembuatan data besar dan unggah ke S3
	reportID := fmt.Sprintf("REP-%d", rand.Intn(1000000))
	storageURI, err := SimulateS3Upload(reportID)
	if err != nil {
		log.Fatalf("Gagal mengunggah file ke S3: %v\n", err)
	}

	// 2. Buat payload event ringan yang hanya memuat pointer lokasi
	event := ReportGeneratedEvent{
		EventID:     fmt.Sprintf("EVT-%d", rand.Intn(1000000)),
		GeneratedAt: time.Now().UTC(),
		ReportID:    reportID,
		StorageURI:  storageURI, // Mengirim pointer, bukan byte data PDF asli
		FileSize:    15728640,  // 15 MB
	}

	body, err := json.Marshal(event)
	if err != nil {
		log.Fatalf("Gagal melakukan serialisasi JSON: %v\n", err)
	}

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

	// 3. Publikasikan pesan pointer ringan ke RabbitMQ
	log.Printf("[PRODUSEN] Mempublikasikan pesan pointer ringan (Ukuran: %d bytes)...\n", len(body))
	err = ch.PublishWithContext(ctx,
		exchangeName,
		routingKey,
		true, // mandatory
		false,
		amqp.Publishing{
			ContentType:  "application/json",
			DeliveryMode: amqp.Persistent, // Pastikan pesan ditulis aman ke disk
			Body:         body,
			CorrelationId: reportID,
		},
	)
	if err != nil {
		log.Fatalf("Gagal mempublikasikan pesan: %v\n", err)
	}

	// Tunggu konfirmasi penerimaan dari broker
	ack := <-confirms
	if ack.Ack {
		log.Println("[PRODUSEN] Berhasil menerima konfirmasi ACK dari broker.")
	} else {
		log.Println("[PRODUSEN] Menerima sinyal NACK. Pesan gagal masuk antrean.")
	}
}

2. Implementasi Konsumen (Consumer) #

Konsumen menerima pesan pointer ringan, mengunduh file besar asli dari S3, dan memprosesnya:

package main

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

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

const (
	amqpURI   = "amqp://admin:securepassword@rabbitmq-cluster:5672/"
	queueName = "queue.report.process"
)

// ReportGeneratedEvent mewakili struktur pesan pointer yang diterima
type ReportGeneratedEvent struct {
	EventID     string    `json:"event_id"`
	GeneratedAt time.Time `json:"generated_at"`
	ReportID    string    `json:"report_id"`
	StorageURI  string    `json:"storage_uri"`
	FileSize    int64     `json:"file_size"`
}

// DownloadFromS3 menyimulasikan unduhan berkas PDF laporan 15MB berdasarkan pointer URI
func DownloadFromS3(storageURI string) ([]byte, error) {
	log.Printf("[S3] Mendownload berkas asli dari lokasi: %s...\n", storageURI)
	// Menyimulasikan I/O latency download file besar lewat jaringan internet
	time.Sleep(300 * time.Millisecond)
	
	log.Println("[S3] Download selesai. Berkas siap diproses.")
	return []byte("%PDF-1.4 ... isi dokumen laporan palsu ..."), nil
}

func main() {
	conn, err := amqp.Dial(amqpURI)
	if err != nil {
		log.Fatalf("Koneksi gagal: %v\n", err)
	}
	defer conn.Close()

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

	// Batasi prefetch limit agar worker tidak terbebani konsumsi paralel berlebih
	err = ch.Qos(5, 0, false)
	if err != nil {
		log.Fatalf("Gagal menetapkan QoS: %v\n", err)
	}

	deliveries, err := ch.Consume(
		queueName,
		"report-processor-worker",
		false, // manual ACK wajib
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		log.Fatalf("Gagal mendaftarkan consumer: %v\n", err)
	}

	log.Println("Worker laporan aktif. Menunggu pesan masuk...")

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

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

	go func() {
		for msg := range deliveries {
			var event ReportGeneratedEvent
			err := json.Unmarshal(msg.Body, &event)
			if err != nil {
				log.Printf("Gagal parsing JSON payload: %v. Message rejected.\n", err)
				msg.Nack(false, false) // Tolak tanpa requeue agar masuk DLQ
				continue
			}

			log.Printf("[KONSUMEN] Menerima event pointer '%s' untuk Report ID '%s'\n", event.EventID, event.ReportID)

			// 1. Klaim data besar asli dari S3 menggunakan pointer URI
			fileBytes, err := DownloadFromS3(event.StorageURI)
			if err != nil {
				log.Printf("[GAGAL] Gagal mendownload data dari S3: %v. Requeuing...\n", err)
				msg.Nack(false, true) // Requeue pesan pointer agar dicoba lagi
				continue
			}

			// 2. Eksekusi logika bisnis (misalnya proses konversi, rendering, atau OCR PDF)
			processReport(event.ReportID, fileBytes)

			// 3. Kirim ACK sukses ke broker
			msg.Ack(false)
			log.Printf("[SUKSES] Pesan dengan tag %d berhasil diselesaikan.\n", msg.DeliveryTag)
		}
	}()

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

func processReport(reportID string, data []byte) {
	log.Printf("[PROSES] Memproses analisis dokumen '%s' (Ukuran: %d bytes)...\n", reportID, len(data))
	// Simulasi CPU processing
	time.Sleep(100 * time.Millisecond)
}

Perbandingan: Payload Besar vs Payload Ringan (Claim Check) #

Berikut adalah matriks perbandingan mendalam untuk memahami perbedaan dampak operasional pada broker saat kita mempublikasikan pesan berukuran besar secara langsung vs menggunakan pola pesan pointer:

Ringkasan #

• Peran Transport Layer — Perlakukan RabbitMQ murni sebagai pipa perantara asinkron yang cepat dan ringan. Hindari menyimpan atau mentransfer data biner besar langsung lewat broker.
• Batasi Ukuran Pesan — Upayakan agar ukuran payload pesan di produksi tetap berada di bawah 100 KB, dan tetapkan batas maksimal toleransi sebesar 1 MB.
• Terapkan Claim Check Pattern — Unggah data besar (PDF, gambar, berkas CSV) ke Object Storage terlebih dahulu, lalu alirkan pesan asinkron yang hanya berisi pointer URI lokasi penyimpanan tersebut.
• Optimalkan Storage Bandwidth — Biarkan infrastruktur Object Storage (seperti AWS S3 atau Google Cloud Storage) memikul beban lalu lintas transfer file besar, menjaga RabbitMQ tetap bersih dan stabil.

← Sebelumnya: Use Quorum   Berikutnya: Monitoring →