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:

# 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:

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:

# 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.

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:

{
  "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:

groups:
  - name: rabbitmq_emergency_alerts
    rules:
      - alert: RabbitMQQueueBacklogHigh
        expr: rabbitmq_queue_messages_ready{queue="payment_queue\