Độ tin cậy — không mất message

Hai thứ phải đi cùng nhau mới không mất message khi broker khởi động lại:

Durable queue — khai báo queue với durable=True để định nghĩa queue được lưu vào đĩa, không biến mất khi broker restart. Persistent message — đặt delivery_mode=2 để message được ghi xuống đĩa thay vì chỉ nằm trong RAM.

# queue bền
ch.queue_declare(queue="task_queue", durable=True)

# message bền (delivery_mode=2 = persistent)
ch.basic_publish(
    exchange="",
    routing_key="task_queue",
    body="work item",
    properties=pika.BasicProperties(delivery_mode=2),
)

Publish thành công không có nghĩa broker đã lưu message. Bật publisher confirms để broker gửi ack lại cho producer khi đã nhận và ghi an toàn — nếu không, producer biết để gửi lại.

ch.confirm_delivery()  # bật chế độ confirm trên channel

try:
    ch.basic_publish(exchange="", routing_key="task_queue",
                     body="work",
                     properties=pika.BasicProperties(delivery_mode=2))
    print("broker đã xác nhận nhận message")
except pika.exceptions.UnroutableError:
    print("message không tới được queue nào -> xử lý lại")

Phía consumer, dùng manual ack để chỉ xóa message khỏi queue khi đã xử lý xong. Nếu consumer chết trước khi ack, broker tự động giao lại message cho consumer khác.

Prefetch / QoS: mặc định broker đẩy hết message tới consumer rảnh nhất, gây mất cân bằng. Đặt basic_qos(prefetch_count=1) để mỗi consumer chỉ giữ tối đa 1 message chưa ack — chỉ nhận message mới sau khi ack cái cũ. Nhờ vậy tải chia đều theo tốc độ xử lý thực tế.

import pika

conn = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
ch = conn.channel()
ch.queue_declare(queue="task_queue", durable=True)

ch.basic_qos(prefetch_count=1)  # công bằng: mỗi lần 1 message

def callback(channel, method, props, body):
    try:
        print("đang xử lý:", body.decode())
        do_work(body)
        channel.basic_ack(delivery_tag=method.delivery_tag)
    except Exception:
        # lỗi -> không requeue, đẩy sang DLX
        channel.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

ch.basic_consume(queue="task_queue", on_message_callback=callback)
ch.start_consuming()

Dead Letter Exchange (DLX) là exchange nhận các message "chết" — bị nack/reject với requeue=False, hết TTL, hoặc queue đầy. Khai báo trên queue gốc qua x-dead-letter-exchange (và tùy chọn x-dead-letter-routing-key). Message chết sẽ được chuyển sang DLX để ghi log, cảnh báo hoặc xử lý lại.

TTL giới hạn thời gian sống của message: đặt cho từng message (expiration) hoặc cho cả queue (x-message-ttl). Message hết TTL mà chưa được tiêu thụ sẽ bị loại — và nếu queue có DLX thì nó chạy thẳng vào DLX.

# queue gốc: hết TTL hoặc bị nack -> đẩy sang DLX
ch.exchange_declare(exchange="dlx", exchange_type="direct", durable=True)
ch.queue_declare(queue="task_queue", durable=True, arguments={
    "x-dead-letter-exchange": "dlx",
    "x-dead-letter-routing-key": "failed",
    "x-message-ttl": 60000,   # 60s
})

# queue chứa message chết để xem xét / retry
ch.queue_declare(queue="task_failed", durable=True)
ch.queue_bind(queue="task_failed", exchange="dlx", routing_key="failed")

Mẫu retry trễ kinh điển: message lỗi không requeue ngay (sẽ lặp vô hạn nóng) mà đẩy vào một queue chờ có TTL. Hết TTL, message dead-letter ngược lại queue gốc để thử lại sau một khoảng nghỉ.

retry_queue có x-message-ttl=30000 và x-dead-letter-exchange trỏ ngược về exchange của work_queue. Sau mỗi vòng, kiểm tra số lần thử (đếm qua header x-death); vượt ngưỡng thì đẩy hẳn sang dead_queue để xử lý thủ công.