บทที่ 07 — RabbitMQ Deep (เข้มข้น)

← บทที่ 06: Broker Fundamentals | สารบัญ | บทที่ 08: Kafka Deep →

TL;DR: บทนี้ตอบ "ใช้ RabbitMQ (อ่าน แรบ-บิท-เอ็ม-คิว — message broker ยอดนิยมที่รองรับ routing ซับซ้อน) ระดับ production ยังไง" ครอบคลุมหัวข้อหลัก ๆ ดังนี้:

• AMQP (Advanced Message Queuing Protocol — โปรโตคอลกลางของ message queue ที่ RabbitMQ เป็น implementation ต้นแบบ) architecture
• Exchange types (direct = ตรง / topic = pattern match / fanout = broadcast / headers = ดู header)
• Publisher confirm + Consumer ack + Prefetch (กลไกกัน message หาย/ล้น)
• Classic vs Mirrored vs Quorum vs Stream queue (เปรียบเทียบ)
• DLX (Dead Letter Exchange) + DLQ (Dead Letter Queue) retry pattern
• Cluster + HA, performance tuning

ข้ามได้ถ้า: ไม่ได้ใช้ RabbitMQ (ใช้ Kafka/SQS/NATS แทน) — กลับมาอ่านเมื่อต้องการ task queue + complex routing

📖 กางศัพท์ RabbitMQ (ที่จะเห็นตลอดบท):

คำ	อ่าน/ความหมาย
Exchange	"ตู้คัดแยกจดหมาย" — รับ message จาก producer แล้ว route ไป queue ตามกฎ
Queue	"คิว/กล่อง" — เก็บ message รอให้ consumer มาเอา
Binding	"กฎผูก" — บอกว่า exchange ส่งไป queue ไหน เมื่อ routing key match อะไร
Routing Key	"กุญแจ routing" — string ที่ producer ติดมากับ message ใช้ตัดสินว่าจะ route ไหน
Channel	"ช่อง" — connection ย่อย (virtual) ภายใน 1 TCP connection — เปิด-ปิดเร็ว, ใช้ใน thread หนึ่ง ๆ
Connection	TCP connection จริงไปยัง broker — เปิดแพง ต้อง reuse
prefetch	"พรี-เฟทช์ = ดึงล่วงหน้า" — กำหนดว่า consumer มี message ค้าง (ยังไม่ ack) ได้กี่ตัว, ป้องกันดึงเกิน memory
DLX	Dead Letter Exchange = exchange ที่รับ "จดหมายตาย" (message ที่ประมวลไม่ได้/หมด TTL) แล้ว route ต่อไป DLQ
DLQ	Dead Letter Queue = queue ปลายทางของจดหมายตาย — รอ ops review
AMQP	(เอ-เอ็ม-คิว-พี) Advanced Message Queuing Protocol — โปรโตคอลที่ RabbitMQ ใช้

Exchange types 4 ชนิด:

• direct (ตรง) — match routing key เป๊ะ ๆ
• topic (หัวข้อ) — pattern match เช่น order.* (ดู Part 3.2)
• fanout (พัด/broadcast) — ส่งไปทุก queue ที่ bind, ไม่สน routing key
• headers — match จาก message header ไม่ใช่ routing key

บทนี้ลึกถึงระดับ "ติดตั้ง + ใช้ production" — ไม่ใช่แค่ "send message ได้"

ครอบคลุม:

• AMQP protocol + architecture
• Exchange + Queue + Binding (ลึก)
• Publisher confirm + consumer ack + prefetch
• Persistent message + durable queue
• Classic, Mirrored, Quorum, Stream queue (เปรียบเทียบ)
• Dead Letter Exchange + Retry Pattern
• Cluster + HA
• Performance tuning
• Monitoring + Operations

ใช้เวลา 5-6 ชั่วโมง (บทใหญ่ที่สุดของหนังสือ)

🗺️ สารบัญด่วน (คลิกข้ามไปหัวข้อที่ต้องการ):

1. Part 1: RabbitMQ คืออะไร
2. Part 2: AMQP Architecture
3. Part 3: Exchange Types
4. Part 4: Publisher Confirms + Returns
5. Part 5: Consumer Ack + Prefetch
6. Part 6: Durable Queue + Persistent Message
7. Part 7: Dead Letter Exchange (DLX) + Retry
8. Part 8: Spring AMQP Integration
9. Part 9: Queue Types
10. Part 10: Dead Letter Exchange + Retry
11. Part 11: Cluster + HA
12. Part 12: Performance Tuning
13. Part 13: Monitoring
14. Part 14: Operational Patterns
15. Part 15: Anti-Patterns
16. Part 16: Lab
17. Part 16.5: Production War Stories
18. Part 16.6: Cheat Sheet
19. Part 17: Checkpoint
20. Part 18: สรุปบทนี้

---

Part 1: RabbitMQ คืออะไร

1.1 ภาพรวม

• Erlang-based message broker (Pivotal/VMware, ตอนนี้อยู่ใต้ Broadcom) — ไม่ต้องรู้ Erlang เพื่อใช้งาน — เป็นแค่ภาษาที่เขียนตัว broker
• Protocol หลัก: AMQP 0-9-1 + AMQP 1.0 native (ตั้งแต่ RabbitMQ 4.0; ก่อนหน้านี้ต้องใช้ plugin)
• รองรับ MQTT 5 (ตั้งแต่ 3.13), STOMP, AMQP 1.0 ด้วย
• เน้น: smart broker, dumb consumer — broker ทำ routing ให้
• Version snapshot (2026): RabbitMQ 4.x (4.0 GA Sep 2024, 4.1.x current as of mid-2026 — ตรวจสอบเวอร์ชันล่าสุดก่อนใช้จริง) สรุปเป็นข้อ ๆ ดังนี้:
  • Quorum Queue เป็น recommended สำหรับ HA ใน 4.x — แต่ไม่ได้เป็น default อัตโนมัติ ต้องตั้ง default_queue_type = quorum ใน rabbitmq.conf หรือระบุ x-queue-type=quorum ตอนสร้าง queue เอง ไม่งั้น broker จะสร้าง Classic Queue v2 ให้แทน
  • Streams queue type พร้อมใช้งานแล้ว
  • metadata store (ที่เก็บ "ข้อมูลโครงสร้าง" ของ broker เอง เช่น มี queue อะไรบ้าง ผูก binding กับอะไร — ไม่ใช่ที่เก็บตัว message) กำลังเปลี่ยนจาก Mnesia → Khepri (เคฟ-รี — metadata store ใหม่ที่ใช้ Raft consensus ทำให้ cluster recovery + schema sync เสถียรกว่าเดิม): ใน RabbitMQ 4.0 ยังใช้ Mnesia เป็น default (Khepri เป็นแค่ตัวเลือก เปิดผ่าน feature flag), แล้วค่อยกลายเป็น default ใน 4.1
