Aditya Pratama

Pendahuluan

Dalam sistem yang membutuhkan penundaan pemrosesan pesan, kita bisa menggunakan Delayed Queue di RabbitMQ. Salah satu cara untuk mengimplementasikannya adalah dengan menggunakan plugin Delayed Message Exchange dari RabbitMQ. Artikel ini akan membahas cara mengatur publisher dan consumer dengan Golang menggunakan library github.com/rabbitmq/amqp091-go.

Persiapan

Sebelum memulai implementasi, pastikan Anda memiliki:

1. Golang terinstal di sistem Anda.
2. Docker untuk menjalankan RabbitMQ dengan plugin delayed exchange.
3. Library amqp091-go, yang dapat diinstal dengan perintah berikut: go get github.com/rabbitmq/amqp091-go

Menjalankan RabbitMQ dengan Plugin Delayed Exchange

Karena RabbitMQ tidak mendukung delayed message secara default, kita perlu menggunakan image Docker khusus:

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 heidiks/rabbitmq-delayed-message-exchange

• Port 5672 → Untuk koneksi AMQP.
• Port 15672 → Untuk RabbitMQ Management UI (http://localhost:15672).
• Username/Password default: guest/guest.

Implementasi Publisher

Publisher bertugas mengirim pesan dengan delay tertentu. Berikut kode untuk mengirim pesan dengan delay 5 detik:

package main

import (
	"context"
	"log"
	"time"

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

const (
	rabbitMQURL  = "amqp://guest:[email protected]:5672/"
	exchangeName = "delayed_exchange"
	queueName    = "delayed_queue"
	routingKey   = "delayed_key"
	delayMs      = 5000 // 5 detik dalam milidetik
	// delayMs      = 5 * 60 * 1000 // 5 menit dalam milidetik
)

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

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

	// Deklarasi Exchange dengan tipe "x-delayed-message"
	args := amqp.Table{
		"x-delayed-type": "direct",
	}
	err = ch.ExchangeDeclare(
		exchangeName, "x-delayed-message", true, false, false, false, args,
	)
	if err != nil {
		log.Fatalf("Gagal mendeklarasikan exchange: %v", err)
	}

	// Mempublish pesan dengan delay
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	body := "Pesan dengan delay 5 menit"
	err = ch.PublishWithContext(ctx,
		exchangeName, routingKey, false, false,
		amqp.Publishing{
			Headers:     amqp.Table{"x-delay": delayMs},
			ContentType: "text/plain",
			Body:        []byte(body),
		})
	if err != nil {
		log.Fatalf("Gagal mengirim pesan: %v", err)
	}

	log.Printf("Pesan terkirim: %s", body)
}

Implementasi Consumer

Consumer bertugas menerima pesan yang telah dikirim oleh publisher setelah delay berakhir. Berikut kodenya:

package main

import (
	"log"

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

const (
	rabbitMQURL  = "amqp://guest:[email protected]:5672/"
	exchangeName = "delayed_exchange"
	queueName    = "delayed_queue"
	routingKey   = "delayed_key"
)

func main() {
	log.Println("🔄 Menghubungkan ke RabbitMQ...")
	conn, err := amqp.Dial(rabbitMQURL)
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	log.Println("✅ Berhasil terhubung ke RabbitMQ")

	ch, err := conn.Channel()
	if err != nil {
		log.Fatal(err)
	}
	defer ch.Close()
	log.Println("✅ Channel terbuka")

	// Deklarasi Queue
	q, err := ch.QueueDeclare(
		queueName, // Nama queue harus sama dengan yang di publisher
		true,      // Durable
		false,     // Auto-delete
		false,     // Exclusive
		false,     // No-wait
		nil,
	)
	if err != nil {
		log.Fatal(err)
	}
	log.Println("✅ Queue berhasil dideklarasikan")

	// Bind Queue ke Exchange
	err = ch.QueueBind(
		q.Name,       // Nama queue
		routingKey,   // Routing key harus sama dengan publisher
		exchangeName, // Nama exchange
		false,
		nil,
	)
	if err != nil {
		log.Fatal("Gagal melakukan binding:", err)
	}
	log.Println("✅ Queue berhasil di-bind ke exchange")

	// Konsumsi pesan dari queue
	msgs, err := ch.Consume(
		q.Name, // Nama queue
		"",     // Consumer
		true,   // Auto Ack
		false,  // Exclusive
		false,  // No Local
		false,  // No Wait
		nil,    // Arguments
	)
	if err != nil {
		log.Fatal(err)
	}
	log.Println("✅ Consumer siap menerima pesan")

	// Loop untuk membaca pesan
	for msg := range msgs {
		log.Printf("📩 Pesan diterima: %s", msg.Body)
	}
}

Cara Menjalankan

1. Jalankan Consumer

go run consumer.go

Consumer akan menunggu pesan dari RabbitMQ.

2. Jalankan Publisher

go run publisher.go

Pesan akan dikirim dengan delay 5 detik sebelum diterima oleh consumer.

Kesimpulan

Dengan menggunakan RabbitMQ Delayed Message Exchange, kita dapat menunda pengiriman pesan tanpa perlu mengubah kode aplikasi secara signifikan. Implementasi ini sangat berguna untuk skenario seperti retry mechanism, job scheduling, atau delayed notifications.

Semoga bermanfaat! 🚀