RabbitMQ Exchange Types: Direct, Fanout, Topic, Headers และการส่งข้อความไม่ให้หาย

ใน RabbitMQ producer ไม่ได้ส่งข้อความเข้า queue ตรง ๆ แต่ publish ไปที่ exchange แล้ว exchange เป็นคนตัดสินว่า queue ไหนจะได้รับสำเนาข้อความนั้น RabbitMQ exchange types มีสี่แบบ คือ direct, fanout, topic และ headers แต่ละแบบ route ต่างกัน และการเลือกให้ถูกคืองานออกแบบส่วนใหญ่ของระบบที่ใช้ RabbitMQ

แต่การ route เป็นแค่ครึ่งเดียว อีกครึ่งคือทำให้ข้อความรอดจาก crash, การ restart และ consumer ที่ทำงานช้า บทความนี้พูดถึงทั้งสองเรื่อง พร้อมตัวอย่างโค้ด Go

ทำไมต้องมี message queue

สมมติ service A ต้องส่งงานให้ service B ทำ ถ้า A เรียก API ของ B ตรง ๆ:

• ถ้า B ล่มตอนนั้น งานหายเลย เว้นแต่ A จะ retry เอง
• A ต้องรอ B ทำเสร็จก่อนถึงจะตอบคนที่เรียกตัวเองได้
• ถ้า traffic พุ่งขึ้นมา B รับเต็ม ๆ และอาจล้ม

Message queue เข้ามาเป็นตัวกลาง A โยนข้อความเข้าคิวแล้วไปทำอย่างอื่นต่อ ส่วน B ค่อยมาหยิบไปทำเมื่อพร้อม ถ้า B ล่ม ข้อความก็รออยู่ในคิว ถ้างานเข้ามาพรวดเดียว คิวก็เก็บไว้ให้ ผลที่ได้คือ decoupling (A ไม่ต้องรู้จัก B), async (A ไม่ต้องรอ) และ buffering (ช่วยรับแรงกระแทกของ traffic)

โมเดลของ RabbitMQ: producer, exchange, queue, consumer

RabbitMQ ใช้โมเดลของ AMQP 0-9-1 ซึ่งมีองค์ประกอบหลักหกตัว:

ทำไมต้องมี exchange คั่นหน้า queue? เพราะ producer รู้แค่ชื่อ exchange กับ routing key ก็พอ ส่วน queue ไหนจะได้รับข้อความเป็นเรื่องของ binding ซึ่งเปลี่ยนได้โดยไม่ต้องแตะโค้ด producer เลย ถ้ามี service ใหม่อยากได้ event ของ order ก็แค่ bind queue ของตัวเองเข้าไป ส่วน order service ไม่ต้องแก้อะไร

ถ้าเคยเห็นโค้ดที่ดูเหมือน publish เข้า queue ตรง ๆ นั่นคือการใช้ default exchange ซึ่งเป็น direct exchange ไม่มีชื่อ ที่ทุก queue ถูก bind ไว้อัตโนมัติด้วยชื่อของตัวเอง สะดวกสำหรับ work queue ง่าย ๆ แต่ก็ยังเป็น exchange อยู่ดี

RabbitMQ exchange types ทั้งสี่แบบ

Direct exchange

Direct exchange ส่งข้อความไปทุก queue ที่ binding key ตรงกับ routing key เป๊ะ หลาย queue bind ด้วย key เดียวกันได้ และ queue เดียวมีหลาย binding ได้

ใช้บ่อยกับการ route ตามหมวดที่ตายตัว เช่น ระดับของ log หรือประเภทงาน

// messages with routing key "error" go to the errors queue
must(ch.ExchangeDeclare("logs", "direct", true, false, false, false, nil))
must(ch.QueueBind("log-errors", "error", "logs", false, nil))
must(ch.QueueBind("log-all", "error", "logs", false, nil))
must(ch.QueueBind("log-all", "info", "logs", false, nil))

Fanout exchange

Fanout exchange ไม่สน routing key เลย มันคัดลอกทุกข้อความไปยัง ทุก queue ที่ bind ไว้ คือการ broadcast

ใช้กับงานแบบ "แจ้งทุกคน" เช่น event user.registered ที่ทั้ง service อีเมล, analytics และ CRM ต้องรู้ โดยแต่ละตัวมี queue ของตัวเอง

