# Membangun Event-Driven Architecture dengan RabbitMQ dan Go > Diterbitkan oleh Literally Web ID pada 21 Oct 2026 > Sumber: https://literally.web.id/articles/membangun-event-driven-architecture-dengan-rabbitmq-dan-go --- ## 1. Alarm Tengah Malam: Saat Layanan Produksi Mulai Tumbang Pukul 02:45 WIB. Ponsel saya bergetar liar di atas nakas, memancarkan notifikasi PagerDuty dengan nada paling memekakkan telinga. *High Error Rate detected on checkout service.* Belum sempat saya mencuci muka, grup Telegram internal tim infrastruktur sudah dibanjiri laporan dari tim *frontend*: pengguna gagal melakukan pembayaran, tombol *checkout* berputar tanpa henti, dan dashboard Grafana menyala merah terang layaknya lampu disko darurat. Sebagai Principal Systems Architect, malam-malam seperti ini selalu menjadi ujian nyata bagi ketahanan mental dan arsitektur yang kami bangun. Sistem kami dirancang untuk memproses jutaan event harian menggunakan arsitektur *event-driven* berbasis RabbitMQ sebagai *message broker* utama yang berinteraksi dengan layanan *backend* berbahasa Go. Namun, malam ini, sistem yang biasanya berlari kencang itu mengalami *choke*. Koneksi TCP ke RabbitMQ menumpuk, *goroutine* di sisi aplikasi Go melonjak hingga ratusan ribu, dan memori Node.js/Go di server mulai menipis karena kehabisan nafas. Ada sesuatu yang salah, dan kami harus segera masuk ke ruang kendali sebelum sistem benar-benar tumbang total. Memasuki menit-menit awal investigasi, kami sadar bahwa masalah ini tidak sekadar lonjakan trafik musiman, melainkan sebuah anomali struktural dalam cara komunikasi asinkron kami dikelola. Mari kita bedah jejak digital yang ditinggalkan oleh bencana tengah malam ini. ## 2. Jejak Digital: Menelusuri Anomali Log & Lonjakan Metrik Begitu mendarat di terminal, saya langsung membuka sesi SSH ke *bastion host* dan melompat ke *node* kluster Go backend. Perintah pertama yang saya eksekusi bukanlah melihat log aplikasi, melainkan memeriksa kesehatan sistem operasi dan koneksi jaringan menggunakan alat investigasi standar lapangan: ```bash # Memeriksa status koneksi TCP yang mengarah ke port RabbitMQ (5672) ss -tan state established '( dport = :5672 or sport = :5672 )' | wc -l # Memantau penggunaan goroutine dan heap memory secara real-time via endpoint pprof go tool pprof -top http://localhost:6060/debug/pprof/heap ``` Hasilnya langsung membuat bulu kuduk berdiri. Jumlah koneksi TCP aktif membengkak hingga menyentuh batas maksimal kernel, sementara dari sisi profil memori, objek bertipe `*amqp.Channel` dan `*amqp.Connection` mendominasi alokasi heap. Di dashboard RabbitMQ Management UI, metrik *Ready Messages* di antrean utama melonjak hingga 4,5 juta pesan, sementara *Unacked Messages* terjebak di angka puluhan ribu. > [!WARNING] > Lonjakan *Unacked Messages* adalah tanda bahaya klasik. Ini berarti *consumer* menerima pesan dari RabbitMQ, tetapi gagal mengirimkan sinyal `ACK` (Acknowledgment) kembali ke broker, entah karena proses *worker* mengalami *deadlock*, panik, atau kewalahan memproses beban I/O ke database MySQL. Penelusuran log Nginx dan PHP-FPM di layanan pendukung juga menunjukkan adanya efek domino (*cascading failure*). Permintaan HTTP dari klien tertahan karena *upstream API* yang digerakkan oleh Go *consumer* tidak kunjung memberikan respons. Kami harus segera mencari tahu akar masalahnya sebelum kluster RabbitMQ kehabisan ruang disk dan mengalami *crash* permanen. ## 3. Membongkar Biang Kerok: Akar Masalah yang Sebenarnya Setelah menggali lebih dalam melalui korelasi log antara *timestamp* kegagalan di Go dan log internal RabbitMQ (`/var/log/rabbitmq/rabbit@hostname.log`), kami menemukan tiga dosa besar arsitektural yang tersembunyi rapi di balik kode produksi: 1. **Ketiadaan Pembatasan Concurrency (Unbounded Concurrency)**: Di sisi Go *consumer*, setiap pesan yang masuk dari RabbitMQ langsung dilempar ke dalam *goroutine* baru tanpa kendali (`go processMessage(msg)`). Ketika trafik melonjak, aplikasi menciptakan jutaan *goroutine* seketika, menyebabkan *context switching overhead* yang ekstrem dan menghabiskan memori. 2. **Kesalahan Desain Exchange dan Routing Key**: Seluruh event dari berbagai domain bisnis (pembayaran, pengiriman, notifikasi) dilempar ke satu *Direct Exchange* dengan *routing key* yang terlalu generik. Akibatnya, ketika layanan notifikasi mengalami kelambatan, ia menyeret antrean layanan pembayaran ikut tertahan (*head-of-line blocking*). 3. **Absennya Dead Letter Queue (DLQ) yang Layak**: Ketika terjadi kesalahan parsing JSON pada payload pesan, *consumer* secara naif melakukan *requeue* (`msg.Ack(false)` atau mengembalikannya dengan `requeue=true`) tanpa batas. Pesan yang rusak tersebut berputar-putar selamanya (*poison pill*), menyedot sumber daya CPU secara sia-sia. > [!TIP] > Dalam arsitektur *event-driven* skala besar, *goroutine* di Go memang ringan, namun koneksi TCP, deskriptor file, dan antrean *channel* AMQP memiliki batas fisik. Mengontrol konkurensi menggunakan *worker pool* atau *channel semaphore* adalah harga mati. Mari kita lihat perbandingan antara pendekatan naif yang kami gunakan sebelumnya dengan arsitektur defensif yang seharusnya diterapkan sejak hari pertama: | Fitur / Aspek | Pendekatan Naif (Sebelum Insiden) | Pendekatan Defensif (Setelah Perbaikan) | | :--- | :--- | :--- | | **Concurrency Model** | `go processMsg(msg)` tanpa batas | *Worker Pool* tetap dengan *buffered channel* | | **Exchange Strategy** | Satu *Direct Exchange* untuk segalanya | *Topic Exchange* terisolasi per domain bisnis | | **Error Handling** | Requeue tanpa batas (*poison pill*) | Routing otomatis ke Dead Letter Exchange (DLX) | | **QoS / Prefetch** | Default (Tidak terbatas) | Prefetch limit ketat (`ch.Qos(50, 0, false)`) | Dengan akar masalah yang sudah teridentifikasi secara presisi, saatnya kita menyusun strategi penyelamatan darurat untuk menstabilkan infrastruktur malam ini juga. ## 4. Tindakan Darurat: Menghidupkan Kembali Layanan dalam Hitungan Menit Untuk menyelamatkan sistem dari kehancuran total, kami tidak boleh menunggu perubahan kode selesai di-deploy melalui jalur normal. Kami membutuhkan intervensi langsung di tingkat infrastruktur dan broker pesan. Langkah pertama adalah memutus koneksi *consumer* yang menggantung untuk menurunkan beban pada RabbitMQ, lalu membersihkan antrean korup, dan menerapkan pembatasan *prefetch* secara manual via manajemen API. Berikut adalah rangkaian perintah terminal darurat yang kami eksekusi untuk meredakan situasi dalam waktu kurang dari 60 detik: ```bash # 1. Hentikan sementara layanan consumer Go di semua server aplikasi sudo systemctl stop payment-consumer.service # 2. Purge antrean yang mengalami poison pill jika data tidak krusial (atau pindahkan via CLI) # Menggunakan rabbitmqadmin untuk menghapus antrean yang macet rabbitmqadmin -u admin -p secure_pass delete queue name=checkout_deadlock_queue # 3. Sesuaikan ulang batas file descriptor sistem operasi di host RabbitMQ ulimit -n 65535 echo 'fs.file-max = 2097152' | sudo tee -a /etc/sysctl.conf sudo sysctl -p # 4. Nyalakan kembali service dengan konfigurasi environment darurat (concurrency dibatasi) sudo systemctl start payment-consumer.service ``` Setelah eksekusi perintah tersebut, grafik metrik di Grafana mulai menunjukkan tren positif. Jumlah koneksi TCP kembali normal, penggunaan memori stabil, dan antrean pesan mulai terkikis secara teratur. Namun, insiden ini adalah pengingat keras bahwa perbaikan sementara bukanlah solusi permanen. Kita harus merombak fondasi kodenya. ## 5. Kebijakan Retrospektif: Menambal Celah agar Tak Pernah Terulang Keesokan harinya, tim berkumpul untuk melakukan *post-mortem* mendalam. Kami merombak total modul integrasi RabbitMQ di aplikasi Go kami. Berikut adalah implementasi kode Go produksi yang menerapkan *Worker Pool*, *QoS Prefetch*, dan penanganan *Dead Letter Queue* yang kokoh. ```go package main import ( "context" "encoding/json" "log" "sync" "time" amqp "github.com/rabbitmq/amqp091-go" ) type EventPayload struct { ID string `json:"id"` Action string `json:"action"` Time time.Time `json:"time"` } // Consumer mendefinisikan struktur worker pool yang aman type Consumer struct { conn *amqp.Connection channel *amqp.Channel queue string workers int } func NewConsumer(connString, queueName string, workers int) (*Consumer, error) { conn, err := amqp.Dial(connString) if err != nil { return nil, err } ch, err := conn.Channel() if err != nil { conn.Close() return nil, err } // KUNCI UTAMA: Batasi prefetch agar worker tidak rakus mengambil pesan err = ch.Qos( 50, // prefetch count: maksimal 50 pesan belum di-ack per worker 0, // prefetch size false, // global ) if err != nil { return nil, err } return &Consumer{ conn: conn, channel: ch, queue: queueName, workers: workers, }, nil } func (c *Consumer) Start(ctx context.Context) error { msgs, err := c.channel.Consume( c.queue, "go-production-consumer", false, // autoAck = false, kita lakukan manual ACK/NACK false, // exclusive false, // noLocal false, // noWait nil, ) if err != nil { return err } var wg sync.WaitGroup jobChan := make(chan amqp.Delivery, c.workers*2) // Spawn Worker Pool for i := 0; i < c.workers; i++ { wg.Add(1) go func(workerID int) { defer wg.Done() for { select { case <-ctx.Done(): return case msg, ok := <-jobChan: if !ok { return } c.processMessage(workerID, msg) } } }(i) } // Distribusi pesan masuk ke job channel for { select { case <-ctx.Done(): close(jobChan) wg.Wait() return nil case msg, ok := <-msgs: if !ok { close(jobChan) wg.Wait() return nil } jobChan <- msg } } } func (c *Consumer) processMessage(workerID int, msg amqp.Delivery) { var payload EventPayload if err := json.Unmarshal(msg.Body, &payload); err != nil { log.Printf("[Worker %d] ERROR: Gagal unmarshal payload: %v. Mengirim ke DLQ.\n", workerID, err) // Tolak pesan dan jangan requeue, biarkan masuk ke DLX/DLQ msg.Nack(false, false) return } // Simulasi proses bisnis yang aman log.Printf("[Worker %d] Memproses event ID: %s\n", workerID, payload.ID) time.Sleep(50 * time.Millisecond) // Konfirmasi sukses msg.Ack(false) } func main() { ctx, cancel := context.WithCancel(context.Background()) defer cancel() consumer, err := NewConsumer("amqp://guest:guest@localhost:5672/", "payment_queue", 10) if err != nil { log.Fatalf("Gagal menginisialisasi consumer: %v", err) } log.Println("Consumer berjalan dengan Worker Pool. Tekan CTRL+C untuk berhenti.") if err := consumer.Start(ctx); err != nil { log.Fatalf("Consumer berhenti dengan error: %v", err) } } ``` Selain perbaikan kode aplikasi, kami juga menata ulang konfigurasi topologi RabbitMQ menggunakan deklarasi *Topic Exchange* dan *Dead Letter Exchange (DLX)* melalui skrip deklarasi infrastruktur: ```json { "exchanges": [ { "name": "ecommerce.topic", "type": "topic", "durable": true, "auto_delete": false, "internal": false, "arguments": {} }, { "name": "ecommerce.dlx", "type": "direct", "durable": true, "auto_delete": false, "internal": false, "arguments": {} } ], "queues": [ { "name": "payment_queue", "durable": true, "arguments": { "x-dead-letter-exchange": "ecommerce.dlx", "x-dead-letter-routing-key": "payment.dlq" } }, { "name": "payment_dlq", "durable": true, "arguments": {} } ], "bindings": [ { "source": "ecommerce.topic", "vhost": "/", "destination": "payment_queue", "destination_type": "queue", "routing_key": "order.payment.#", "arguments": {} }, { "source": "ecommerce.dlx", "vhost": "/", "destination": "payment_dlq", "destination_type": "queue", "routing_key": "payment.dlq", "arguments": {} } ] } ``` Arsitektur baru ini memastikan bahwa jika ada pesan rusak atau gagal diproses setelah *retry* tertentu, ia akan otomatis terisolasi ke dalam `payment_dlq` tanpa mengganggu aliran lalu lintas utama. ## 6. Pasang Alarm: Otomasi Deteksi Dini untuk Tim On-Call Agar kejadian pukul 02:45 WIB ini tidak pernah terulang dalam bentuk kejutan buta, kami memasang lapisan pemantauan proaktif menggunakan Prometheus dan Alertmanager yang terintegrasi langsung dengan webhook PagerDuty. Berikut adalah aturan *alerting* Prometheus (`prometheus-alerts.yml`) yang memantau penumpukan pesan dan kesehatan *consumer* secara real-time: ```yaml groups: - name: rabbitmq_emergency_alerts rules: - alert: RabbitMQQueueBacklogHigh expr: rabbitmq_queue_messages_ready{queue="payment_queue\