Acknowledgements

คำสัญญา: อย่าทำงานหาย

คำถามหลักที่ queue ต้องตอบให้ได้คือ: ถ้า consumer หยิบ message ไปแล้ว crash ก่อนทำเสร็จ งานนั้นจะเป็นอย่างไร? acknowledgement คือคำตอบ และเป็นกลไก reliability ที่สำคัญที่สุดใน RabbitMQ

acknowledgement (ack) คือการที่ consumer บอก broker ว่า: “ฉันประมวลผล message นี้เสร็จสมบูรณ์แล้ว — ลบได้เลย” จนกว่า ack จะมาถึง RabbitMQ จะถือ message นั้นเป็น unacknowledged และจะ redeliver ให้ใหม่ถ้า channel ของ consumer ตาย

manual vs automatic ack

มีสอง mode และความต่างมหาศาล:

automatic ack (autoAck: true / auto_ack=True): message ถือว่า deliver แล้ว ทันทีที่ออกจาก broker ก่อนที่ code ของคุณจะได้แตะ เร็วดี แต่ถ้า consumer crash กลางคัน message นั้นก็ หายไป — RabbitMQ ลืม message นั้นไปแล้ว นี่คือ at-most-once และทำงานหายเงียบ ๆ
manual ack: RabbitMQ deliver message แต่ยังเก็บไว้เป็น unacknowledged จนกว่าคุณจะ ack อย่างชัดเจน ถ้า channel ปิดโดยไม่มี ack message จะถูก redeliver ไปยัง consumer ตัวอื่น นี่คือ at-least-once และเป็นสิ่งที่คุณต้องการสำหรับงานที่สำคัญ

กฎคือ: ใช้ manual ack กับทุกอย่างที่คุณเสียไปไม่ได้

ack, nack, reject

ด้วย manual ack คุณมีสามคำตอบ:

ack — เสร็จ ลบได้
nack (หรือ reject) พร้อม requeue: true — ประมวลผลไม่ได้; เอากลับเข้า queue เพื่อลองใหม่
nack/reject พร้อม requeue: false — ประมวลผลไม่ได้และลองใหม่ก็ไม่ช่วย; ทิ้งไปเลย (หรือส่งไป dead-letter exchange ถ้าตั้งค่าไว้)

redelivered flag บน delivery บอกคุณว่า “คุณเคยเห็นตัวนี้มาแล้ว” — เป็นสัญญาณว่าความพยายามครั้งก่อน fail คุณจึงจัดการอย่างระวังขึ้นได้

lifecycle

sequenceDiagram
  participant Q as Queue
  participant C as Consumer
  Q->>C: deliver (unacked)
  alt ประมวลผลสำเร็จ
    C->>C: ทำงาน
    C->>Q: ack
    Note over Q: message ถูกลบ
  else consumer crash
    C--xC: crash ก่อน ack
    Note over Q: ไม่ได้รับ ack
    Q->>C: redeliver (redelivered=true)
  end

ack ตอนสำเร็จ เทียบกับ redelivery ตอน crash

manual ack ใน code

// noAck: false → manual acknowledgement
channel.consume('orders', async (msg) => {
  if (!msg) return;
  try {
    await handle(msg.content);
    channel.ack(msg);                 // done
  } catch (err) {
    channel.nack(msg, false, true);   // requeue for another try
  }
}, { noAck: false });

def on_message(ch, method, properties, body):
    try:
        handle(body)
        ch.basic_ack(delivery_tag=method.delivery_tag)          # done
    except Exception:
        ch.basic_nack(delivery_tag=method.delivery_tag,
                      requeue=True)                             # retry

# auto_ack=False → manual acknowledgement
channel.basic_consume(queue="orders", on_message_callback=on_message, auto_ack=False)

// fourth arg autoAck=false → manual acknowledgement
msgs, _ := ch.Consume(q.Name, "", false, false, false, false, nil)
for d := range msgs {
    if err := handle(d.Body); err != nil {
        d.Nack(false, true) // requeue for another try
        continue
    }
    d.Ack(false) // done
}