RabbitMQ: consumer ACK, prefetch và publisher confirm

Ở bài 8.1 ta đã biết: broker giao at-least-once, consumer phải idempotent. Bài này đi vào cơ chế làm nên "at-least-once": ACK ở phía consumer và confirm ở phía producer.

1. Bài toán: không được mất OrderPlaced

Trong shop-app có hai chỗ có thể mất message:

order-service ──(1)──> RabbitMQ ──(2)──> inventory-service
   publish                                  xử lý
(1) Broker có thật sự nhận chưa? Network rớt giữa chừng thì sao?
(2) Consumer nhận rồi chết giữa chừng thì message đi đâu?

2. Vấn đề

• rabbitTemplate.convertAndSend() mặc định là "bắn rồi quên": không lỗi không có nghĩa broker đã lưu.
• Routing key sai (typo order.placd) → exchange không tìm được queue → message bị bỏ im lặng.
• Nếu broker coi message là "xong" ngay khi giao cho consumer, consumer chết là mất.
• Một message lỗi bị requeue liên tục sẽ quay vòng mãi, đốt CPU.

3. Giải pháp

Phía consumer: acknowledgement.

Spring AMQP có 3 chế độ acknowledge-mode:

• AUTO (mặc định trong Spring): listener return bình thường → container ACK; listener ném exception → container NACK và mặc định requeue.
• MANUAL: code tự gọi channel.basicAck.
• NONE: broker coi như ACK ngay khi gửi (autoAck của RabbitMQ) → at-most-once, có thể mất.

Lưu ý thuật ngữ: "auto ack" của RabbitMQ client = NONE của Spring, khác AUTO của Spring.

Prefetch (QoS): số message tối đa đã giao nhưng chưa ACK cho mỗi consumer. Spring Boot mặc định 250.

Phía producer: publisher confirm và return.

• Confirm: broker báo ack khi đã nhận và xử lý message (với queue durable + message persistent: đã ghi xuống đĩa; với quorum queue: đã được đa số node ghi). nack khi broker lỗi nội bộ.
• Return (kèm mandatory=true): broker trả message lại khi không route được vào queue nào.

4. Code từng bước

Bước 1: Dependency: spring-boot-starter-amqp.

Bước 2: Config

spring:
  rabbitmq:
    publisher-confirm-type: correlated   # bật confirm
    publisher-returns: true              # bật return
    template:
      mandatory: true                    # unroutable → return về producer
    listener:
      simple:
        acknowledge-mode: auto           # hoặc manual
        prefetch: 10
        concurrency: 2
        max-concurrency: 5
        default-requeue-rejected: false  # exception → không requeue vô hạn

Bước 3a: Producer với confirm (order-service)

@Configuration
public class RabbitTemplateConfig {

    @Bean
    RabbitTemplateCustomizer callbacks() {
        return template -> template.setReturnsCallback(returned ->
            log.error("Unroutable message: exchange={}, key={}, reply={}",
                returned.getExchange(), returned.getRoutingKey(), returned.getReplyText()));
    }
}

public void publish(OrderPlaced event) {
    CorrelationData cd = new CorrelationData(event.eventId().toString());
    rabbitTemplate.convertAndSend("shop.events", "order.placed", event, cd);
    try {
        CorrelationData.Confirm confirm = cd.getFuture().get(5, TimeUnit.SECONDS);
        if (!confirm.isAck()) {
            throw new PublishFailedException("Broker nack: " + confirm.getReason());
        }
    } catch (TimeoutException | ExecutionException e) {
        throw new PublishFailedException("No confirm", e);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new PublishFailedException("Interrupted", e);
    }
}

Chờ confirm từng message thì chậm. Thực tế thường gửi nhiều message rồi xử lý confirm bất đồng bộ, hoặc dùng Outbox: chỉ đánh dấu bản ghi outbox là "đã gửi" khi nhận confirm.

Bước 3b: Consumer manual ACK (inventory-service)

@RabbitListener(queues = "inventory.order-placed", ackMode = "MANUAL")
public void onOrderPlaced(OrderPlaced event, Channel channel,
                          @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
    try {
        inventoryService.reserveIdempotent(event);   // bài 8.1
        channel.basicAck(tag, false);
    } catch (TransientDataAccessException e) {
        channel.basicNack(tag, false, true);         // lỗi tạm thời → thử lại
    } catch (Exception e) {
        channel.basicNack(tag, false, false);        // lỗi dữ liệu → bỏ / sang DLX
    }
}

Với AUTO mode, cùng ý tưởng: ném AmqpRejectAndDontRequeueException để reject không requeue.

Bước 4: Test

# 1. Thử unroutable: đổi routing key thành "order.placd"
curl -X POST http://localhost:8081/orders -d '...' -H "Content-Type: application/json"
# log: Unroutable message: exchange=shop.events, key=order.placd, reply=NO_ROUTE

# 2. Thử consumer chết: đặt breakpoint/sleep trong listener, đặt đơn, rồi kill -9 inventory-service
# Management UI: queue inventory.order-placed: Unacked=1 → sau khi kill: Ready=1
# Khởi động lại → message được giao lại, header redelivered=true

5. Cách hoạt động

Producer                     Broker                         Consumer
   │ publish (seq=7)           │                               │
   │──────────────────────────>│ route → queue, ghi đĩa        │
   │<──────── confirm ack 7 ───│                               │
   │                           │── deliver (tag=1) ───────────>│ Unacked=1
   │                           │                               │ xử lý...
   │                           │<──────────── basicAck(1) ─────│
   │                           │ xóa message                   │

Consumer chết khi còn Unacked:
   channel/connection đóng → broker đưa message về Ready, redelivered=true
   → giao cho consumer khác