• License: RabbitMQ OSS ยังเป็น MPL 2.0; ถ้าซื้อ commercial support → ตรวจ Tanzu policy หลัง Broadcom takeover (2023-2024)

🆕 2026 baseline highlights (RabbitMQ 4.0/4.1):

• Classic Queue v1 ถูก REMOVED ใน 4.0 — เหลือแค่ Classic v2 (CQv2). ต้อง migrate v1 → v2 ใน 3.13 ก่อน upgrade
• Mirrored queues (ha-mode: all) ถูก REMOVED ใน 4.0 → ใช้ Quorum Queue แทน
• Lazy queue mode (x-queue-mode=lazy) ถูก REMOVED ใน 4.0 — deprecated ตั้งแต่ 3.12. CQv2 + Quorum เก็บ disk-first โดย default อยู่แล้ว
• Khepri (เคฟ-รี — Raft-based metadata store) แทน Mnesia — schema sync + cluster recovery เสถียรกว่า (opt-in ใน 4.0, กลายเป็น default ใน 4.1)
• AMQP 1.0 native (ไม่ต้อง plugin), MQTT 5 (ตั้งแต่ 3.13)
• Streams (3.9+) + Super-streams (3.11+) — partitioned streams + Single Active Consumer
• OAuth 2.0 + JWT auth (3.13+), LDAP plugin ยังรองรับ
• Priority queue = Classic only (Quorum + Stream ไม่รองรับ — limitation ที่ต้องรู้)

ตรงข้ามกับ Kafka ที่ "dumb broker, smart consumer"

📖 กางศัพท์ "smart vs dumb broker":

• Smart broker (RabbitMQ, SQS) = broker "ฉลาด" — รู้ว่า message ไหนไปไหน (routing rule), track ว่าใคร ack แล้ว, ลบ message หลัง consume → consumer แค่รับแล้วทำงาน (dumb)
• Dumb broker (Kafka) = broker "เก็บอย่างเดียว" ไม่ track ว่าใคร consume — consumer ฉลาด จำ offset เอง, decide ว่าจะอ่านซ้ำหรือไม่
• Trade-off: Smart broker = setup ง่าย แต่ scale ที่ broker จำกัด (track state เปลือง). Dumb broker = scale ได้สูงมาก (broker แค่ append-only log) แต่ client ต้องฉลาดกว่า

1.2 เมื่อไหร่ใช้ RabbitMQ

✅ Task queue / background job ✅ RPC over message (RPC = Remote Procedure Call — เรียก function ข้าม service โดยใช้ message + reply queue แทน HTTP) ✅ Complex routing (header, topic, fanout) ✅ Per-message TTL (Time-To-Live = อายุ message — หมดอายุ → broker ลบ/ส่งไป DLX), priority ✅ ต้องการ ack per message (แต่ละ message มีการยืนยันรับเป็นรายตัว — กัน message หาย) ✅ Low latency (< 5ms typical)

❌ Event log / replay ❌ Throughput > 100K msg/sec ต่อ queue (ข้อจำกัดนี้หมายถึง Classic/Quorum queue — Stream queue ทำ throughput สูงกว่านี้ได้ ดู Part 9.4) ❌ Long retention (ใช้ Stream queue ได้บ้างแต่ Kafka เก่งกว่า)

1.3 ติดตั้ง

รัน RabbitMQ ผ่าน Docker — image -management มี Management UI ติดมาด้วย:

# ⚠️ admin/admin เป็น credentials สำหรับ local dev เท่านั้น — production ต้องใช้ secret manager
docker run -d --name rabbit \
  -p 5672:5672 -p 15672:15672 -p 15692:15692 \
  -e RABBITMQ_DEFAULT_USER=admin \
  -e RABBITMQ_DEFAULT_PASS=admin \
  rabbitmq:4.1-management

• Port 5672 = AMQP (จำเป็น)
• Port 15672 = Management UI (จำเป็น)
• Port 15692 = Prometheus metrics — optional สำหรับผู้เพิ่งเริ่ม ใช้ใน Part 13 (monitoring); ถ้าเพิ่งเริ่ม ข้ามได้ก่อน ใช้แค่ 5672 และ 15672
• เปิด http://localhost:15672 (admin/admin)

---

Part 2: AMQP Architecture

2.1 องค์ประกอบ

• Producer ส่งไปที่ Exchange (ไม่ใช่ queue ตรง ๆ!)
• Exchange route ไป Queue ตาม binding (= "กฎผูก exchange กับ queue ตามเงื่อนไข routing key")
• Consumer อ่านจาก Queue

💡 Mermaid render note: ถ้า diagram ข้างบนไม่แสดง subgraph "Routing Rule" ครบใน viewer ของคุณ (เช่น GitHub Pages บางธีม) — fallback ข้อความ: "Exchange มี routing rule ภายในที่ดูจาก routing key + exchange type + headers เพื่อตัดสินใจส่งไป queue ไหนตาม binding"

2.2 Connection vs Channel

Rules:

• ใช้ น้อย connection (เปิด TCP แพง)
• ใช้ หลาย channel บน 1 connection (cheap, multiplex)
• 1 channel ห้าม share ระหว่าง thread → 1 thread = 1 channel
• ใช้ channel pool

---

Part 3: Exchange Types

3.1 Direct Exchange

📖 ปูพื้นก่อน: binding = "กฎผูก queue เข้ากับ exchange ตามเงื่อนไข routing key" — สร้างผ่าน BindingBuilder.bind(queue).to(exchange).with(key) หรือ Management UI. ทุก queue ที่อยากรับ message ต้อง bind กับ exchange ก่อน

Direct exchange route message ไป queue ที่มี routing key "ตรงเป๊ะ" กับที่ producer ส่งมา ถ้า key ไม่ตรง queue ไหนเลย message จะถูกทิ้ง:

binding routing_key = exact match

Producer publishes to exchange with routing key "order.created"
   ↓
Exchange routes to queue bound with key "order.created"

ตัวอย่าง:

Exchange: "orders"
Queue "high-priority" bound with "high"
Queue "normal" bound with "normal"

Publish with key="high"   → "high-priority" queue
Publish with key="normal" → "normal" queue
Publish with key="urgent" → ไม่ไป queue ไหน (ลบทิ้ง — unless unrouted handling)

3.2 Topic Exchange

