Publisher Confirms

publish ธรรมดาคือ fire-and-forget จริง ๆ

มีความจริงที่น่าอึดอัดเกี่ยวกับ publish พื้นฐาน: โดย default ไม่ return อะไรกลับมา client เขียน message ลง socket แล้วไปต่อ ถ้า broker โหลดหนัก, connection หลุดตอนนั้นพอดี หรือ message ถูก reject — คุณจะไม่มีวันรู้ message หายไปเฉย ๆ แต่ code คิดว่าสำเร็จ

สำหรับ analytics event สัก event อาจไม่เป็นไร แต่สำหรับ “ลูกค้าจ่ายเงินให้เราแล้ว” แบบนั้นรับไม่ได้เลย publisher confirm ปิดช่องโหว่นี้

Publisher confirm: broker ack การ publish ของคุณ

เมื่อคุณตั้ง channel เป็น confirm mode broker จะส่ง acknowledgement กลับมาสำหรับทุก message เมื่อ broker รับผิดชอบ message นั้นแล้ว (route ไปยัง durable queue ที่ match ทั้งหมด และ — สำหรับ persistent message — เขียนลง disk แล้ว) คุณจะได้ ack (broker มีแล้ว) หรือในกรณีที่หายากคือ nack (broker รับไม่ได้)

sequenceDiagram
  participant P as Publisher
  participant B as Broker
  P->>B: publish (confirm mode)
  B->>B: route + persist
  B-->>P: ack (I have it)
  Note over P: safe to consider it sent
  P->>B: publish (bad case)
  B-->>P: nack (could not accept)
  Note over P: retry or alert

publish ที่เปิด confirm

confirm เป็นแบบ asynchronous: broker ทยอยส่ง ack กลับมาระหว่างที่ process message ดังนั้นคุณไม่ต้อง block หลังทุก publish publisher ที่ throughput สูงจะเก็บ window ของ message ที่ยังไม่ confirm ไว้ในระหว่างทาง และถือว่า message durable ก็ต่อเมื่อ ack ของตัวเองมาถึงแล้วเท่านั้น

// amqplib: a confirm channel gives per-message callbacks
const channel = await conn.createConfirmChannel();

channel.publish('orders', 'order.created', Buffer.from(body), { persistent: true },
  (err) => {
    if (err) console.error('NACK — message not confirmed, retry:', err);
    else console.log('ACK — broker has the message');
  });

# pika: enable confirms; publish raises/returns on failure
channel.confirm_delivery()

try:
    channel.basic_publish(
        exchange="orders", routing_key="order.created", body=body,
        properties=pika.BasicProperties(delivery_mode=2),  # persistent
        mandatory=True,
    )
    print("ACK — broker has the message")
except pika.exceptions.UnroutableError:
    print("returned — no queue matched")
except pika.exceptions.NackError:
    print("NACK — broker could not accept, retry")

ch.Confirm(false) // put channel in confirm mode
confirms := ch.NotifyPublish(make(chan amqp.Confirmation, 1))

ch.PublishWithContext(ctx, "orders", "order.created", true, false, amqp.Publishing{
    DeliveryMode: amqp.Persistent,
    Body:         body,
})

if c := <-confirms; c.Ack {
    log.Println("ACK — broker has the message")
} else {
    log.Println("NACK — broker could not accept, retry")
}

Mandatory flag: จับ message ที่ route ไม่ได้

confirm บอกว่า broker ได้ message แล้ว — แต่ message ที่ route ไป ศูนย์ queue ก็ยังถือว่า broker จัดการสำเร็จ message แค่หายไปเฉย ๆ ถ้าคุณ publish order.created แต่ไม่มี queue ไหน bind ไว้รับ (พิมพ์ผิด หรือลืม binding) message ก็ถูก drop และคุณจะไม่มีวันรู้

mandatory flag แก้เรื่องนี้: ถ้า mandatory message route ไปไม่เจอ queue ไหนเลย broker จะ return message กลับไปหา publisher แทนที่จะ drop ทิ้ง ฟัง returned message แล้วคุณจะจับ misrouting ได้ทันที

รวมกันแล้ว confirm + mandatory ตอบทั้งสองคำถาม: broker รับไหม? และ message ถึง queue จริงไหม?

Trade-off: ความปลอดภัยแลกมาด้วย throughput

confirm ไม่ฟรี การรอ ack หลังทุก publish จะทำให้ publisher ทำงานแบบ serial และ throughput ตกฮวบ ทางสายกลางที่ดี:

Async confirm พร้อม window — เก็บ N message ไว้ในระหว่างทาง จัดการ ack ตอนที่ทยอยกลับมา throughput เกือบเต็ม ปลอดภัยเต็ม
Batch confirm — publish เป็น batch แล้วรอทั้ง batch confirm ง่ายกว่า throughput ต่ำลงนิดหน่อย
Publish-and-wait ต่อ message — เข้าใจง่ายสุด ช้าสุด ใช้เฉพาะ publish ที่ critical และ volume ต่ำ