# Arsitektur Event-Driven Berbasis Kafka dan Go: Menangani Juta Pesan Tanpa Block > Diterbitkan oleh Literally Web ID pada 21 Oct 2026 > Sumber: https://literally.web.id/articles/arsitektur-event-driven-berbasis-kafka-dan-go-menangani-juta-pesan-tanpa-block --- Dalam 15 menit ke depan, kita akan merakit sebuah sistem pemrosesan pesan asinkron berkecepatan tinggi menggunakan Apache Kafka dan producer/consumer berbasis Go dari nol sampai berjalan mulus. Bayangkan Anda sedang menghadapi lonjakan jutaan trafik di platform e-commerce saat flash sale; jika arsitektur backend Anda masih menggunakan HTTP synchronous standar atau antrean database biasa, dipastikan database akan *down* karena kehabisan koneksi. Di sinilah kekuatan Event-Driven Architecture (EDA) dengan Kafka menyelamatkan karir Anda, dan Go memberikan performa eksekusi tanpa *block*. ## 1. Menu Kita Hari Ini & Alat yang Perlu Disiapkan Pernahkah Anda melihat sistem *checkout* yang sukses mencatat transaksi secara instan padahal email konfirmasi dan poin *loyalty* baru dikirim beberapa menit kemudian? Itu adalah keajaiban sistem asinkron. Hari ini kita tidak akan menghabiskan waktu berteori tentang definisi pesan atau broker, melainkan langsung membangun pipa penyalur data industri kelas berat di mesin lokal Anda. Sebelum kita mulai meracik kodenya, pastikan beberapa bahan dapur utama sudah terpasang di terminal Anda: - **Go (Golang)** versi 1.21 atau yang lebih baru (karena kita akan memanfaatkan manajemen *concurrency* bawaannya yang elegan). - **Docker & Docker Compose** untuk menjalankan kluster Apache Kafka dan Zookeeper tanpa ribet. - **Kafka CLI tools** atau *extension* VS Code jika ingin mengintip isi topik secara visual. > [!TIP] > Pastikan port `9092` untuk Kafka dan `2181` untuk Zookeeper tidak sedang digunakan oleh aplikasi lokal lain seperti database instans lama Anda. Mari kita siapkan fondasi infrastruktur ini agar pipa data kita berdiri kokoh sebelum kita mulai menulis baris kode Go pertama. ## 2. Peta Jalan Singkat: Apa yang Sedang Kita Rakit Untuk memahami bagaimana jutaan pesan bisa mengalir tanpa ada yang *block*, kita perlu melihat peta jalurnya secara utuh. Arsitektur kita hari ini dibagi menjadi tiga lapisan utama yang saling bahu-membahu: 1. **Producer (Go):** Menerima *payload* request dari klien web, lalu secara asinkron mengirimkannya ke Kafka tanpa menunggu respons *acknowledgment* yang memblokir *thread* utama. 2. **Broker (Apache Kafka):** Mengatur partisioning pesan berdasarkan kunci tertentu (misal: `user_id`) agar urutan data tetap terjaga rapi di dalam *partition*, sekaligus mendistribusikannya ke *Consumer Group*. 3. **Consumer (Go):** Mengambil pesan secara paralel, melakukan proses bisnis berat, dan jika terjadi kegagalan sistemik, pesan tersebut akan dilempar secara otomatis ke *Dead Letter Queue* (DLQ) alih-alih membuat aplikasi *crash* berulang kali. Dari peta jalan ini, mari kita langsung eksekusi langkah pertama dengan merakit file konfigurasi infrastruktur. ## 3. Langkah 1: Mempersiapkan Fondasi & Environment Kita mulai dengan menyiapkan wadah penampung pesan kita. Buat sebuah direktori kerja baru di komputer Anda, lalu buat file bernama `docker-compose.yml` untuk menghidupkan Zookeeper dan Kafka. ```yaml version: '3.8' services: zookeeper: image: confluentinc/cp-zookeeper:7.5.0 environment: ZOOKEEPER_CLIENT_PORT: 2181 ZOOKEEPER_TICK_TIME: 2000 ports: - "2181:2181" kafka: image: confluentinc/cp-kafka:7.5.0 depends_on: - zookeeper ports: - "9092:9092" environment: KAFKA_BROKER_ID: 1 KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181' KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092 KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 ``` Jalankan perintah `docker-compose up -d` di terminal Anda. Lihat responsnya—jika kontainer menyala tanpa error, itu tanda bahwa fondasi *broker* pesan kita sudah siap menerima aliran data. Sekarang, saatnya kita masuk ke inti pemrograman Go untuk berinteraksi langsung dengan kluster Kafka ini. ## 4. Langkah 2: Meracik Konfigurasi Inti Tanpa Basa-basi Inisialisasi modul Go baru dengan perintah `go mod init kafka-event-driven`. Kita akan menggunakan pustaka `github.com/IBM/sarama` (penerus pustaka Sarama dari Shopify) yang terkenal tangguh di lingkungan produksi berskala masif. Jalankan `go get github.com/IBM/sarama`. Setelah itu, mari kita buat file `producer.go` untuk merancang *asynchronous producer* yang tidak akan membuat HTTP server Anda *lag* saat terjadi lonjakan trafik. ```go package main import ( "encoding/json" "log" "os" "os/signal" "time" "github.com/IBM/sarama" ) type OrderEvent struct { OrderID string `json:"order_id"` UserID string `json:"user_id"` Amount float64 `json:"amount"` Timestamp time.Time `json:"timestamp"` } func main() { config := sarama.NewConfig() config.Producer.Return.Successes = true config.Producer.RequiredAcks = sarama.WaitForLocal config.Producer.Retry.Max = 5 producer, err := sarama.NewAsyncProducer([]string{"localhost:9092"}, config) if err != nil { log.Fatalf("Gagal membuat async producer: %v", err) } defer producer.Close() // Menangani error dan success secara asynchronous agar goroutine utama tidak block go func() { for { select { case err := <-producer.Errors(): log.Printf("Gagal mengirim pesan: %v\n", err.Err) case succ := <-producer.Successes(): log.Printf("Pesan sukses terkirim ke topik %s [partisi %d] offset %d\n", succ.Topic, succ.Partition, succ.Offset) } } }() signals := make(chan os.Signal, 1) signal.Notify(signals, os.Interrupt) // Simulasi pengiriman 5 juta pesan secara cepat tanpa block for i := 1; i <= 5; i++ { order := OrderEvent{ OrderID: "ORD-" + time.Now().Format("20060102150405") + "-" + string(rune(i)), UserID: "USR-99", Amount: 150.50, Timestamp: time.Now(), } payload, _ := json.Marshal(order) msg := &sarama.ProducerMessage{ Topic: "orders.created", Key: sarama.StringEncoder(order.UserID), Value: sarama.ByteEncoder(payload), } // Kirim ke kanal asinkron producer.Input() <- msg } time.Sleep(2 * time.Second) // Menunggu goroutine callback menyelesaikan tugasnya } ``` Perhatikan bagaimana kita memanfaatkan *channel* bawaan Go dan `producer.Input()` untuk mengirim data. Pemanggilan ini tidak menunggu jaringan merespons secara sinkron, sehingga latensi aplikasi web Anda tetap berada di level mikrodetik. ## 5. Langkah 3: Sentuhan Akhir Keamanan & Optimasi Sebuah sistem *streaming* data tidak akan lengkap tanpa penanganan konsumen yang cerdas dan mekanisme pengamanan data gagal (*Dead Letter Queue* / DLQ). Buat file `consumer.go` untuk membaca pesan tersebut, memprosesnya melalui *Consumer Group*, dan mengalihkan pesan yang rusak ke DLQ. ```go package main import ( "context" "encoding/json" "log" "os" "os/signal" "sync" "syscall" "github.com/IBM/sarama" ) type ConsumerGroupHandler struct{} func (ConsumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil } func (ConsumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil } func (h ConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error { for message := range claim.Messages() { var data map[string]interface{} if err := json.Unmarshal(message.Value, &data); err != nil { log.Printf("[DLQ] Format pesan tidak valid, melempar ke Dead Letter Queue: %v", err) // Di sini Anda bisa menambahkan logika meneruskan ke topik DLQ session.MarkMessage(message, "") continue } // Proses bisnis utama berhasil log.Printf("Berhasil memproses Order ID: %v dari partisi %d", data["order_id"], message.Partition) session.MarkMessage(message, "") } return nil } func main() { config := sarama.NewConfig() config.Consumer.Group.Rebalance.Strategy = sarama.NewBalanceStrategyRoundRobin() config.Consumer.Offsets.Initial = sarama.OffsetOldest client, err := sarama.NewConsumerGroup([]string{"localhost:9092"}, "order-processing-group", config) if err != nil { log.Fatalf("Gagal menginisialisasi consumer group: %v", err) } defer client.Close() ctx, cancel := context.WithCancel(context.Background()) defer cancel() var wg sync.WaitGroup ım, handler := client, ConsumerGroupHandler{} wg.Add(1) go func() { defer wg.Done() for { if err := im.Consume(ctx, []string{"orders.created"}, handler); err != nil { log.Printf("Error pada consumer: %v", err) } if ctx.Err() != nil { return } } }() sigterm := make(chan os.Signal, 1) signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM) <-sigterm cancel() wg.Wait() } ``` Dengan membagi tugas melalui *Consumer Group*, Anda dapat melipatgandakan jumlah instans *consumer* Go di server berbeda tanpa khawatir pesan diproses dua kali. ## 6. Tes Rasa: Memastikan Semuanya Berjalan Sempurna Saatnya menguji racikan kita di dapur lokal. Jalankan *consumer* terlebih dahulu pada terminal pertama: ```bash go run consumer.go ``` Lalu, di terminal kedua, jalankan *producer* kita: ```bash go run producer.go ``` Perhatikan layar *consumer* Anda. Pesan-pesan akan langsung terproses secara instan dengan distribusi partisi yang seimbang tanpa ada jeda atau *blocking thread*. Anda baru saja membangun jembatan data berkecepatan tinggi yang siap menampung jutaan transaksi harian. ## 7. Pertolongan Pertama: Jika Terjadi Kendala Kecil > [!WARNING] > Jika Anda mendapati error `kafka: client has run out of available brokers` saat menjalankan program Go, itu biasanya karena konfigurasi `KAFKA_ADVERTISED_LISTENERS` di Docker Compose masih menggunakan `localhost` atau belum dapat dijangkau oleh *container* lain. Pastikan *host* dan port sudah dipetakan dengan benar ke `localhost:9092`. Jika terjadi kendala *rebalancing lag* pada *consumer group*, pastikan Anda selalu memanggil `session.MarkMessage(message, "")` hanya setelah proses bisnis selesai dengan sukses. Selamat mencoba, dan nikmati performa arsitektur event-driven tanpa hambatan!