Topic exchange ยืดหยุ่นกว่า direct — routing key เป็น dotted path คั่นจุด แล้ว bind queue ด้วย wildcard:

routing key เป็น dotted path + wildcard:

• * = 1 word เป๊ะ (asterisk = small star = 1 ตัว)
• # = 0 หรือหลาย word (hash = "เยอะ")

Queue A bound with "order.*.created"
Queue B bound with "order.#"
Queue C bound with "#.created"

Publish "order.book.created":
  → A (A binds "order.*.created", * รับคำ "book" ตรงกลาง)
  → B (# รับ "book.created")
  → C (# รับ "order.book" ก่อน ".created")

Publish "order.created":
  → B
  → C
  (A ไม่ตรง — `*` ต้องมี 1 word ตรงกลาง)

ดีสำหรับ event-driven กับ topic ที่มี hierarchy

3.3 Fanout Exchange

Producer → Exchange → ทุก queue ที่ bound (ignore routing key)

ใช้ broadcast — "ทุกคนรู้เรื่องเดียวกัน"

ตัวอย่าง: log aggregation, notification fanout

3.4 Headers Exchange

Headers exchange route โดยดูจาก message header แทน routing key:

match จาก message header (ไม่ใช่ routing key):

queue binding:
  "x-match" = "all"   // ต้องตรงทุก header
  format = "pdf"
  type = "invoice"

publish message with headers: format=pdf, type=invoice → match
publish with format=pdf, type=quote → no match

ไม่ค่อยใช้ — topic exchange มักพอ

3.5 Default Exchange (Built-in)

Exchange ชื่อ "" (empty) — direct exchange ที่ทุก queue auto-bound กับ routing key = queue name

channel.basicPublish("", "myqueue", null, body.getBytes());
// → ส่งตรงไป queue "myqueue"

ใช้ใน test / simple case แต่ production ไม่ควร — ขาด routing flexibility

3.6 Custom Exchange Types (Plugin)

นอกจาก exchange มาตรฐาน RabbitMQ ยังเพิ่มความสามารถผ่าน plugin ได้:

• Consistent Hash Exchange — partition by key (mimic Kafka)
• Delayed Message Exchange — schedule delivery
• Random Exchange — pick random queue

---

Part 4: Queue Properties

QueueBuilder.durable("orders")
    .withArgument("x-message-ttl", 60_000)              // TTL ต่อ message
    .withArgument("x-expires", 1_800_000)               // queue auto-delete หลัง idle
    .withArgument("x-max-length", 100_000)              // max queue size
    .withArgument("x-max-length-bytes", 100_000_000)
    .withArgument("x-overflow", "reject-publish")        // drop-head, reject-publish, reject-publish-dlx
    .withArgument("x-dead-letter-exchange", "dlx")
    .withArgument("x-dead-letter-routing-key", "dlq")
    .withArgument("x-max-priority", 10)                  // priority 0-10
    .withArgument("x-queue-type", "quorum")              // default: classic
    .build();

4.1 Durable / Transient

• durable=true → queue survive broker restart (metadata)
• durable=false → ลบเมื่อ broker restart

Default: ใช้ durable เกือบทุกครั้งใน production

4.2 Exclusive

.exclusive()

→ queue ใช้ได้แค่ connection ที่สร้าง — ปิด connection = queue หาย

ใช้ใน RPC reply queue

4.3 Auto-delete

.autoDelete()

→ queue ลบเมื่อไม่มี consumer

---

Part 5: Message Properties

MessageProperties props = MessagePropertiesBuilder.newInstance()
    .setContentType("application/json")
    .setMessageId(UUID.randomUUID().toString())
    .setCorrelationId(correlationId)
    .setReplyTo("reply-queue")              // สำหรับ RPC
    .setExpiration("60000")                  // per-message TTL ms
    .setPriority(5)                           // 0-10
    .setDeliveryMode(MessageDeliveryMode.PERSISTENT)   // persistent / non-persistent
    .setHeader("source", "order-service")
    .setHeader("trace-id", traceId)
    .build();

5.1 Persistent vs Non-persistent

• Persistent (delivery_mode=2) → broker เขียน disk
• Non-persistent (delivery_mode=1) → memory only

⚠️ Persistent ต้อง queue durable + exchange durable ด้วย ถึงจะ survive

5.2 TTL

Per-message TTL (กำหนดตอน publish):

.setExpiration("60000")   // 60 sec

Per-queue TTL (กำหนดที่ queue → ทุก message ในนั้นใช้ค่านี้):

.withArgument("x-message-ttl", 60_000)

📝 ต่างกันยังไง?

• Per-message TTL = flexible — message แต่ละตัวกำหนด TTL ของตัวเอง (เช่น notification สำคัญ 1 ชม., notification ทั่วไป 5 นาที). ใช้เมื่อ TTL หลากหลายในคิวเดียว
• Per-queue TTL (x-message-ttl) = uniform — ทุก message ในคิวใช้ค่าเดียวกัน. ถ้าทั้งสองตั้งพร้อมกัน → broker ใช้ค่าที่น้อยกว่า

หลัง TTL → broker ลบ (หรือ route ไป DLX ถ้ามี)

5.3 Priority

.withArgument("x-max-priority", 10)   // queue support priority

// publish
.setPriority(9)  // urgent

⚠️ Priority queue หนัก performance ลดลง ตามจำนวน level — official guidance: ≤ 5 levels เพื่อ performance (technical max 255 — แต่ใช้เกิน 5 broker ต้อง maintain heap แยก per priority แล้วจะช้ามาก). Priority queue รองรับเฉพาะ Classic queue — Quorum queue ไม่รองรับ priority, Stream queue ก็ไม่รองรับ priority (append-only log = no priority semantics). ถ้าต้องการ priority + HA ให้แยกหลาย queue (high/normal/low) แล้ว consumer ดึงตามลำดับเอง

---

Part 6: Spring AMQP — Setup

pom.xml:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

application.yml:

spring:
  rabbitmq:
    host: rabbit
    port: 5672
    username: admin
    password: ${RABBIT_PASSWORD}
    virtual-host: /
    connection-timeout: 5s
    publisher-confirm-type: correlated
    publisher-returns: true
    template:
      mandatory: true
      retry:
        enabled: true
        max-attempts: 3
        initial-interval: 1s
        max-interval: 10s
        multiplier: 2
    listener:
      simple:
        acknowledge-mode: manual
        prefetch: 10
        concurrency: 3
        max-concurrency: 10
        retry:
          enabled: true
          max-attempts: 3

6.1 Topology Declaration

ประกาศ topology เป็น @Bean — Spring สร้างทุกอย่างใน broker ให้อัตโนมัติตอน app start:

@Configuration
public class RabbitConfig {
    public static final String ORDERS_EX = "orders.exchange";
    public static final String ORDERS_Q = "orders.created";
    public static final String ORDERS_DLX = "orders.dlx";
    public static final String ORDERS_DLQ = "orders.dlq";

    @Bean
    public TopicExchange ordersExchange() {
        return ExchangeBuilder.topicExchange(ORDERS_EX).durable(true).build();
    }

    @Bean
    public DirectExchange ordersDlx() {
        return ExchangeBuilder.directExchange(ORDERS_DLX).durable(true).build();
    }

    @Bean
    public Queue ordersQueue() {
        return QueueBuilder.durable(ORDERS_Q)
            .withArgument("x-queue-type", "quorum")
            .withArgument("x-dead-letter-exchange", ORDERS_DLX)
            .withArgument("x-dead-letter-routing-key", "orders.dlq")
            .build();
    }

    @Bean
    public Queue ordersDlq() {
        return QueueBuilder.durable(ORDERS_DLQ).build();
    }

    @Bean
    public Binding ordersBinding() {
        return BindingBuilder.bind(ordersQueue())
            .to(ordersExchange())
            .with("order.created.*");
    }

    @Bean
    public Binding dlqBinding() {
        return BindingBuilder.bind(ordersDlq())
            .to(ordersDlx())
            .with("orders.dlq");
    }

    @Bean
    public Jackson2JsonMessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory cf, MessageConverter mc) {
        RabbitTemplate t = new RabbitTemplate(cf);
        t.setMessageConverter(mc);
        t.setMandatory(true);
        t.setConfirmCallback((cd, ack, cause) -> {
            if (!ack) log.error("nack: {} {}", cd, cause);
        });
        t.setReturnsCallback(returned -> {
            log.error("returned: exchange={} key={} reply={}",
                returned.getExchange(), returned.getRoutingKey(), returned.getReplyText());
        });
        return t;
    }
}

---

Part 7: Publisher Confirm + Returns

7.1 Publisher Confirm

ปัญหา: producer send() return ทันที — ไม่รู้ว่า broker เก็บได้ไหม

Solution: ขอ broker ส่ง confirm กลับ

spring.rabbitmq.publisher-confirm-type: correlated

rabbit.setConfirmCallback((correlationData, ack, cause) -> {
    if (ack) {
        log.info("confirmed: {}", correlationData);
    } else {
        log.error("nack: {}", cause);
        // retry / save outbox
    }
});

rabbit.convertAndSend(exchange, key, payload,
    new CorrelationData("msg-id-123"));

⚠️ อย่า block ใน confirm callback — callback นี้ถูก invoke บน I/O thread ของ Rabbit client. ถ้าทำงานหนัก (เช่น DB write, network call) ใน callback → stall I/O ของ publisher ทั้งหมด. ให้ delegate ไป executor: executor.submit(() -> updateOutboxStatus(...))

3 ผลลัพธ์:

• ack = broker เก็บได้
• nack = broker ปฏิเสธ (error)
• timeout = ไม่ตอบ → retry

7.2 Returns (mandatory)

ปัญหา: ส่ง message ไป exchange ที่ไม่มี queue bound → message หาย

Solution:

spring.rabbitmq.publisher-returns: true
spring.rabbitmq.template.mandatory: true

rabbit.setReturnsCallback(returned -> {
    log.error("undelivered: {} {} {}",
        returned.getExchange(),
        returned.getRoutingKey(),
        returned.getReplyText());
});

→ broker คืน message + reason

7.3 ⚠️ Confirm + Outbox

Producer:
1. INSERT order ใน DB
2. INSERT outbox event ใน DB (same tx)
3. COMMIT

Background process:
1. Read unsent outbox
2. Publish to RabbitMQ
3. Wait confirm
4. ถ้า ack → mark sent
5. ถ้า nack/timeout → retry

→ atomic + at-least-once

---

Part 8: Consumer Ack

8.1 3 modes

spring.rabbitmq.listener.simple.acknowledge-mode: <mode>

• none — auto-ack (consumer ลบ message ทันที, อันตราย)
• auto — Spring ack หลัง method return; nack ถ้า exception
• manual — control เอง ← production!

8.2 Manual ack

@RabbitListener(queues = "orders.created")
public void onOrder(
    @Payload OrderEvent event,
    Channel channel,
    @Header(AmqpHeaders.DELIVERY_TAG) long tag,
    @Header(AmqpHeaders.REDELIVERED) boolean redelivered) throws IOException {

    try {
        if (redelivered) {
            log.warn("redelivered: {}", event.id());
        }
        process(event);
        channel.basicAck(tag, false);                  // ack
    } catch (TransientException e) {
        channel.basicNack(tag, false, true);           // requeue=true
    } catch (BusinessException e) {
        channel.basicReject(tag, false);                // → DLX
    } catch (Exception e) {
        // unexpected — ส่งไป DLX
        channel.basicReject(tag, false);
    }
}

⚠️ basicReject(tag, false) ทำงานคู่กับ DLX config:

• ถ้า queue มี x-dead-letter-exchange → message ไป DLX ตามตั้ง
• ถ้าไม่มี DLX config → message จะถูก drop ทันที (หายไป!)
• แนะนำตั้ง Spring default-requeue-rejected: false + config DLX ทุก queue สำคัญ

basicAck(tag, multiple):

• multiple=false — ack แค่ tag นี้
• multiple=true — ack ทุก tag ≤ tag นี้ (batch ack)

8.3 Prefetch (QoS)

spring.rabbitmq.listener.simple.prefetch: 10

= 1 consumer เก็บ unacked message ไม่เกิน 10

Trade-off:

• prefetch ใหญ่ → throughput สูง แต่ failover ช้า (message ค้าง consumer ที่ตาย)
• prefetch เล็ก → throughput ต่ำ แต่ fairer distribution

Rule of thumb: start จาก 10 → ปรับตาม monitor

---

Part 9: Queue Types เปรียบเทียบ

RabbitMQ มี 4 queue type — เลือกผิดมีผลใหญ่

9.1 Classic Queue (default เก่า)

• CQv2 (RabbitMQ 4.0+) เก็บ disk-first เป็น default — memory เฉพาะ hot data (Erlang process per queue)
• CQv1 (เก่า) เก็บ memory + disk → ขัดข้องเมื่อ memory pressure
• ไม่มี replication built-in (single-node only)
• 🔴 Classic Queue v1 ถูก REMOVED ใน RabbitMQ 4.0 — เหลือแค่ Classic v2 (CQv2). ต้อง migrate v1 → v2 ใน 3.13 ก่อน upgrade ไป 4.0
  • Migration paths: classic v2 (default), quorum queues, หรือ streams
• ⚠️ Classic Queue Mirror policy (HA, ha-mode: all/ha-all) ถูก REMOVED ใน 4.0 — ถ้าต้องการ HA ให้ใช้ Quorum แทน
• CQv2 ยังเหมาะสำหรับ dev / single-node / non-HA workload

9.2 Mirrored Queue (removed in 4.0)

• Classic + replicate ไป mirror node (leader/follower)
• ⚠️ Removed ใน RabbitMQ 4.0 (Sep 2024) — deprecated มาตั้งแต่ 3.8
• ถ้า upgrade จาก 3.x → ต้องย้ายไป Quorum ก่อน

9.3 Quorum Queue (recommended ใน 4.x)

• Replicated using Raft consensus
• Survive node failure (quorum = N/2 + 1)
• เน้น durability + safety > throughput
• Recommended สำหรับ HA ตั้งแต่ RabbitMQ 4.0 (แทน Mirror policy ที่ถูกลบ) — แต่ "recommended" ไม่ใช่ "automatic default": ต้องตั้ง default_queue_type = quorum ใน rabbitmq.conf ถ้าต้องการให้ broker สร้าง quorum โดย default. ไม่งั้นได้ Classic v2 เมื่อ client ไม่ระบุ x-queue-type
• Available since 3.8 (GA, Oct 2019)
• ไม่รองรับ priority queue (ใช้ Classic ถ้าต้องการ priority)
• รองรับ Single Active Consumer (SAC) สำหรับงานที่ต้อง strict ordering

QueueBuilder.durable("orders")
    .withArgument("x-queue-type", "quorum")
    .withArgument("x-quorum-initial-group-size", 3)  // ต้อง <= cluster broker count
    .build();

📝 x-quorum-initial-group-size = จำนวน replica เริ่มต้น ต้อง ≤ จำนวน broker ใน cluster (เช่น 3 ในตัวอย่างต้องมี ≥ 3 node)

ถ้ารัน RabbitMQ node เดียว (เช่น ตาม Docker command ใน Part 1.3 สำหรับ dev) ให้ตั้งค่านี้ = 1 หรือข้าม Quorum queue ไปก่อน — Quorum queue ต้องมีหลาย node ถึงจะได้ประโยชน์เรื่อง replication

9.4 Stream Queue (Kafka-like)

• Append-only log
• Replay จาก offset ได้
• Throughput สูง — แต่ตัวเลขที่มักถูกอ้างต้องแยกเงื่อนไขให้ชัด (ตรวจสอบ benchmark ล่าสุดก่อนใช้วางแผนจริง เพราะตัวเลขจาก vendor เปลี่ยนได้เร็ว):
  • ตัวเลข best-case: benchmark จาก RabbitMQ Labs เคยรายงานหลักแสนถึงเกือบ 1M msg/sec — แต่เป็นตัวเลขรวมจากหลาย stream/connection ไม่ใช่ node เดียวโดด ๆ
  • เงื่อนไข benchmark: ทดสอบด้วย payload ขนาดเล็ก, ยิง loop ต่อเนื่องไม่หยุด, ไม่มี fsync (ไม่บันทึกลง disk จริงระหว่างทดสอบ) — คือสภาพที่ดีที่สุดเท่าที่จะทำได้ ไม่ใช่สภาพใช้งานจริง
  • ตัวเลขที่ควรใช้วางแผนจริง: ~50-200K msg/sec ต่อ node ขึ้นกับ payload size, fsync policy, network — นี่คือตัวเลขที่ใกล้เคียง production
  • ทำไมสูงกว่าตัวเลขรวมของ RabbitMQ ในบทที่ 10 (~50K msg/sec ทั่วไป): เพราะ Stream queue เป็น append-only log ที่ออกแบบมาเพื่อ throughput โดยเฉพาะ ต่างจาก Classic/Quorum queue ที่เน้น routing + ack semantics มากกว่า
• Retention by time/size
• Server-side offset tracking (track offset ฝั่ง broker ได้ — ต่างจาก Kafka ที่ client เป็นคนจำ)
• Super-streams (3.11+) = partitioned streams (กระจาย stream เป็นหลาย partition แบบ Kafka)
• รองรับ Single Active Consumer (SAC) สำหรับงานที่ต้องการ strict ordering แม้มีหลาย consumer
• Use case: event sourcing, replay, audit log

QueueBuilder.durable("events")
    .withArgument("x-queue-type", "stream")
    .withArgument("x-max-length-bytes", 2_000_000_000)
    .withArgument("x-max-age", "7D")
    .build();

9.5 ภาพรวม

	Classic	Quorum	Stream
Replication	❌	✅ Raft	✅
Durability	กลาง	สูง	สูง
Throughput	กลาง	กลาง	สูง
Replay	❌	❌	✅
Use case	legacy	task/event production	log/event sourcing

2026 default: Quorum สำหรับงาน critical, Stream สำหรับ event ที่ต้อง replay

---

Part 10: Dead Letter Exchange + Retry

10.1 Basic DLX

@Bean
public Queue ordersQueue() {
    return QueueBuilder.durable("orders")
        .withArgument("x-dead-letter-exchange", "dlx")
        .withArgument("x-dead-letter-routing-key", "orders.dlq")
        .build();
}

Message → DLX เมื่อ:

1. Consumer reject (basicNack/basicReject with requeue=false)
2. TTL หมด
3. Queue length เกิน + overflow=drop-head

10.2 Retry with Delay Queue Pattern

@Bean
public Queue mainQueue() {
    return QueueBuilder.durable("orders")
        .withArgument("x-dead-letter-exchange", "orders.dlx")
        .withArgument("x-dead-letter-routing-key", "orders.retry")
        .build();
}

@Bean
public Queue retryQueue() {
    return QueueBuilder.durable("orders.retry")
        .withArgument("x-message-ttl", 5000)             // wait 5s
        .withArgument("x-dead-letter-exchange", "orders.ex")
        .withArgument("x-dead-letter-routing-key", "orders.created")
        .build();
}

@Bean
public Queue finalDlq() {
    return QueueBuilder.durable("orders.dlq").build();
}

// ต้องประกาศ exchange + binding ของ retry path ด้วย
@Bean DirectExchange retryExchange() { return new DirectExchange("orders.retry.ex"); }
@Bean Binding retryBinding() {
    return BindingBuilder.bind(retryQueue()).to(retryExchange()).with("orders.retry");
}

ใส่ retry counter ใน header — ถ้าเกิน 3 → ไป DLQ จริง:

public void onMessage(Message msg, Channel ch) throws IOException {
    long tag = msg.getMessageProperties().getDeliveryTag();
    try {
        process(msg);
        ch.basicAck(tag, false);
    } catch (Exception e) {
        // ดึง header เป็น Number (RabbitMQ จาก network อาจส่ง Long/LongString ไม่ใช่ Integer
        // → cast ตรงเป็น Integer เสี่ยง ClassCastException, ใช้ Number ปลอดภัยกว่า)
        Object headerObj = msg.getMessageProperties().getHeader("x-retry-count");
        int retry = headerObj instanceof Number n ? n.intValue() : 0;
        if (retry >= 3) {
            // เกินกว่า retry → final DLQ (reject ไม่ requeue)
            ch.basicReject(tag, false);
        } else {
            msg.getMessageProperties().setHeader("x-retry-count", retry + 1);
            rabbitTemplate.send("orders.retry.ex", "", msg);
            ch.basicAck(tag, false);
        }
    }
}

หรือใช้ Delayed Message Plugin (rabbitmq_delayed_message_exchange):

@Bean
public CustomExchange delayedExchange() {
    return new CustomExchange("delayed-ex", "x-delayed-message", true, false,
        Map.of("x-delayed-type", "direct"));
}

// publish with delay
rabbit.convertAndSend("delayed-ex", "key", payload, msg -> {
    msg.getMessageProperties().setHeader("x-delay", 5000);
    return msg;
});

---

Part 11: Cluster + HA

11.1 Cluster

3 nodes:
  rabbit1 (master discovery)
  rabbit2
  rabbit3

# node 2 + 3
rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl join_cluster rabbit@rabbit1
rabbitmqctl start_app

⚠️ Hostname resolution บังคับ: format rabbit@<hostname> ใช้ Erlang node name ที่ต้อง resolve hostname ได้ — ถ้า rabbit1 resolve ไม่เจอ (ไม่มี DNS, ไม่ได้ตั้ง /etc/hosts) → join fail. ต้องทำอย่างใดอย่างหนึ่ง:

1. ตั้ง DNS / /etc/hosts ให้ทุก node เห็น hostname กันได้
2. ใช้ RABBITMQ_USE_LONGNAME=true + FQDN
3. ใน Kubernetes ใช้ StatefulSet + headless Service (pod ได้ stable hostname อัตโนมัติ)

และทุก node ต้องมี Erlang cookie เดียวกัน (/var/lib/rabbitmq/.erlang.cookie) ไม่งั้น auth fail

→ metadata (exchange, queue definition) sync ทุก node → Quorum queue replicate ทุก node

11.2 Quorum Group Size

.withArgument("x-quorum-initial-group-size", 3)

ใน cluster 3 node:

• Survive 1 down (need 2 → 2/3 quorum OK)
• ไม่ survive 2 down (1/3 ไม่ pass quorum)

Recommended: 3 หรือ 5 node

11.3 Network Partition

ถ้า cluster แตก:

• Quorum side ทำงานต่อได้
• Minority side หยุด (avoid split-brain)

Setting:

cluster_partition_handling = autoheal | pause_minority | ignore

→ ใช้ pause_minority ใน production

11.4 Load Balancer

Client connect ผ่าน LB → ถ้า node ตาย → reconnect

Spring AMQP รองรับ multi-host:

spring.rabbitmq.addresses: rabbit1:5672,rabbit2:5672,rabbit3:5672

---

Part 12: Performance Tuning

12.1 Throughput Tips

1. Persistent message + durable queue = ช้ากว่า 3-5x — ใช้เท่าจำเป็น
2. Batch publish — เปิด tx หรือ publisher-confirms แบบ async
3. Prefetch สูงขึ้น = throughput สูง
4. Concurrent consumer — concurrency: 10
5. Stream queue ถ้าต้องการ throughput สูง
6. Lazy queue mode (x-queue-mode=lazy) — 🔴 REMOVED ใน RabbitMQ 4.0 (deprecated ตั้งแต่ 3.12). CQv2 + Quorum queue เก็บ disk-first โดย default อยู่แล้ว — ไม่ต้องตั้งเพิ่ม. ถ้ายังใช้ 3.x ที่มี policy queue-mode=lazy → migrate ก่อน upgrade
7. Disk-first paging อัตโนมัติ — CQv2/Quorum จัดการ paging ระหว่าง memory/disk เองตาม load. ตั้ง max-length + overflow=reject-publish-dlx แทนการบังคับ lazy mode
8. HiPE / JIT compile Erlang — 🔴 HiPE ถูก REMOVED จาก Erlang/OTP 24 (2021). RabbitMQ 4.x รันบน OTP 26+ ที่มี BeamAsm JIT enabled by default บน x86_64/arm64 — ไม่ต้องตั้งเอง, ไม่มีอะไรให้เปิด

12.2 ปัญหา Memory High Watermark

RabbitMQ ตั้ง threshold:

vm_memory_high_watermark.relative = 0.6   # 60% ของ RAM (default ใน 3.13/4.x; เดิม 0.4 ใน 3.x ต้น)

เมื่อ memory ใช้เกิน threshold นี้ จะเกิดอะไรขึ้น อธิบายเป็นขั้น ๆ:

• broker ยิง memory alarm (สัญญาณเตือนภายในว่า memory ใกล้เต็ม)
• alarm นี้ทำให้ publisher flow control (กลไก "หรี่" การรับ message ใหม่ เพื่อไม่ให้ memory ยิ่งพุ่ง) ทำงาน — จะ throttle (ชะลอ ไม่ใช่ปิด) channel ที่กำลัง publish และ pause/block ไม่ให้ connection ใหม่เข้ามา
• แต่ consumer + ack ยังทำงานต่อได้ตามปกติ เพื่อช่วยระบาย queue ให้ memory ลดลง

พฤติกรรมนี้มีมานานก่อน 3.13 แล้ว — flow control on memory alarm ไม่ใช่ของใหม่

→ ดู Management UI: queue blocked? → ลด producer, เพิ่ม consumer, เพิ่ม memory

12.3 Disk Free Limit

disk_free_limit.absolute = 5GB

เมื่อ disk เหลือน้อย — block publisher

12.4 Connection Limit

file_handle_cache.connection_limit = 50000

OS file descriptor limit (ulimit -n) ต้องสูงพอ

12.5 Benchmark Tool

docker run --rm -it --net host pivotalrabbitmq/perf-test \
  --uri amqp://admin:admin@localhost \
  --producers 10 --consumers 10 \
  --rate 1000 --time 60 \
  --queue test --auto-delete false

วัด throughput + latency จริง

---

Part 13: Monitoring

13.1 Management UI

http://localhost:15672

ดู:

• Overview — message rate, connection count
• Queues — depth, consumer count, ack rate, redeliver
• Exchanges — publish rate per exchange
• Channels — unacked count, prefetch

13.2 Prometheus Plugin

rabbitmq-plugins enable rabbitmq_prometheus

✅ Bundled + enabled by default ใน RabbitMQ 4.0+ — ไม่ต้อง enable แยกอีกแล้ว. command นี้ใช้เมื่อรัน 3.x ที่ปิดไว้

→ metric ที่ :15692/metrics

ใน Prometheus:

scrape_configs:
  - job_name: rabbitmq
    static_configs:
      - targets: ['rabbit:15692']

13.3 Metric ที่ดู

• rabbitmq_queue_messages_ready — message รอ consume
• rabbitmq_queue_messages_unacknowledged — consumed but not acked
• rabbitmq_queue_messages_published_total
• rabbitmq_queue_messages_delivered_total
• rabbitmq_connections, rabbitmq_channels
• rabbitmq_memory_total
• rabbitmq_disk_free

13.4 Alert

• Queue depth > N (backlog)
• Memory > 80%
• Disk free < 5GB
• Unacked > 1000 (consumer ค้าง)
• Connection drop > threshold
• DLQ has message (เลย count > 0)

---

Part 14: Operational Patterns

14.1 Backup

rabbitmqctl export_definitions /tmp/defs.json
# import:
rabbitmqctl import_definitions /tmp/defs.json

→ backup exchange, queue, binding, policy

Message data → ถ้า durable + cluster = replicate auto

14.2 Shovel / Federation

Move message ระหว่าง broker:

• Shovel — pull from one broker, push to another (1-way)
• Federation — exchange/queue federation (more transparent)

Use case:

• Cross-DC replication
• Migration broker
• Archive

14.3 Policy

ใช้ policy ตั้ง property ทั้งกลุ่ม queue:

# ✅ 2026 (RabbitMQ 4.x) — quorum-friendly policy, ห้ามใช้ queue-mode=lazy หรือชื่อ ha-all อีก
rabbitmqctl set_policy orders-policy "^orders\." \
  '{"max-length":1000000,"overflow":"reject-publish-dlx","message-ttl":86400000}' \
  --apply-to queues

→ queue ขึ้นต้น "orders." ทุกตัว apply

🔴 อย่าใช้ queue-mode=lazy หรือ ha-mode: all ใน policy ของ 4.x — ทั้งสองตัวถูก REMOVED ใน 4.0 (lazy mode + classic mirrored queues หมดอายุไปแล้ว). ใช้ Quorum queue + max-length + overflow=reject-publish-dlx แทน. ชื่อ policy เดิม ha-all ก็ควรเปลี่ยน — มันสะท้อนยุค mirrored queue ที่ผ่านไปแล้ว

14.4 Vhost (Virtual Host)

แบ่ง logical broker:

/prod-billing
/prod-shipping
/staging
/dev

spring.rabbitmq.virtual-host: /prod-billing

→ isolation ระหว่าง app/team

---

Part 15: Anti-Patterns

15.1 1 Connection per Operation

// ❌
public void send() {
    Connection c = factory.newConnection();   // expensive!
    Channel ch = c.createChannel();
    ch.basicPublish(...);
    c.close();
}

→ ใช้ connection pool / Spring CachingConnectionFactory (default)

15.2 Auto-Ack ใน Production

@RabbitListener(ackMode = "NONE")    // ❌

→ message หาย ถ้า consumer crash หลังรับ

15.3 ใช้ Default Exchange

rabbit.convertAndSend("", "queueName", payload);

→ ใช้ direct exchange "" + ไม่ flexible

15.4 ไม่มี DLQ

→ poison message ค้าง

15.5 Hardcoded Queue Name

→ rename ยาก, env mix

15.6 Synchronous RPC Over RabbitMQ

Service A ─request──→ RPC queue
Service A ←reply──── Service B (via reply queue)

→ ทำได้แต่ slow + complex. ใช้ HTTP/gRPC ดีกว่าสำหรับ sync

15.7 ใช้ Classic Mirrored Queue ใน 2026

→ Classic Mirrored Queue ถูก REMOVED ใน RabbitMQ 4.0 — ต้อง migrate ไป Quorum ก่อน upgrade. Classic Queue v2 (CQv2) ยังโอเคสำหรับ dev / single-node / non-HA workload — ไม่ใช่ anti-pattern ถ้าใช้ตรงตามนั้น. anti-pattern ที่แท้จริงคือใช้ Classic Mirror (ha-mode: all) ใน production

• ยังเจอ x-queue-mode=lazy หรือ queue-mode=lazy ใน policy → REMOVED ใน 4.0 ด้วย, ลบทิ้งก่อน upgrade

---

Part 16: Lab

Lab 1: Basic publish + consume

setup RabbitMQ → declare exchange, queue, binding → produce + consume

Lab 2: Publisher confirm + DLQ

ทำ flow: ส่ง → confirm → fail consumer → DLX → DLQ → alert

Lab 3: Quorum cluster

3 node cluster → kill 1 → ยังทำงาน → return

Lab 4: Performance test

perf-test → ลอง persistent vs non-persistent, prefetch ต่าง ๆ

---

Part 16.5: Production War Stories

🔥 บริษัท SaaS (2021): Mirrored queue split-brain ทำ message หาย 1M ฉบับ

Cluster 3 node, ใช้ Classic mirrored queue → network partition → split-brain → resync ผิด → message ที่ใน "minority" partition หาย

Lesson:

• Migrate ทุก queue เป็น Quorum Queue (Raft consensus, partition-safe)
• ตั้ง cluster_partition_handling = pause_minority — minority pod stops กว่าจะ rejoin
• Backup ก่อน upgrade RabbitMQ version

🔥 Slack (2018): consumer prefetch 1000 → 1 slow consumer ทำ partition ติด 30 นาที

📝 หมายเหตุ (เหมือนเคสอื่นในหนังสือเล่มนี้ เช่นบทที่ 08): เคสนี้รวบรวมจาก pattern ที่เห็นในวงการ + รายงานข่าวสั้นปี 2018 — ไม่ใช่ public postmortem ฉบับเต็มจาก Slack โดยตรง. ใช้เป็นบทเรียนเรื่อง prefetch ได้ แต่ตัวเลข "30 นาที" คือประมาณการ ไม่ใช่ตัวเลขทางการ

Default prefetch สูง → 1 consumer ดึง 1000 message แล้ว slow → 1000 message ค้างที่ consumer นั้น → consumer อื่นไม่ได้รับ

Lesson:

• Prefetch ต่ำ (10-100) สำหรับ task ที่ process ช้า
• Prefetch สูง (500-1000) เฉพาะ task ที่ process เร็ว (< 10ms)
• Monitor queue depth + consumer utilization

🔥 บริษัท e-commerce (2023): Memory high watermark = pause publish 6 ชม.

Promo flash sale → consumer process ไม่ทัน → queue โต 4GB → broker hit memory high watermark (40%) → block publisher → web app freeze ทั้งหมด

Lesson:

• Use Lazy queues — Lazy queue mode ถูก REMOVED ใน 4.0. ใช้ CQv2 หรือ Quorum ที่เก็บ disk-first โดย default แทน
• Quorum queue + max-length + overflow=reject-publish-dlx — reject (ส่งไป DLX) ก่อนถึง watermark, กัน publisher freeze
• Monitor rabbitmq_memory_used + alert ที่ 60% (ก่อน watermark)
• Capacity plan: consumer × 10 ใน peak

---

Part 16.6: 📋 Cheat Sheet

Production config baseline

# rabbitmq.conf
cluster_partition_handling = pause_minority
disk_free_limit.absolute = 5GB
vm_memory_high_watermark.relative = 0.6   # ค่า default ปัจจุบัน (3.13+/4.x). ปรับลงถ้า broker มี RAM น้อยกว่า 8 GB หรือเพื่อให้ alert มี buffer ก่อนถึง watermark
heartbeat = 60
channel_max = 200
consumer_timeout = 1800000             # 30 min — equals default ตั้งแต่ 3.12+ (1800000ms = 30 min).
                                       # งานยาวกว่านี้ (เช่น video processing) ควรใช้ heartbeat
                                       # ack pattern แทนการขยาย timeout ต่อไปอีก
default_queue_type = quorum            # ★ ตั้งให้ broker สร้าง Quorum เป็น default
                                       #   เมื่อ client ไม่ระบุ x-queue-type
                                       #   (ไม่งั้นได้ Classic v2)

# queue policy (Quorum, recommended 2026)
rabbitmqctl set_policy orders-policy "^orders\." \
  '{"queue-type":"quorum","max-length":1000000,"overflow":"reject-publish-dlx"}' \
  --apply-to queues

⚠️ queue-type ผ่าน policy ไม่เปลี่ยน type ของ queue ที่มีอยู่แล้ว — queue type เป็น immutable หลังสร้าง. policy มีผลกับ new queue เท่านั้น — queue เก่าต้อง delete + recreate (พร้อม drain message ออกก่อน) หรือใช้ Shovel/Federation ย้าย

Producer config baseline

ConnectionFactory factory = new ConnectionFactory();
factory.setAutomaticRecoveryEnabled(true);
factory.setNetworkRecoveryInterval(5000);
factory.setRequestedHeartbeat(60);
factory.setConnectionTimeout(10_000);
factory.setChannelRpcTimeout(30_000);

// Publisher Confirm (async)
channel.confirmSelect();
channel.addConfirmListener(
    (seq, multi) -> log.debug("ack {}", seq),
    (seq, multi) -> log.error("nack {} — outbox-redeliver", seq));

channel.basicPublish(exchange, routingKey,
    new BasicProperties.Builder()
        .deliveryMode(2)               // persistent
        .messageId(uuid)               // for idempotency
        .timestamp(new Date())
        .contentType("application/json")
        .headers(Map.of("trace_id", traceId))
        .build(),
    bytes);

Consumer config baseline (Spring AMQP)

spring.rabbitmq:
  listener.simple:
    acknowledge-mode: manual
    prefetch: 50                       # tune per workload (10-500)
    concurrency: 5
    max-concurrency: 20
    default-requeue-rejected: false    # send to DLX on reject
    retry:
      enabled: true
      initial-interval: 500ms
      max-attempts: 3
      multiplier: 2

Exchange + queue topology

Queue type selection (2026)

Queue Type	Use case	Notes
Quorum (recommended)	All HA / production queues	Raft, partition-safe, introduced 3.8 GA (Oct 2019), recommended (not auto-default) since 4.0 — ตั้ง default_queue_type=quorum ใน config
Classic Mirrored	❌ REMOVED in 4.0	Migrate to Quorum
Classic v1	❌ REMOVED in 4.0	Migrate to Classic v2 (in 3.13) before upgrading
Stream	Replay, log-like, high throughput	Kafka-like API, since 3.9; Super-streams since 3.11
Classic v2 (CQv2)	Single-node, dev/non-HA	Disk-first by default, no replication

Anti-pattern → Fix

❌	✅
Classic mirrored queue	Quorum queue (Raft)
Auto-ack consumer	Manual ack after side effect
No publisher confirm	confirmSelect + listener
Prefetch 1000 + slow process	Prefetch 10-100
1 connection per request	Connection pool (1 conn, N channel)
No DLX configured	DLX + retry topic + monitoring
default-requeue-rejected: true + poison message	false + DLX
Single broker (no HA)	3-node cluster + pause_minority

---

Part 17: Checkpoint

1. ทำไม RabbitMQ ใช้ "smart broker, dumb consumer"?
2. ความต่าง Connection vs Channel?
3. Direct vs Topic vs Fanout exchange?
4. Publisher Confirm + Returns แก้ปัญหาอะไร?
5. Prefetch สำคัญยังไง?
6. Classic vs Quorum vs Stream queue?
7. DLX + retry pattern ทำงานยังไง?
8. Quorum queue ต้องมีกี่ node?
9. Memory high watermark = อะไร?
10. RabbitMQ vs Kafka — เลือกอันไหนสำหรับ task queue?

---

Part 18: สรุปบทนี้

• RabbitMQ = smart broker — flexible routing via exchange + queue + binding
• Connection per app, Channel per thread, Channel pool
• Exchange types: direct, topic, fanout, headers (+ custom plugins)
• Publisher Confirm + Returns + Outbox = guarantee delivery
• Manual ack + prefetch = control consumer
• Quorum queue = 2026 default (Raft replication)
• Stream queue = Kafka-like replay
• DLX + retry pattern = handle poison message
• Cluster ≥ 3 node + pause_minority + load balancer
• Tuning: persistent trade-off, lazy queue, prefetch, concurrent
• Monitor: Management UI + Prometheus + alert queue depth/memory/disk

บทถัดไป — Kafka Deep — บทใหญ่ที่สุดของหนังสือ (~2000 บรรทัด)

---

← บทที่ 06 | สารบัญ | บทที่ 08: Kafka Deep →