Topic exchange

Topic exchange เทียบ routing key กับ pattern โดย routing key เป็นคำที่คั่นด้วยจุด และ binding pattern ใช้ wildcard ได้สองแบบ:

• * แทนคำได้หนึ่งคำพอดี
• # แทนคำได้ศูนย์คำหรือมากกว่า

Binding ที่เป็น # ตัวเดียวจะได้ทุกข้อความ ทำให้ topic exchange ทำตัวเหมือน fanout ส่วน binding ที่ไม่มี wildcard เลยก็ทำตัวเหมือน direct ความยืดหยุ่นนี้ทำให้ topic exchange เป็นตัวเลือกยอดนิยมของระบบแบบ event-driven

Headers exchange

Headers exchange ไม่ดู routing key แต่เทียบค่าใน header ของข้อความ argument x-match ของ binding กำหนดว่าต้องตรง all (ทุกตัว) หรือ any (ตัวใดตัวหนึ่ง):

must(ch.QueueBind("pdf-reports", "", "documents", false, amqp.Table{
	"x-match": "all",
	"format":  "pdf",
	"type":    "report",
}))

เหมาะกับการ route ตามหลาย attribute ที่ยัดลง key แบบมีจุดไม่ลงตัว แต่ในทางปฏิบัติเป็นแบบที่ใช้น้อยที่สุดในสี่แบบ

เปรียบเทียบ RabbitMQ exchange types

ระบบส่วนใหญ่ใช้แค่ direct กับ topic ก็ครอบคลุมเกือบทุกอย่างแล้ว

ตัวอย่าง topic exchange ครบวงจรใน Go

ตัวอย่างนี้ใช้ client ทางการของ Go คือ github.com/rabbitmq/amqp091-go ฟังก์ชัน must มีไว้ให้โค้ดสั้น ถ้าเป็น service จริงให้ return error แทนการปิดโปรแกรม

package main

