kiến trúc RabbitMQ

Trong Event-Driven Architecture và Microservices, Message Broker giống như hệ thần kinh của cả hệ thống. Nó gánh mấy đợt tải tăng đột ngột (Load Leveling / Peak Shaving), tách các service ra khỏi nhau (Decoupling), và giữ cho không giao dịch nào bị thất lạc khi có service sập.

RabbitMQ chạy trên máy ảo Erlang OTP (BEAM), vốn làm ra cho hệ thống viễn thông của Ericsson (con số "chín số 9" hay được nhắc là Ericsson công bố cho tổng đài AXD301 viết bằng Erlang). Theo trang chủ RabbitMQ, đây là message broker mã nguồn mở được triển khai rộng rãi nhất, và là cài đặt AMQP 0-9-1 phổ biến nhất.

Có điều ở hệ thống lớn, cứ cấu hình RabbitMQ theo mấy tutorial cơ bản là sớm muộn cũng dính sự cố:

• App hết socket descriptor vì thread nào cũng tự mở một kết nối TCP riêng, không dùng Channel Multiplexing.
• Mất dữ liệu khi node broker sập vì vẫn dùng Classic Mirrored Queues (đã bị bỏ trong RabbitMQ 4.0) thay vì Quorum Queues.
• Consumer crash vì tràn RAM do để prefetchCount = 0 (Unbounded). Spring AMQP mặc định đã đặt prefetch 250, nhưng client Java thuần hay thư viện khác thì không chắc.
• Một message lỗi (Poison Pill Message) làm tắc cả hàng đợi vì không có Dead Letter Exchange (DLX) và Exponential Backoff.

Phần dưới đi lần lượt từ tầng giao thức AMQP lên tới cách dựng RabbitMQ chịu lỗi tốt cho production.

# kiến trúc amqp 0-9-1: connection vs channel multiplexing

Lỗi cơ bản mà hậu quả lớn: mỗi worker thread mở một kết nối TCP riêng.

• Mỗi TCP Connection tốn một lần bắt tay 3 bước (3-way handshake), tốn mã hoá TLS, và chiếm một file descriptor của Linux.
• Có 500 Virtual Threads hoặc 200 HTTP workers cùng mở TCP connection tới RabbitMQ là broker hết tài nguyên mạng rất nhanh rồi bắt đầu từ chối kết nối.

AMQP 0-9-1 giải bài này bằng Channel Multiplexing (ghép nhiều kênh trên một kết nối):

# nguyên tắc cần nhớ

1. Mỗi app chỉ nên giữ một (hoặc vài) TCP Connection tới RabbitMQ Broker.
2. Từng thread sẽ mở AMQP Channel, một kênh ảo rất nhẹ nằm bên trong kết nối TCP đó.
3. Channel không thread-safe. Mỗi thread phải có Channel riêng (RabbitTemplate của Spring tự lo chuyện này qua pool).

# 4 loại Exchange & Consistent Hash Exchange

Message từ Producer không bao giờ đi thẳng vào Queue. Nó luôn phải qua một Exchange, Exchange dựa vào Routing Key để chuyển tiếp:

# wildcard trong topic exchange

• * (Star): khớp đúng 1 từ (ví dụ order.*.created khớp order.vn.created, nhưng không khớp order.vn.retail.created).
• # (Hash): khớp 0 hoặc nhiều từ (ví dụ audit.# khớp mọi routing key bắt đầu bằng audit.).

# scale ngang: consistent hash exchange

Khi cần chia một lượng dữ liệu cực lớn (ví dụ 100,000 event/giây) ra 10 Queue, mà vẫn phải đảm bảo message của cùng một user_id luôn rơi vào cùng một Queue để giữ đúng thứ tự:

• Consistent Hash Exchange dùng consistent hashing trên routing key để chọn queue, nhờ vậy scale-out consumer theo chiều ngang mà không lo race condition.

Loại exchange này không nằm trong 4 loại chuẩn, nó là plugin rabbitmq_consistent_hash_exchange có sẵn trong bản cài nhưng phải bật (rabbitmq-plugins enable rabbitmq_consistent_hash_exchange). Và hash chỉ đảm bảo cùng user_id vào cùng queue. Muốn giữ thứ tự thật thì mỗi queue chỉ được một consumer xử lý tại một thời điểm (bật x-single-active-consumer, concurrency 1). Mấy consumer cùng đọc một queue là thứ tự lại vỡ.

# quorum queues (raft consensus)

Hơn chục năm nay, High Availability của RabbitMQ dựa vào Mirrored Queues (Ha-mode). Nhưng:

• Mirrored Queues không dùng thuật toán đồng thuận đàng hoàng, mạng bị chia cắt là dễ dính Split-Brain.
• Lúc đồng bộ (synchronization) thì cả cluster bị nghẽn.
• Classic queue mirroring bị thông báo deprecated từ 2021, RabbitMQ 3.13 là dòng cuối còn hỗ trợ, và RabbitMQ 4.0 bỏ hẳn.