Không có timeout xử lý kiểu "30 giây không ACK thì giao lại". Broker chỉ giao lại khi channel đóng. Nếu consumer treo mà connection còn sống, message nằm Unacked mãi (RabbitMQ có consumer_timeout, mặc định 30 phút, quá hạn sẽ đóng channel).

Prefetch quyết định độ "tham" của consumer:

prefetch=1    consumer giữ 1 message, chia đều nhưng throughput thấp (mỗi message chờ 1 round-trip)
prefetch=250  throughput cao, nhưng 1 consumer chậm ôm 250 message, instance khác rảnh
              consumer chết → 250 message cùng được giao lại

6. So sánh

7. Khi nào nên dùng / không nên dùng

• Nên bật confirm + mandatory cho event nghiệp vụ (OrderPlaced, PaymentSucceeded).
• AUTO ack đủ cho đa số listener; dùng MANUAL khi cần phân biệt lỗi tạm thời / lỗi dữ liệu hoặc ACK sau xử lý bất đồng bộ.
• Không nên dùng NONE (autoAck) cho dữ liệu quan trọng. Chấp nhận được với log/metric không cần chính xác.
• Prefetch: việc nhanh, đều → 50–300; việc chậm (gửi email qua provider) → 1–10.

Đúng, đó là bẫy kinh điển: requeue loop. Message được đưa lại gần đầu queue và giao lại ngay, không có độ trễ:

deliver msg#42 → exception → nack(requeue) → deliver msg#42 → exception → ...
                 hàng nghìn lần mỗi giây, log tràn, CPU cao

• Với classic queue: vòng lặp vô hạn. Nếu prefetch=1 và một consumer, các message phía sau gần như bị chặn.
• Với quorum queue: có delivery-limit (RabbitMQ 4.x mặc định 20). Vượt giới hạn thì message bị bỏ hoặc sang DLX.

Cách xử lý:

1. default-requeue-rejected: false để exception không requeue.
2. Bật retry có backoff trong Spring (giới hạn số lần) rồi đẩy sang DLQ.
3. Phân loại lỗi: lỗi dữ liệu (JSON hỏng, SKU không tồn tại) không bao giờ requeue.

Retry, DLX, poison message học chi tiết ở bài 8.5.

Chưa. Bạn cần tách hai đoạn: confirm chỉ nói broker đã nhận trách nhiệm với message.

confirm ack   = broker đã nhận (và đã ghi đĩa / được đa số node ghi nếu cần)
             ≠ đã vào queue          (unroutable vẫn ack → cần mandatory + return)
             ≠ consumer đã xử lý     (cần consumer ACK + idempotency)

Bức tranh đầy đủ để không mất message trong shop-app:

order-service ── Outbox + confirm + mandatory ──> RabbitMQ (durable / quorum queue)
                                                   ──> inventory (ACK sau xử lý + idempotent)

Còn producer cần biết consumer xử lý xong chưa thì phải dùng một event ngược lại (ví dụ inventory.reserved), không phải confirm.

Q1. Consumer ACK hoạt động thế nào? Consumer chết thì sao?

🇻🇳 Tiếng Việt:

Broker giao message, message ở trạng thái unacked. Consumer xử lý xong gửi ack thì broker xóa. Nếu consumer chết, channel đóng, broker đưa các message unacked về lại queue với cờ redelivered và giao cho consumer khác. Vì vậy là at-least-once, consumer phải idempotent. Trong Spring, ack mode AUTO sẽ ack khi listener return và nack khi ném exception.

🇬🇧 English:

The broker delivers a message and keeps it unacked. When the consumer finishes, it acks and the broker deletes it. If the consumer dies, its channel closes and the broker requeues the unacked messages with the redelivered flag, so another consumer gets them. That is at-least-once delivery, so consumers must be idempotent. In Spring's AUTO mode the container acks when the listener returns and nacks when it throws.

Q2. Prefetch là gì, chọn giá trị thế nào?

🇻🇳 Tiếng Việt:

Prefetch là số message tối đa đã giao mà chưa ack cho mỗi consumer. Cao thì throughput tốt nhưng một consumer chậm ôm nhiều message và phân phối không đều. Thấp thì chia đều nhưng chậm. Việc nhanh em để vài chục đến vài trăm, việc chậm như gọi email provider em để 1 đến 10.

🇬🇧 English:

Prefetch is the maximum number of unacked messages delivered to a consumer. A high value gives better throughput, but a slow consumer hoards messages and the load becomes uneven. A low value spreads work evenly but is slower. For fast work I use tens to hundreds; for slow work like calling an email provider I use 1 to 10.

Q3. Làm sao producer biết message không bị mất?

🇻🇳 Tiếng Việt:

Bật publisher confirm: broker ack khi đã nhận và lưu message. Confirm không báo unroutable, nên bật thêm mandatory để broker return message không vào được queue nào. Kết hợp queue durable hoặc quorum, message persistent, và Outbox pattern để không mất event khi app chết sau khi commit DB.

🇬🇧 English:

Enable publisher confirms: the broker acks once it has accepted and stored the message. Confirms do not report unroutable messages, so I also set mandatory so the broker returns messages that match no queue. Together with durable or quorum queues, persistent messages and the Outbox pattern, events are not lost even if the app crashes after committing to the database.

Cách nhớ ngắn gọn

• Consumer: ACK sau khi xử lý → at-least-once → idempotent.
• Consumer chết → unacked quay về queue, redelivered=true.
• Producer: confirm = broker đã nhận; mandatory + return = routing sai.
• Requeue vô điều kiện = vòng lặp; giới hạn retry rồi DLQ.