1. Giải thích các tham số quan trọng trong RabbitMQ

Khi làm việc với RabbitMQ, việc hiểu rõ các tham số cấu hình là điều kiện tiên quyết để xây dựng hệ thống ổn định:

• durable: Nếu đặt là true, hàng đợi hoặc exchange sẽ được lưu trữ trên đĩa, không bị mất khi broker khởi động lại.
• exclusive: Nếu là true, hàng đợi chỉ được sử dụng bởi kết nối hiện tại và sẽ bị xóa khi kết nối đó đóng lại.
• autoDelete: Khi giá trị là true, hàng đợi sẽ tự động bị xóa nếu không còn người tiêu thụ (consumer) nào sử dụng.
• routingKey: Khóa định tuyến dùng để điều hướng tin nhắn từ Exchange đến các Queue tương ứng.
• deliveryTag: Số thứ tự của tin nhắn trên một channel, bắt đầu từ 1.
• alternate-exchange: Exchange dự phòng, được cấu hình để tiếp nhận các tin nhắn không thể định tuyến bởi exchange chính.
• x-match: Sử dụng trong Header Exchange. any nghĩa là chỉ cần khớp một cặp header, all nghĩa là phải khớp toàn bộ.

2. Cơ chế gửi và nhận tin nhắn cơ bản

Dưới đây là cách thiết lập một Consumer và Producer cơ bản sử dụng thư viện RabbitMQ Java Client.

Consumer (Người nhận tin)

public class SimpleReceiver {
    private final static String QUEUE_NAME = "task_queue";

    public static void main(String[] args) throws Exception {
        Connection factory = MqHelper.createConnection();
        Channel mqChannel = factory.createChannel();
        
        mqChannel.queueDeclare(QUEUE_NAME, true, false, false, null);
        
        Consumer messageConsumer = new DefaultConsumer(mqChannel) {
            @Override
            public void handleDelivery(String tag, Envelope env, AMQP.BasicProperties props, byte[] data) {
                String content = new String(data, StandardCharsets.UTF_8);
                System.out.println("Nhận tin nhắn: " + content);
                System.out.println("Routing Key: " + env.getRoutingKey());
            }
        };
        
        mqChannel.basicConsume(QUEUE_NAME, true, messageConsumer);
    }
}

Producer (Người gửi tin)

public class SimpleSender {
    public static void main(String[] args) throws Exception {
        try (Connection conn = MqHelper.createConnection();
             Channel channel = conn.createChannel()) {
            
            String targetQueue = "task_queue";
            channel.queueDeclare(targetQueue, true, false, false, null);
            
            String payload = "Dữ liệu hệ thống RabbitMQ";
            channel.basicPublish("", targetQueue, null, payload.getBytes());
            System.out.println("Đã gửi: " + payload);
        }
    }
}

3. Xử lý Callback và xác nhận tin nhắn

Để đảm bảo tin nhắn không bị thất thoát, RabbitMQ cung cấp cơ chế xác nhận (Confirm/Return) từ cả phía Producer và Consumer.

public class AdvancedProducer {
    private static final String MAIN_EX = "main_exchange";
    private static final String BACKUP_EX = "backup_exchange";

    public static void main(String[] args) throws Exception {
        Connection connection = MqHelper.createConnection();
        Channel channel = connection.createChannel();

        Map<String, Object> argsMap = new HashMap<>();
        argsMap.put("alternate-exchange", BACKUP_EX);

        channel.exchangeDeclare(MAIN_EX, BuiltinExchangeType.DIRECT, true, false, argsMap);
        channel.exchangeDeclare(BACKUP_EX, BuiltinExchangeType.FANOUT, true);

        // Lắng nghe xác nhận từ Broker
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long seq, boolean multi) {
                System.out.println("Broker đã nhận tin: " + seq);
            }
            @Override
            public void handleNack(long seq, boolean multi) {
                System.err.println("Broker từ chối tin: " + seq);
            }
        });

        // Lắng nghe tin nhắn bị trả về (không tìm thấy hàng đợi)
        channel.addReturnListener((replyCode, replyText, exchange, routingKey, properties, body) -> {
            System.out.println("Tin nhắn bị trả lại: " + new String(body));
        });

        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                .correlationId(UUID.randomUUID().toString())
                .deliveryMode(2) // Persistent
                .build();

        channel.basicPublish(MAIN_EX, "invalid_key", true, props, "Data test".getBytes());
        Thread.sleep(5000);
    }
}

4. Phân loại Exchange trong RabbitMQ

Direct Exchange

Định tuyến tin nhắn dựa trên sự trùng khớp hoàn toàn giữa routingKey của tin nhắn và khóa liên kết của hàng đợi.

channel.exchangeDeclare("direct_logs", BuiltinExchangeType.DIRECT);
channel.queueBind("queue_info", "direct_logs", "info_key");
channel.basicPublish("direct_logs", "info_key", null, "Log info message".getBytes());

Fanout Exchange

Cơ chế Broadcast, gửi tin nhắn đến tất cả các hàng đợi đang gắn (bind) vào nó mà không quan tâm đến routing key.

channel.exchangeDeclare("broadcast_ex", BuiltinExchangeType.FANOUT);
String tempQueue = channel.queueDeclare().getQueue();
channel.queueBind(tempQueue, "broadcast_ex", "");

Headers Exchange

Định tuyến dựa trên các thuộc tính trong phần header của tin nhắn thay vì routing key.

Map<String, Object> bindHeaders = new HashMap<>();
bindHeaders.put("x-match", "all");
bindHeaders.put("format", "pdf");
bindHeaders.put("type", "report");

channel.queueBind("pdf_queue", "header_ex", "", bindHeaders);

Topic Exchange

Định tuyến linh hoạt dựa trên pattern matching. Sử dụng ký tự đại diện:

• *: Thay thế cho đúng 1 từ.
• #: Thay thế cho 0 hoặc nhiều từ.

// Consumer binding
channel.queueBind("usa_news", "topic_ex", "usa.#"); // Nhận mọi tin về usa
channel.queueBind("all_weather", "topic_ex", "*.weather"); // Nhận tin thời tiết của bất kỳ vùng nào

// Producer publishing
channel.basicPublish("topic_ex", "usa.news.economic", null, "US Market".getBytes());
channel.basicPublish("topic_ex", "hanoi.weather", null, "Rainy".getBytes());