# quorum queues: lựa chọn mặc định cho enterprise

Quorum Queues dựng trên thuật toán đồng thuận Raft Consensus:

• Không mất message đã confirm: message chỉ được xác nhận khi đã ghi xuống đĩa (fsync) của đa số node (Quorum: N/2 + 1). Còn đa số node sống là message đã confirm còn nguyên.
• Tự phục hồi: Leader sập thì hai Follower còn lại tự bầu Leader mới, thường mất cỡ vài giây tuỳ cấu hình phát hiện lỗi, không mất message đã confirm nào.
• Cái giá: mỗi message phải ghi đĩa và replicate, nên throughput thấp hơn classic queue không replicate, và cần ít nhất 3 node (2 node thì sập một node là mất quorum, queue đứng).

# không để mất message: producer confirms & manual ack

Muốn có At-Least-Once Delivery thật sự thì cả chuỗi từ Producer tới Consumer, khâu nào cũng phải có xác nhận:

# hai thứ bắt buộc phải có

1. Producer Confirms: đừng bao giờ "bắn rồi quên" (fire and forget). Producer phải chờ broker confirm, với Spring thì gắn CorrelationData vào lúc gửi và đọc kết quả từ nó (xem code ở dưới).
2. Consumer chỉ ack sau khi xử lý xong: không dùng auto-ack của giao thức AMQP (trong Spring là AcknowledgeMode.NONE), vì broker coi message đã xong ngay lúc gửi đi, consumer chết giữa chừng là mất. Chỉ gửi basicAck SAU KHI database transaction của consumer đã commit xong.

Chỗ này hay bị hiểu nhầm: AcknowledgeMode.AUTO của Spring không phải auto-ack của AMQP. Ở chế độ AUTO, container của Spring gọi listener, listener chạy xong không ném exception thì Spring mới basicAck, ném exception thì Spring basicNack/basicReject. Listener có @Transactional thì transaction commit trước khi method trả về, nên ack luôn đi sau commit. Đây là chế độ nên dùng trong đa số trường hợp. MANUAL chỉ cần khi bạn muốn tự ack ở thời điểm khác (ack theo lô, ack bất đồng bộ), và khi đó listener phải nhận Channel + deliveryTag để tự gọi basicAck; quên gọi hoặc ném exception mà không nack là message nằm "unacked" mãi, prefetch đầy dần rồi consumer đứng hình.

# xử lý message độc: dead letter exchange (dlx) & exponential retry

Một lỗi nặng hay gặp ở hệ thống xử lý message là Poison Pill Message (message độc):

• Một message sai định dạng hoặc chứa dữ liệu rác làm Consumer ném NullPointerException.
• Nếu consumer lỗi rồi gọi basicNack(requeue = true):
• Message bị đẩy ngược lại đầu hàng đợi, consumer nhận lại đúng message đó ngay → lại crash → lại nack!
• Vòng lặp vô tận này ăn 100% CPU và chặn hết các message hợp lệ phía sau!

# cấu hình spring boot 3 cho production

Cấu hình dưới đây gom các ý ở trên: quorum queue, DLX, deliveryLimit, prefetch, retry có backoff. Mình đã biên dịch thử với Spring Boot 3.5 (Spring AMQP 3.2).

