Phần trước kết thúc bằng một quy tắc: đừng dùng requeue=true cho lỗi xử lý, vì nó quay 5 834 vòng mỗi giây và chặn đứng hàng đợi. Nhưng nhiều lỗi là tạm thời — dịch vụ hạ nguồn đang sập, CSDL đang bận — và thử lại sau vài giây là đúng. Bài này dựng cơ chế đó bằng hai thứ đã có: TTL và dead letter exchange.

Vòng thử lại

Hàng đợi việc và hàng đợi chờ dead-letter qua lại nhau

ch.queueDeclare("viec", true, false, false, Map.of("x-dead-letter-exchange", "x-cho"));
ch.queueDeclare("cho",  true, false, false, Map.of("x-dead-letter-exchange", "x-viec",
                                                   "x-message-ttl", 1000));

Consumer gặp lỗi tạm thời thì basicNack(requeue=false) — thông điệp rời hàng đợi việc ngay lập tức, không chặn ai. Nó nằm trong hàng đợi chờ đúng một giây rồi tự quay về. Đo bằng đồng hồ thật:

 0.00 s  lần giao 1  |  x-death: (chưa có)
 1.01 s  lần giao 2  |  x-death: cho/expired=1 viec/rejected=1
 2.03 s  lần giao 3  |  x-death: cho/expired=2 viec/rejected=2
 3.04 s  lần giao 4  |  x-death: cho/expired=3 viec/rejected=3

Mỗi vòng cách nhau đúng một giây, sai số 30 mili giây. Và consumer rảnh hoàn toàn trong khoảng chờ đó — đó là khác biệt cốt lõi so với Thread.sleep trong listener, thứ giữ luôn cả luồng.

Vì sao vòng này không bị xoá

Phần 14 đo được một điều đáng sợ: thông điệp quay lại một hàng đợi mà nó đã chết ở đó, với mọi lần chết đều vì hết hạn, sẽ bị broker xoá âm thầm. Vòng thử lại này đi qua đúng hình dạng đó, nên tôi kiểm tra trước khi khuyên ai dùng.

Nó an toàn, và bảng x-death ở trên nói rõ lý do: hai mục, hai nguyên nhân khác nhau — viec/rejected và cho/expired. Cơ chế chống vòng lặp chỉ kích hoạt khi mọi lý do đều là expired.

Hệ quả: đừng thay bước basicNack bằng một TTL nữa cho "gọn". Một vòng gồm hai hàng đợi cùng đá nhau bằng TTL sẽ mất thông điệp, và không có log nào cho bạn biết.

Bẫy: một hàng đợi chờ cho mọi mức độ trễ

Cách làm thang backoff trực giác nhất là đặt TTL trên từng thông điệp rồi đẩy tất cả vào chung một hàng đợi chờ. Gửi hai cái, một hẹn 5 giây, một hẹn 0,5 giây:

 5.00 s  ra khỏi hàng đợi chờ: CHAM-5s
 5.00 s  ra khỏi hàng đợi chờ: NHANH-0,5s

Cái hẹn nửa giây ra sau năm giây — muộn gấp mười lần.

Hàng đợi là hàng đợi: broker chỉ xét hạn của thông điệp ở đầu hàng. Cái 5 giây đứng trước, nên cái 0,5 giây phía sau không được nhìn tới cho tới khi cái trước đi. Đây là head-of-line blocking, và nó không hiện ra ở bất kỳ chỉ số nào — hàng đợi trông hoàn toàn khoẻ mạnh.

Trong thang backoff thật, thông điệp ở lần thử thứ năm (chờ lâu) sẽ chặn thông điệp vừa lỗi lần đầu (đáng lẽ chờ ngắn). Nghĩa là mọi độ trễ bị kéo lên bằng độ trễ dài nhất đang có trong hàng.

Cách đúng: mỗi mức độ trễ một hàng đợi

Đặt TTL trên hàng đợi, không trên thông điệp, và có bao nhiêu mức thì bấy nhiêu hàng đợi:

ch.queueDeclare("cho-500",  true, false, false,
        Map.of("x-dead-letter-exchange", "x-viec", "x-message-ttl", 500));
ch.queueDeclare("cho-5000", true, false, false,
        Map.of("x-dead-letter-exchange", "x-viec", "x-message-ttl", 5000));

 0.50 s  ra khỏi hàng đợi chờ: NHANH-0,5s
 5.00 s  ra khỏi hàng đợi chờ: CHAM-5s

Mỗi hàng đợi chỉ chứa thông điệp cùng hạn, nên thứ tự vào cũng là thứ tự hết hạn và không ai chặn ai. Cái giá là vài hàng đợi nữa — mà phần 15 đã đo: khoảng 65 KB mỗi cái, không đáng kể.

Thang backoff và điểm dừng

Ghép lại thành cơ chế hoàn chỉnh: consumer đọc x-death để biết đây là lần thử thứ mấy, rồi chọn hàng đợi chờ tương ứng.

long lan = soLanDaThu(props);           // đọc count trong x-death
if (lan >= 5) {
    ch.basicPublish("", "nghia-dia", props, body);   // bỏ cuộc
    ch.basicAck(tag, false);
} else {
    ch.basicPublish("", "cho-" + THANG[(int) lan], props, body);
    ch.basicAck(tag, false);
}

Ba điều đáng chú ý trong đoạn trên:

Đẩy đi rồi basicAck, không basicNack. Tự chọn hàng đợi chờ nghĩa là tự publish; nack sẽ đưa nó vào DLX cố định của hàng đợi việc, không chọn được mức trễ.

Phải có điểm dừng. Phần 20 cho thấy x-delivery-limit không đếm nack, nên không có lưới an toàn tự động ở đây — bộ đếm duy nhất là x-death.count, và bạn phải tự đọc nó.

Thang nên nhân lên, đừng cộng. 1 s, 5 s, 30 s, 2 phút, 10 phút. Lỗi tạm thời thường tự hết trong vài giây; lỗi không tự hết thì thử lại dày chỉ làm dịch vụ hạ nguồn đang ốm thêm nặng.

Còn plugin delayed message thì sao

RabbitMQ có plugin rabbitmq_delayed_message_exchange làm việc này bằng một dòng. Nó tiện, nhưng đổi lại: thông điệp đang chờ nằm trong một kho riêng của plugin chứ không phải trong hàng đợi, nên bạn không nhìn thấy chúng bằng list_queues, không đếm được, và không dựng cảnh báo theo cách quen thuộc. Cách TTL + DLX ở trên chỉ dùng những thứ có sẵn trong lõi, và mọi thông điệp đang chờ đều hiện ra ở đúng chỗ bạn quen nhìn.

Bài sau: nhận lại nhiều lần và tính lũy đẳng — vì sao mọi cơ chế thử lại trong bài này đều bắt buộc consumer phải chịu được việc xử lý cùng một thông điệp hai lần.

Thử ba mươi giây

# hang doi cho cua ban co dang bi tac dau hang khong
docker exec -u rabbitmq rmq rabbitmqctl -q list_queues name messages message_bytes \
  | grep '^cho'

Hàng đợi chờ đúng ra phải vơi đều theo nhịp TTL. Nếu số thông điệp trong đó chỉ tăng, bạn đang có một thông điệp hạn dài nằm chặn ở đầu — hoặc đang dùng chung một hàng đợi cho nhiều mức độ trễ.