Belajar RabbitMQ - Producers & Consumers - Hello World

#Pendahuluan

Inilah momen paling membahagiakan dalam series ini: kalian akan menulis program pertama yang benar-benar berkomunikasi lewat RabbitMQ. Kita mulai dari pola paling sederhana — satu producer mengirim pesan "Hello World" ke queue, dan satu consumer menampilkannya.

Meskipun sederhana, episode ini memuat fondasi penting yang tidak akan berubah sepanjang karier kalian: cara membuka koneksi, cara membuat channel, cara mempublikasikan pesan, dan cara memproses pesan masuk. Kesalahan pengelolaan connection dan channel di tahap ini akan menjadi masalah serius di produksi.

Kami menunjukkan implementasi di tiga bahasa yang paling populer — Python dengan pika, Node.js dengan amqplib, dan Go dengan amqp091-go. Pilih satu bahasa yang paling nyaman, tapi coba pahami pola umumnya: mereka semua berbagi alur yang identik.

#Alur Pesan Sederhana

#Producer Pertama

Install library client di bahasa pilihan kalian. Untuk Python:

pip install pika

Lalu tulis producer. Producer membuka koneksi, membuat channel, mendeklarasikan queue, dan mem-publish pesan:

Producer Hello World (pika)

import pika
 
connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = connection.channel()
 
channel.queue_declare(queue="hello")
 
channel.basic_publish(exchange="", routing_key="hello", body=b"Hello World")
print("pesan terkirim")
connection.close()

Perhatikan bahwa routing_key di sini berperan sebagai nama queue, karena kita memakai default exchange (exchange kosong). Detail exchange akan dibahas di episode 6.

#Consumer Pertama

Consumer menggunakan callback yang dipanggil setiap kali pesan tiba:

Consumer Hello World (pika)

import pika
 
connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = connection.channel()
 
channel.queue_declare(queue="hello")
 
def callback(ch, method, properties, body):
    print(f"menerima: {body}")
 
channel.basic_consume(queue="hello", on_message_callback=callback, auto_ack=True)
print("menunggu pesan...")
channel.start_consuming()

Fungsi channel.start_consuming memblokir program dan terus mendengarkan pesan. auto_ack=True memberitahu broker bahwa pesan langsung dianggap terproses tanpa ack manual.

#Contoh di Bahasa Lain

#Producer dan Consumer di Node.js

Untuk Node.js, install amqplib lalu tulis producer:

npm install amqplib

Producer Hello World (amqplib)

const amqp = require("amqplib");
 
async function main() {
  const conn = await amqp.connect("amqp://localhost");
  const ch = await conn.createChannel();
  await ch.assertQueue("hello");
  ch.sendToQueue("hello", Buffer.from("Hello World"));
  console.log("pesan terkirim");
  await conn.close();
}
 
main();

amqp.connect mengembalikan promise, dan assertQueue mendeklarasikan queue jika belum ada. Di Node.js, pola async di atas akan kita perbaiki dengan connection recovery di episode 28.

#Producer di Go

Untuk Go, install package resmi amqp091-go:

go get github.com/rabbitmq/amqp091-go

package main
 
import (
	"log"
 
	amqp "github.com/rabbitmq/amqp091-go"
)
 
func main() {
	conn, err := amqp.Dial("amqp://localhost")
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
 
	ch, err := conn.Channel()
	if err != nil {
		log.Fatal(err)
	}
	defer ch.Close()
 
	err = ch.Publish("", "hello", false, false, amqp.Publishing{Body: []byte("Hello World")})
	if err != nil {
		log.Fatal(err)
	}
	log.Println("pesan terkirim")
}

Perhatikan bahwa Publish di Go tidak mendeklarasikan queue otomatis — deklarasi harus dilakukan eksplisit. Ini pengingat bahwa deklarasi queue adalah tanggung jawab aplikasi, bukan broker.

#Best Practices Pengelolaan Connection

#Connection Pooling dan Channel per Thread

Satu hal yang wajib diingat: connection adalah sumber daya mahal, channel adalah sumber daya murah. Sebuah aplikasi sebaiknya memakai satu atau beberapa koneksi yang dibagi, dan membuat channel baru untuk setiap task bersamaan. Di Python, jika kalian memakai BlockingConnection di banyak thread, buat satu connection per thread. Di Node.js dan Go, satu connection bisa dipakai concurrency karena channel-nya thread-safe.

#Heartbeat dan Graceful Shutdown

Setiap koneksi harus mengatur heartbeat — ping berkala agar broker tahu client masih hidup. Default heartbeat RabbitMQ adalah 60 detik. Jika client terlalu lama tidak mengirim data, broker akan memutus koneksi yang dianggap mati.

Untuk graceful shutdown, consumer harus berhenti memproses pesan baru, lalu menutup channel dan connection dalam urutan yang benar:

Shutdown yang benar (pika)

channel.stop_consuming()
channel.close()
connection.close()

Perintah connection.close() harus dipanggil setelah semua channel ditutup, agar broker melepas resource dengan bersih.

#Penutup

Di episode 4 ini kalian sudah menulis producer dan consumer pertama di Python, Node.js, dan Go; memahami alur connection-channel-publish-consume; serta belajar praktik terbaik pengelolaan koneksi, heartbeat, dan shutdown yang bersih.

Inti yang harus dibawa pulang:

• Producer mem-publish ke exchange; routing key menunjuk ke queue pada default exchange.
• Consumer bekerja dengan callback yang dipanggil saat pesan tiba.
• Satu connection dipakai untuk banyak channel; jangan boros koneksi.
• Heartbeat menjaga koneksi tetap hidup dan terdeteksi saat mati.
• Graceful shutdown menutup consumer, channel, lalu connection.
• auto_ack=True mempermudah demo, tapi berbahaya untuk produksi.

Di episode 5 selanjutnya kita akan membangun work queues — mendistribusikan task berat ke banyak worker dengan round-robin, memakai manual acknowledgment agar pesan tidak hilang saat worker crash, dan mengatur prefetch dengan QoS agar setiap worker bekerja sesuai kapasitasnya. Konsep ini adalah fondasi dari semua sistem task processing di dunia nyata!