package com.company.messaging.config;
 
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class EnterpriseRabbitMqConfig {
 
    public static final String ORDERS_EXCHANGE = "orders.direct.exchange";
    public static final String ORDERS_QUEUE = "orders.processing.queue";
    public static final String ORDERS_ROUTING_KEY = "orders.process";
 
    public static final String DLX_EXCHANGE = "orders.dlx.exchange";
    public static final String DLQ_QUEUE = "orders.dlq.queue";
    public static final String DLX_ROUTING_KEY = "orders.dlx";
 
    // 1. Khai báo Dead Letter Queue (Quorum Queue)
    @Bean
    public Queue deadLetterQueue() {
        return QueueBuilder.durable(DLQ_QUEUE)
            .quorum() // Cấu hình Quorum Queue (Raft)
            .build();
    }
 
    @Bean
    public DirectExchange deadLetterExchange() {
        return new DirectExchange(DLX_EXCHANGE);
    }
 
    @Bean
    public Binding dlqBinding() {
        return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange()).with(DLX_ROUTING_KEY);
    }
 
    // 2. Khai báo Main Processing Queue có gắn Dead Letter Routing
    @Bean
    public Queue mainOrdersQueue() {
        return QueueBuilder.durable(ORDERS_QUEUE)
            .quorum() // Sử dụng Quorum Queue thay cho Classic Queue
            .deadLetterExchange(DLX_EXCHANGE)
            .deadLetterRoutingKey(DLX_ROUTING_KEY)
            .deliveryLimit(5) // Lưới an toàn cuối: bị trả về quá 5 lần thì sang DLQ
            // Dead-letter kiểu at-least-once (mặc định quorum queue là at-most-once,
            // broker lỗi giữa chừng thì message dead-letter có thể mất)
            .overflow(QueueBuilder.Overflow.rejectPublish)
            .withArgument("x-dead-letter-strategy", "at-least-once")
            .build();
    }
 
    @Bean
    public DirectExchange mainOrdersExchange() {
        return new DirectExchange(ORDERS_EXCHANGE);
    }
 
    @Bean
    public Binding mainOrdersBinding() {
        return BindingBuilder.bind(mainOrdersQueue()).to(mainOrdersExchange()).with(ORDERS_ROUTING_KEY);
    }
 
    // 3. Cấu hình Listener Container với Prefetch Count chuẩn mực
    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
            CachingConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
 
        // BẮT BUỘC: Khống chế Prefetch Count để tránh OOM
        factory.setPrefetchCount(50); // Mỗi consumer chỉ nhận tối đa 50 messages chưa ack
 
        // AUTO của Spring: listener chạy xong mới ack, ném exception thì nack
        factory.setAcknowledgeMode(AcknowledgeMode.AUTO);
 
        // Retry trong bộ nhớ với exponential backoff: 1s, 2s (tối đa 10s), thử 3 lần.
        // Hết lượt thì reject không requeue -> message sang DLX
        factory.setAdviceChain(RetryInterceptorBuilder.stateless()
            .maxAttempts(3)
            .backOffOptions(1000, 2.0, 10000)
            .recoverer(new RejectAndDontRequeueRecoverer())
            .build());
 
        return factory;
    }
}

Có hai tầng retry ở đây. RetryInterceptorBuilder retry ngay trong consumer, có backoff, nhưng trong lúc chờ thì thread consumer đứng yên, nên chỉ hợp với backoff ngắn (vài giây). Muốn chờ lâu hơn (vài phút, vài giờ) mà không giữ thread thì dùng mô hình retry queue có TTL như sơ đồ ở trên. Còn deliveryLimit của quorum queue là lưới an toàn cuối: consumer crash, mất kết nối, message bị trả về quá 5 lần thì broker tự đẩy sang DLX. Từ RabbitMQ 4.0, quorum queue có delivery-limit mặc định là 20 kể cả khi bạn không đặt.

Phía producer, bật confirm trong application.yml rồi chờ kết quả qua CorrelationData:

spring:
  rabbitmq:
    publisher-confirm-type: correlated # broker trả confirm cho từng message
    publisher-returns: true # message không route được vào queue nào thì bị trả lại
    template:
      mandatory: true

public void publish(OrderEvent event) throws Exception {
    CorrelationData correlation = new CorrelationData(UUID.randomUUID().toString());
    rabbitTemplate.convertAndSend(ORDERS_EXCHANGE, ORDERS_ROUTING_KEY, event, correlation);
 
    // Chờ broker confirm (quorum queue: đã ghi xuống đa số node)
    CorrelationData.Confirm confirm = correlation.getFuture().get(5, TimeUnit.SECONDS);
    if (!confirm.isAck()) {
        throw new IllegalStateException("Broker nack: " + confirm.getReason());
    }
    if (correlation.getReturned() != null) {
        throw new IllegalStateException("Message không route được tới queue nào");
    }
}

Chờ đồng bộ từng message như trên là dễ hiểu nhất nhưng chậm. Lưu lượng lớn thì gửi cả loạt rồi mới chờ các future, hoặc xử lý confirm bất đồng bộ qua rabbitTemplate.setConfirmCallback(...). Message của Spring mặc định đã là persistent (deliveryMode = 2), không cần set thêm.

# so sánh: rabbitmq vs apache kafka

# tổng kết

Muốn dùng RabbitMQ cho chắc tay thì phải hiểu khá kỹ giới hạn của phần cứng và giao thức mạng:

• Tiết kiệm tài nguyên: luôn dùng Channel Multiplexing trên một kết nối TCP, và giới hạn prefetchCount để Consumer khỏi tràn RAM.
• Không mất dữ liệu: chuyển hẳn sang Quorum Queues (Raft Consensus), đi kèm Producer Confirms và Manual ACK.
• Phòng thủ từ trước: dựng Dead Letter Queue (DLQ) cùng deliveryLimit để cách ly message độc, giữ cho cả hệ thống xử lý event chạy thông suốt.

Chỉ là những ghi chép cá nhân với hy vọng mang lại chút giá trị. Nếu thấy hữu ích, đừng ngại chia sẻ cho bạn bè & đồng nghiệp nhé!

Happy coding 😎 👍🏻 🚀 🔥.