Your First Queue

ตัวอย่างที่เล็กที่สุดที่มีประโยชน์

มาส่ง message หนึ่งตัวผ่าน RabbitMQ กัน เพื่อให้เรียบง่ายที่สุด เราจะใช้ default exchange — direct exchange ที่ไม่มีชื่อ ซึ่ง RabbitMQ ให้มาอัตโนมัติ โดย routing key ก็คือชื่อ queue นั่นเอง ทำให้เรา “publish ตรงเข้า queue” ได้โดยยังไม่ต้องประกาศ exchange (ทุกบทเรียนถัด ๆ ไปใช้ named exchange จริง)

ก่อนอื่น run broker บนเครื่องด้วย Docker:

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management

port 5672 คือ AMQP (สำหรับแอปคุณ); 15672 คือ management UI (เปิด http://localhost:15672 login ด้วย guest/guest)

producer: publish หนึ่ง message

producer connect, เปิด channel, ประกาศ queue (เพื่อให้แน่ใจว่า queue มีอยู่จริง) แล้ว publish message เข้าไป

import amqp from 'amqplib';

const conn = await amqp.connect('amqp://localhost');
const channel = await conn.createChannel();

const queue = 'hello';
await channel.assertQueue(queue, { durable: false });

channel.sendToQueue(queue, Buffer.from('Hello, RabbitMQ!'));
console.log('sent: Hello, RabbitMQ!');

await channel.close();
await conn.close();

import pika

conn = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = conn.channel()

channel.queue_declare(queue="hello", durable=False)

channel.basic_publish(exchange="", routing_key="hello", body="Hello, RabbitMQ!")
print("sent: Hello, RabbitMQ!")

conn.close()

conn, _ := amqp.Dial("amqp://guest:guest@localhost:5672/")
defer conn.Close()
ch, _ := conn.Channel()
defer ch.Close()

q, _ := ch.QueueDeclare("hello", false, false, false, false, nil)

ch.PublishWithContext(ctx, "", q.Name, false, false, amqp.Publishing{
    Body: []byte("Hello, RabbitMQ!"),
})
log.Println("sent: Hello, RabbitMQ!")

สังเกต exchange ที่เป็น string ว่าง ("") — นั่นคือ default exchange และ routing key คือชื่อ queue

consumer: รับ message

consumer connect, ประกาศ queue เดียวกัน (การประกาศเป็น idempotent — ทำทั้งสองฝั่งได้อย่างปลอดภัย) แล้ว subscribe callback ของตัวเองจะทำงานทุกครั้งที่มี message ถูก deliver

import amqp from 'amqplib';

const conn = await amqp.connect('amqp://localhost');
const channel = await conn.createChannel();

const queue = 'hello';
await channel.assertQueue(queue, { durable: false });

console.log('waiting for messages...');
channel.consume(queue, (msg) => {
  if (msg) {
    console.log('received:', msg.content.toString());
    channel.ack(msg);
  }
});

import pika

conn = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = conn.channel()
channel.queue_declare(queue="hello", durable=False)

def on_message(ch, method, properties, body):
    print("received:", body.decode())
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(queue="hello", on_message_callback=on_message)
print("waiting for messages...")
channel.start_consuming()

conn, _ := amqp.Dial("amqp://guest:guest@localhost:5672/")
defer conn.Close()
ch, _ := conn.Channel()
defer ch.Close()

q, _ := ch.QueueDeclare("hello", false, false, false, false, nil)
msgs, _ := ch.Consume(q.Name, "", false, false, false, false, nil)

log.Println("waiting for messages...")
for d := range msgs {
    log.Printf("received: %s", d.Body)
    d.Ack(false)
}

เกิดอะไรขึ้นบ้าง

flowchart LR
  p["Producer
sendToQueue('hello')"] -->|default exchange| q["Queue: hello"]
  q -->|deliver| c["Consumer
print + ack"]

flow ของ hello-world

รายละเอียดสำคัญ แม้จะเล็กน้อย:

ทั้งสองฝั่งประกาศ queue การประกาศเป็น idempotent และมั่นใจว่า queue มีอยู่ไม่ว่าใครจะ start ก่อน
message ก็แค่ bytes RabbitMQ ไม่สนใจ format ของคุณ — string, JSON, protobuf อะไรก็ได้ การ serialize เป็นหน้าที่ของคุณ
consumer ack หลัง process แล้ว เป็นการบอก RabbitMQ ว่า “เสร็จแล้ว” เพื่อให้ message ถูกลบ ถ้าข้ามอันนี้ไป คุณจะได้ message เดิมอีกครั้งตอน reconnect (ที่เป็นความปลอดภัยที่คุณอยากได้พอดี — เพิ่มเติมในบท acknowledgement)