import (
	"context"
	"log"
	"time"

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

func must(err error) {
	if err != nil {
		log.Fatal(err)
	}
}

func main() {
	conn, err := amqp.Dial("amqp://app:secret@localhost:5672/")
	must(err)
	defer conn.Close()

	ch, err := conn.Channel()
	must(err)
	defer ch.Close()

	// durable topic exchange
	must(ch.ExchangeDeclare("orders", "topic", true, false, false, false, nil))

	// durable quorum queue that dead-letters failed messages
	_, err = ch.QueueDeclare("orders.asia", true, false, false, false, amqp.Table{
		"x-queue-type":           "quorum",
		"x-dead-letter-exchange": "orders.dlx",
		"x-delivery-limit":       5,
	})
	must(err)
	must(ch.QueueBind("orders.asia", "order.*.asia", "orders", false, nil))

	// publisher confirms: the broker acknowledges each publish
	must(ch.Confirm(false))

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	dc, err := ch.PublishWithDeferredConfirmWithContext(ctx,
		"orders", "order.created.asia",
		true,  // mandatory: return the message if no queue matches
		false, // immediate: not supported by RabbitMQ
		amqp.Publishing{
			ContentType:  "application/json",
			DeliveryMode: amqp.Persistent,
			MessageId:    "ord-10231-created",
			Body:         []byte(`{"orderId":10231,"region":"asia"}`),
		})
	must(err)
	if !dc.Wait() {
		log.Println("broker nacked the message, publish again")
	}
}

ทำให้ RabbitMQ ส่งข้อความได้อย่างเชื่อถือได้

ข้อความหายได้สามจุด คือระหว่าง producer กับ broker, ภายใน broker เอง และระหว่าง broker กับ consumer แต่ละจุดต้องตั้งค่าแยกกัน

Publisher confirms: จาก producer ถึง broker

ถ้าไม่เปิด confirm Publish จะ return ทันทีที่ข้อมูลออกจากโปรเซส ถ้า broker ล่มหลังจากนั้นนิดเดียว เราจะไม่มีทางรู้ พอเปิด ch.Confirm แล้ว broker จะยืนยันข้อความแต่ละตัวเมื่อรับผิดชอบมันแล้ว ซึ่งสำหรับข้อความแบบ persistent หมายถึงหลังจากเก็บลง storage เรียบร้อย ให้นับว่า publish สำเร็จก็ต่อเมื่อได้ confirm กลับมาเท่านั้น

ตั้ง mandatory ไว้ด้วย แล้วรับข้อความที่ถูกตีกลับผ่าน ch.NotifyReturn ไม่อย่างนั้นข้อความที่ไม่ตรงกับ binding ไหนเลยจะหายไปเงียบ ๆ

Durable queue และ persistent message: รอดจากการ restart

Queue แบบ durable รอดจากการ restart ของ broker ส่วนข้อความแบบ persistent (DeliveryMode: amqp.Persistent) จะถูกเขียนลง disk ต้องมีทั้งสองอย่าง เพราะข้อความ persistent ที่อยู่ใน queue ที่ไม่ durable ก็ยังหายตอน restart อยู่ดี

ข้อมูลที่หายไม่ได้ให้ใช้ quorum queue ซึ่ง replicate ทุกข้อความไปยังโหนดส่วนใหญ่ของ cluster ด้วย Raft และเป็น durable เสมอ Classic mirrored queue ถูกถอดออกไปแล้วใน RabbitMQ 4.0 ดังนั้นตัวเลือกแบบ replicate ในปัจจุบันคือ quorum queue (หรือ stream)

Consumer acknowledgement: จาก broker ถึง consumer

เมื่อใช้ manual acknowledgement RabbitMQ จะเก็บข้อความไว้จนกว่า consumer จะยืนยันว่าประมวลผลเสร็จ ถ้า connection หรือ channel ของ consumer ปิดไปก่อน ack ข้อความจะถูกนำกลับเข้าคิวแล้วส่งให้ consumer ตัวอื่น

must(ch.Qos(20, 0, false)) // prefetch: at most 20 unacked messages for this consumer

msgs, err := ch.Consume("orders.asia", "billing-worker", false, false, false, false, nil)
if err != nil {
	log.Fatal(err)
}
for d := range msgs {
	if err := handleOrder(d.Body); err != nil {
		d.Nack(false, false) // don't requeue: send to the dead letter exchange
		continue
	}
	d.Ack(false)
}

ส่ง ack หลัง งานเสร็จและ commit แล้ว ไม่ใช่ตอนที่ข้อความมาถึง โหมด auto-ack (autoAck: true) จะลบข้อความทิ้งทันทีที่ส่งออกไป ถ้า crash ระหว่างทำงาน ข้อความนั้นก็หาย

ผลที่ได้คือการส่งแบบ at-least-once consumer อาจทำงานเสร็จแล้ว crash ก่อน ack แล้วได้รับข้อความเดิมอีกรอบ consumer จึงต้อง idempotent ให้บันทึก message ID ที่ประมวลผลแล้วใน transaction เดียวกับงานนั้น ซึ่งเป็นแนวคิดเดียวกับ idempotency key ในบทความ ป้องกัน race condition ในระบบ payment

Dead letter exchange: จัดการข้อความที่ทำยังไงก็ fail

บางข้อความทำยังไงก็ไม่มีวันสำเร็จ เช่น JSON ผิดรูปแบบ หรืออ้างถึง record ที่ถูกลบไปแล้ว ถ้าวน requeue ไปเรื่อย ๆ (Nack(false, true)) จะเกิด loop ที่ขวางข้อความอื่นทั้งหมด ให้ตั้ง dead letter exchange (DLX) แทน ข้อความจะถูกส่งไป DLX เมื่อ:

• consumer reject หรือ nack ด้วย requeue=false
• TTL หมดอายุ (x-message-ttl ที่ queue หรือ expiration ที่ตัวข้อความ)
• queue ยาวเกิน length limit ที่ตั้งไว้
• ถูกส่งซ้ำเกิน x-delivery-limit ของ quorum queue

Bind queue เข้ากับ DLX, ตั้ง alert ตามจำนวนข้อความที่ค้าง และเตรียมช่องทางให้ทีมเข้าไปตรวจและ replay ได้ ถ้าต้องการ retry แบบหน่วงเวลา pattern ที่ใช้กันบ่อยคือมี retry queue ที่ตั้ง TTL ไว้ และ DLX ของมันชี้กลับไปที่ exchange หลัก

ควรตั้ง DLX และ TTL ผ่าน policy (rabbitmqctl set_policy) มากกว่าใส่เป็น queue argument เพราะ policy เปลี่ยนทีหลังได้โดยไม่ต้องลบแล้วประกาศ queue ใหม่

Prefetch: กระจายงานให้สม่ำเสมอ

ค่า default ของ consumer คือไม่จำกัด prefetch worker ตัวเดียวจึงอาจกอดข้อความที่ยังไม่ ack ไว้เป็นพัน ๆ ขณะที่ตัวอื่นนั่งว่าง Qos(n, 0, false) จำกัดจำนวนข้อความที่ยังไม่ ack ต่อ consumer งานหนักและช้าให้เริ่มจากค่าน้อย งานเบาและเร็วใช้ค่ามากขึ้นได้ แล้วปรับจากการดู throughput และการใช้งานของ consumer จริง

เมื่อไหร่ควรใช้ RabbitMQ และเมื่อไหร่ไม่ควร

RabbitMQ เหมาะกับการกระจายงาน การ route ที่ยืดหยุ่น การ ack รายข้อความ และ pattern แบบ request/reply แต่ถ้าต้องการ event log ที่เก็บไว้นานและ replay ได้ โดยมี consumer หลายกลุ่มอ่านตาม offset ของตัวเอง โมเดลของ Kafka จะเหมาะกว่า อ่านต่อที่ Apache Kafka: topic, partition และ consumer group ส่วนการ broadcast แบบยิงแล้วลืมที่ข้อความหายได้ Redis Pub/Sub เบากว่า

คำถามที่พบบ่อย

RabbitMQ exchange types มีอะไรบ้าง?

มีสี่แบบ คือ direct (routing key ตรงกันเป๊ะ), fanout (broadcast ไปทุก queue ที่ bind ไว้), topic (wildcard pattern บน key ที่คั่นด้วยจุด) และ headers (เทียบค่าใน header ของข้อความ)

Direct exchange กับ topic exchange ต่างกันอย่างไร?

Direct เทียบ routing key กับ binding key แบบตรงตัว ส่วน topic เทียบกับ pattern ที่ใช้ * (หนึ่งคำ) และ # (ศูนย์คำขึ้นไป) binding เดียวจึงครอบคลุม key ที่เกี่ยวข้องกันได้หลายตัว

Default exchange ใน RabbitMQ คืออะไร?

คือ direct exchange ไม่มีชื่อ ที่ทุก queue ถูก bind ไว้อัตโนมัติโดยใช้ชื่อ queue เป็น binding key ถ้า publish ด้วย exchange "" และ routing key my-queue ข้อความจะไปที่ my-queue

RabbitMQ รับประกัน exactly-once ไหม?

ไม่ ถ้าใช้ publisher confirm, durable queue, persistent message และ manual ack จะได้ at-least-once ถ้าต้องการให้ ผลลัพธ์ เกิดครั้งเดียว ต้องทำ consumer ให้ idempotent

ข้อความที่ยังไม่ ack จะเป็นอย่างไรเมื่อ consumer crash?

เมื่อ channel หรือ connection ของ consumer ปิด RabbitMQ จะนำข้อความที่ยังไม่ ack กลับเข้าคิว แล้วส่งให้ consumer ตัวอื่นโดยมีเครื่องหมาย redelivered ติดไปด้วย

Checklist สำหรับ RabbitMQ ที่เชื่อถือได้

• เลือก exchange type อย่างตั้งใจ ส่วนใหญ่คือ direct หรือ topic
• ใช้ durable exchange และ quorum queue กับข้อมูลสำคัญ
• ฝั่ง producer ใช้ persistent message และ publisher confirm
• ตั้ง mandatory และจัดการข้อความที่ถูกตีกลับ
• ใช้ manual ack และ ack หลัง commit งานแล้วเท่านั้น
• Consumer idempotent โดยอิง message ID
• มี dead letter exchange พร้อม monitoring และช่องทาง replay
• ตั้ง prefetch และปรับให้เหมาะกับงาน

ถ้ากำลังออกแบบระบบ event-driven และอยากให้ชั้น messaging ถูกต้องตั้งแต่ต้น Vectorkub พัฒนาระบบ backend บน RabbitMQ และ Kafka อยู่เป็นประจำ