1. Giới thiệu về Hàng Đợi Tin Nhắn (Message Queue) và Lợi Ích

Hệ thống hàng đợi tin nhắn đóng vai trò quan trọng trong việc xây dựng các ứng dụng phân tán, mang lại nhiều lợi ích đáng kể:

• Xử lý bất đồng bộ (Asynchronous Processing): Thay vì xử lý trực tiếp một tác vụ tốn thời gian, ứng dụng có thể đẩy tác vụ đó vào hàng đợi và phản hồi ngay lập tức. Một tiến trình khác sẽ lấy tác vụ từ hàng đợi và xử lý sau. Điều này giúp cải thiện hiệu suất và trải nghiệm người dùng.
• Gỡ bỏ phụ thuộc giữa các thành phần (Application Decoupling): Các dịch vụ không cần biết về sự tồn tại hay cách thức hoạt động của nhau. Chúng chỉ cần gửi hoặc nhận tin nhắn thông qua hàng đợi, giúp hệ thống trở nên linh hoạt và dễ bảo trì hơn.
• Điều tiết lưu lượng truy cập (Traffic Shaping/Buffering): Khi hệ thống nhận được lượng yêu cầu lớn đột biến, hàng đợi có thể lưu trữ tạm thời các yêu cầu đó. Các dịch vụ xử lý sẽ lấy và xử lý dần theo khả năng, tránh tình trạng quá tải và sập hệ thống.
• Xử lý log tập trung (Centralized Logging): Các ứng dụng có thể gửi thông tin log của mình vào một hàng đợi chung. Một dịch vụ chuyên biệt sẽ lắng nghe hàng đợi này để thu thập, phân tích và lưu trữ log một cách tập trung.

2. Cài Đặt RabbitMQ

Để triển khai RabbitMQ, chúng ta cần cài đặt Erlang trước, sau đó là RabbitMQ Server.

2.1. Cài đặt Erlang

RabbitMQ được viết bằng ngôn ngữ Erlang, do đó Erlang là một tiền đề bắt buộc. Truy cập trang tải xuống Erlang và chọn phiên bản phù hợp với hệ điều hành của bạn. Thực hiện các bước cài đặt theo hướng dẫn mặc định.

Sau khi cài đặt xong, bạn cần cấu hình biến môi trường hệ thống. Đảm bảo rằng đường dẫn đến thư mục bin của Erlang đã được thêm vào biến Path. Để kiểm tra, mở Command Prompt (Windows + R, gõ cmd) và nhập lệnh erl. Nếu hiển thị thông tin phiên bản Erlang, quá trình cài đặt và cấu hình đã thành công.

2.2. Cài đặt RabbitMQ Server

Tải xuống RabbitMQ Server từ trang cài đặt RabbitMQ cho Windows (hoặc chọn phiên bản tương ứng với hệ điều hành của bạn). Thực hiện cài đặt theo các bước mặc định.

Sau khi cài đặt, bạn cần kích hoạt plugin quản lý RabbitMQ để có thể truy cập giao diện web:

1. Mở Command Prompt với quyền quản trị.
2. Điều hướng đến thư mục sbin trong thư mục cài đặt RabbitMQ (ví dụ: C:\Program Files\RabbitMQ Server\rabbitmq_server-3.x.x\sbin).
3. Chạy lệnh sau để kích hoạt plugin quản lý:

   rabbitmq-plugins enable rabbitmq_management
   

4. Kiểm tra trạng thái của RabbitMQ Server và đảm bảo nó đang chạy:

   rabbitmqctl status
   

   Nếu hiển thị thông tin trạng thái thành công, RabbitMQ Server đã được cài đặt và khởi động.

5. Để khởi động RabbitMQ Server thủ công (nếu chưa chạy hoặc sau khi tắt máy), bạn có thể chạy file rabbitmq-server.bat trong thư mục sbin.
6. Tru cập giao diện quản lý RabbitMQ qua trình duyệt tại địa chỉ: http://localhost:15672. Tài khoản và mật khẩu mặc định là guest/guest.

2.3. Quản lý Người Dùng và Virtual Host

Virtual Host (VHost) trong RabbitMQ cung cấp một không gian logic riêng biệt cho các ứng dụng, giúp cô lập dữ liệu và cấu hình. Mỗi VHost có thể có người dùng, hàng đợi và trao đổi riêng.

Ví dụ: Thêm người dùng và VHost qua giao diện quản lý

1. Đăng nhập vào giao diện quản lý (ví dụ: http://localhost:15672).
2. Vào mục Admin, chọn Users để thêm người dùng mới. Cung cấp tên người dùng, mật khẩu và cấp quyền (ví dụ: administrator).
3. Vào mục Admin, chọn Virtual hosts để thêm một VHost mới (ví dụ: /my_app_vhost).
4. Quay lại mục Users, chọn người dùng vừa tạo, sau đó click vào Permissions. Chọn VHost mới tạo và cấp quyền (ví dụ: .* cho configure, write, read).

3. Các Thành Phần Cơ Bản Trong RabbitMQ

Để hiểu rõ hơn về các mô hình làm việc của RabbitMQ, chúng ta cần nắm vững các khái niệm cốt lõi của AMQP (Advanced Message Queuing Protocol):

• Producer (Nhà Sản Xuất): Ứng dụng gửi tin nhắn.
• Consumer (Người Tiêu Thụ): Ứng dụng nhận và xử lý tin nhắn.
• Queue (Hàng Đợi): Nơi lưu trữ tin nhắn cho đến khi Consumer xử lý.
• Exchange (Bộ Trao Đổi): Nhận tin nhắn từ Producer và định tuyến chúng tới một hoặc nhiều Queue dựa trên các quy tắc (Exchange Type và Routing Key).
• Binding (Liên Kết): Một mối quan hệ giữa Exchange và Queue, xác định cách tin nhắn được định tuyến.
• Routing Key (Khóa Định Tuyến): Một thuộc tính của tin nhắn mà Exchange sử dụng để quyết định Queue nào sẽ nhận tin nhắn.

4. Các Mô Hình Trao Đổi Tin Nhắn Phổ Biến Với RabbitMQ

RabbitMQ hỗ trợ nhiều mô hình trao đổi tin nhắn khác nhau, chủ yếu được xác định bởi loại Exchange được sử dụng.

4.1. Mô Hình Hàng Đợi Đơn Giản (Simple Queue)

Đây là mô hình cơ bản nhất: một Producer gửi tin nhắn đến một Queue, và một Consumer nhận tin nhắn từ Queue đó.

Hạn chế: Mô hình này có tính ghép nối cao. Nếu muốn có nhiều Consumer xử lý các bản sao của cùng một tin nhắn, hoặc muốn định tuyến tin nhắn phức tạp hơn, mô hình này không phù hợp.

Ví dụ mã nguồn (Java với Spring AMQP)

Đầu tiên, bạn cần thêm dependency Spring AMQP vào file pom.xml:

<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

Cấu hình RabbitMQ trong application.properties:

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/

Cấu hình Hàng đợi (Queue Configuration)

package com.example.rabbitmqdemo.config;

import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class SimpleQueueConfig {

    public static final String MY_SIMPLE_QUEUE = "simpleQueue";

    // Khởi tạo một hàng đợi đơn giản
    @Bean
    public Queue simpleMessageQueue() {
        // Tham số thứ hai (true) nghĩa là hàng đợi này sẽ bền vững (durable)
        // Tin nhắn sẽ không bị mất nếu RabbitMQ server khởi động lại
        return new Queue(MY_SIMPLE_QUEUE, true);
    }
}

Nhà Sản Xuất (Producer)

package com.example.rabbitmqdemo.producer;

import com.example.rabbitmqdemo.config.SimpleQueueConfig;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class SimpleMessageProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("/sendSimpleMessage")
    public String sendSimpleMessage() {
        String message = "Xin chào từ hàng đợi đơn giản!";
        rabbitTemplate.convertAndSend(SimpleQueueConfig.MY_SIMPLE_QUEUE, message);
        System.out.println("Đã gửi tin nhắn đơn giản: " + message);
        return "Gửi tin nhắn thành công!";
    }
}

Người Tiêu Thụ (Consumer)

package com.example.rabbitmqdemo.consumer;

import com.example.rabbitmqdemo.config.SimpleQueueConfig;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = SimpleQueueConfig.MY_SIMPLE_QUEUE)
public class SimpleMessageConsumer {

    @RabbitHandler
    public void receiveMessage(String messageContent) {
        System.out.println("Consumer nhận được tin nhắn: " + messageContent);
    }
}

4.2. Mô Hình Hàng Đợi Công Việc (Work Queues)

Mô hình này cho phép nhiều Consumer cùng nhận tin nhắn từ một Queue. Các tin nhắn sẽ được phân phối lần lượt cho các Consumer theo cơ chế Round-robin.

Phân phối công bằng (Fair Dispatch): Theo mặc định, RabbitMQ phân phối tin nhắn theo cơ chế Round-robin mà không quan tâm đến khả năng xử lý của Consumer. Để đảm bảo phân phối công bằng hơn (Consumer nào xử lý xong tin nhắn trước sẽ nhận tin nhắn tiếp theo), chúng ta cần sử dụng chế độ xác nhận thủ công (Manual Acknowledgment) và cấu hình prefetch count.

• Xác nhận tin nhắn (Message Acknowledgment): Consumer gửi tín hiệu xác nhận cho RabbitMQ sau khi xử lý xong tin nhắn. Nếu Consumer chết trước khi xác nhận, tin nhắn sẽ được trả lại Queue và gửi cho Consumer khác.
• Prefetch Count: Số lượng tin nhắn tối đa mà RabbitMQ sẽ gửi cho một Consumer mà chưa nhận được xác nhận. Điều này giúp ngăn chặn một Consumer nhận quá nhiều tin nhắn và bị quá tải.

Độ bền của tin nhắn (Message Durability)

Để đảm bảo tin nhắn không bị mất khi RabbitMQ Server gặp sự cố hoặc khởi động lại, chúng ta cần cấu hình Queue và tin nhắn là bền vững (durable).

• Durable Queue: Khi khai báo Queue, đặt thuộc tính durable là true.
• Persistent Message: Khi gửi tin nhắn, đặt thuộc tính MessageDeliveryMode.PERSISTENT cho tin nhắn.

Lưu ý: RabbitMQ không cho phép định nghĩa lại một hàng đợi đã tồn tại với các thuộc tính khác. Nếu bạn thay đổi thuộc tính durable của một hàng đợi đã tồn tại trong mã nguồn, nó có thể gây ra lỗi khi ứng dụng khởi động.

Ví dụ mã nguồn (Java với Spring AMQP)

Cấu hình Hàng đợi Công Việc

package com.example.rabbitmqdemo.config;

import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class WorkQueueConfig {

    public static final String WORK_TASK_QUEUE = "taskProcessingQueue";

    @Bean
    public Queue taskProcessingQueue() {
        return new Queue(WORK_TASK_QUEUE, true); // durable = true
    }
}

Nhà Sản Xuất (Producer)

package com.example.rabbitmqdemo.producer;

import com.example.rabbitmqdemo.config.WorkQueueConfig;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class WorkTaskProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("/sendTaskMessage")
    public String sendTaskMessage(@RequestParam(defaultValue = "1") int count) {
        for (int i = 0; i < count; i++) {
            String message = "Công việc số " + (i + 1);
            // Đặt tin nhắn là persistent
            rabbitTemplate.convertAndSend(WorkQueueConfig.WORK_TASK_QUEUE, message, m -> {
                m.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                return m;
            });
            System.out.println("Đã gửi: " + message);
        }
        return "Gửi " + count + " tin nhắn công việc thành công!";
    }
}

Người Tiêu Thụ (Consumers)

Cấu hình application.properties cho phép xác nhận thủ công và đặt prefetch count:

# Cấu hình cho Work Queues
spring.rabbitmq.listener.simple.acknowledge-mode=manual
spring.rabbitmq.listener.simple.prefetch=1

Consumer 1:

package com.example.rabbitmqdemo.consumer;

import com.example.rabbitmqdemo.config.WorkQueueConfig;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = WorkQueueConfig.WORK_TASK_QUEUE)
public class TaskConsumerOne {

    @RabbitHandler
    public void processTask(String taskMessage, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws Exception {
        System.out.println("Consumer 1 đang xử lý: " + taskMessage + " (Delivery Tag: " + deliveryTag + ")");
        Thread.sleep(100); // Giả lập thời gian xử lý
        channel.basicAck(deliveryTag, false); // Xác nhận đã xử lý
        System.out.println("Consumer 1 đã hoàn thành: " + taskMessage);
    }
}

Consumer 2:

package com.example.rabbitmqdemo.consumer;

import com.example.rabbitmqdemo.config.WorkQueueConfig;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = WorkQueueConfig.WORK_TASK_QUEUE)
public class TaskConsumerTwo {

    @RabbitHandler
    public void processTask(String taskMessage, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws Exception {
        System.out.println("    Consumer 2 đang xử lý: " + taskMessage + " (Delivery Tag: " + deliveryTag + ")");
        Thread.sleep(500); // Giả lập thời gian xử lý lâu hơn
        channel.basicAck(deliveryTag, false); // Xác nhận đã xử lý
        System.out.println("    Consumer 2 đã hoàn thành: " + taskMessage);
    }
}

4.3. Mô Hình Publish/Subscribe (Fanout Exchange)

Trong mô hình này, Producer gửi tin nhắn đến một Fanout Exchange. Exchange này sẽ phát tán (broadcast) tin nhắn đến tất cả các Queue đã được liên kết với nó, bất kể Routing Key là gì.

Điều này rất hữu ích cho các trường hợp bạn muốn gửi cùng một tin nhắn tới nhiều ứng dụng khác nhau để xử lý độc lập.

Ví dụ mã nguồn (Java với Spring AMQP)

Cấu hình Fanout Exchange và Queues

package com.example.rabbitmqdemo.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FanoutExchangeConfig {

    public static final String FANOUT_EXCHANGE_NAME = "eventFanoutExchange";
    public static final String NOTIFICATION_QUEUE = "notificationQueue";
    public static final String ANALYTICS_QUEUE = "analyticsQueue";

    // Khởi tạo Fanout Exchange
    @Bean
    public FanoutExchange eventFanoutExchange() {
        return new FanoutExchange(FANOUT_EXCHANGE_NAME);
    }

    // Khởi tạo các hàng đợi
    @Bean
    public Queue notificationQueue() {
        return new Queue(NOTIFICATION_QUEUE, true);
    }

    @Bean
    public Queue analyticsQueue() {
        return new Queue(ANALYTICS_QUEUE, true);
    }

    // Liên kết Notification Queue với Fanout Exchange
    @Bean
    public Binding bindNotificationQueue(Queue notificationQueue, FanoutExchange eventFanoutExchange) {
        return BindingBuilder.bind(notificationQueue).to(eventFanoutExchange);
    }

    // Liên kết Analytics Queue với Fanout Exchange
    @Bean
    public Binding bindAnalyticsQueue(Queue analyticsQueue, FanoutExchange eventFanoutExchange) {
        return BindingBuilder.bind(analyticsQueue).to(eventFanoutExchange);
    }
}

Nhà Sản Xuất (Producer)

package com.example.rabbitmqdemo.producer;

import com.example.rabbitmqdemo.config.FanoutExchangeConfig;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class FanoutEventProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("/publishFanoutEvent")
    public String publishFanoutEvent() {
        String eventMessage = "Sự kiện mới vừa xảy ra!";
        // Gửi tin nhắn đến Fanout Exchange. Routing key sẽ bị bỏ qua (có thể là null hoặc rỗng)
        rabbitTemplate.convertAndSend(FanoutExchangeConfig.FANOUT_EXCHANGE_NAME, "", eventMessage);
        System.out.println("Đã gửi sự kiện Fanout: " + eventMessage);
        return "Gửi sự kiện Fanout thành công!";
    }
}

Người Tiêu Thụ (Consumers)

Consumer cho Notification:

package com.example.rabbitmqdemo.consumer;

import com.example.rabbitmqdemo.config.FanoutExchangeConfig;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = FanoutExchangeConfig.NOTIFICATION_QUEUE)
public class NotificationConsumer {

    @RabbitHandler
    public void processNotification(String message) {
        System.out.println("Consumer [Notification] nhận được: " + message);
        // Xử lý logic gửi thông báo cho người dùng
    }
}

Consumer cho Analytics:

package com.example.rabbitmqdemo.consumer;

import com.example.rabbitmqdemo.config.FanoutExchangeConfig;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = FanoutExchangeConfig.ANALYTICS_QUEUE)
public class AnalyticsConsumer {

    @RabbitHandler
    public void processAnalytics(String message) {
        System.out.println("Consumer [Analytics] nhận được: " + message);
        // Xử lý logic phân tích dữ liệu
    }
}

4.4. Mô Hình Định Tuyến Trực Tiếp (Direct Exchange)

Direct Exchange định tuyến tin nhắn đến các Queue dựa trên sự khớp chính xác của Routing Key. Một tin nhắn với Routing Key "error" sẽ chỉ được gửi đến các Queue đã liên kết với Direct Exchange bằng Routing Key "error".

Ví dụ mã nguồn (Java với Spring AMQP)

Cấu hình Direct Exchange và Queues

package com.example.rabbitmqdemo.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class DirectExchangeConfig {

    public static final String DIRECT_EXCHANGE_NAME = "appLogExchange";
    public static final String ERROR_QUEUE = "errorLogQueue";
    public static final String INFO_QUEUE = "infoLogQueue";
    public static final String ROUTING_KEY_ERROR = "error";
    public static final String ROUTING_KEY_INFO = "info";

    // Khởi tạo Direct Exchange
    @Bean
    public DirectExchange appLogDirectExchange() {
        return new DirectExchange(DIRECT_EXCHANGE_NAME);
    }

    // Khởi tạo các hàng đợi
    @Bean
    public Queue errorLogQueue() {
        return new Queue(ERROR_QUEUE, true);
    }

    @Bean
    public Queue infoLogQueue() {
        return new Queue(INFO_QUEUE, true);
    }

    // Liên kết Error Queue với Direct Exchange bằng routing key "error"
    @Bean
    public Binding bindErrorLogQueue(Queue errorLogQueue, DirectExchange appLogDirectExchange) {
        return BindingBuilder.bind(errorLogQueue).to(appLogDirectExchange).with(ROUTING_KEY_ERROR);
    }

    // Liên kết Info Queue với Direct Exchange bằng routing key "info"
    @Bean
    public Binding bindInfoLogQueue(Queue infoLogQueue, DirectExchange appLogDirectExchange) {
        return BindingBuilder.bind(infoLogQueue).to(appLogDirectExchange).with(ROUTING_KEY_INFO);
    }
}

Nhà Sản Xuất (Producer)

package com.example.rabbitmqdemo.producer;

import com.example.rabbitmqdemo.config.DirectExchangeConfig;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class DirectLogProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("/sendLog")
    public String sendLogMessage(@RequestParam String level, @RequestParam String message) {
        String routingKey = level.toLowerCase(); // level có thể là "error" hoặc "info"
        rabbitTemplate.convertAndSend(DirectExchangeConfig.DIRECT_EXCHANGE_NAME, routingKey, message);
        System.out.println("Đã gửi log [" + routingKey + "]: " + message);
        return "Gửi log thành công!";
    }
}

Người Tiêu Thụ (Consumers)

Consumer cho Error Logs:

package com.example.rabbitmqdemo.consumer;

import com.example.rabbitmqdemo.config.DirectExchangeConfig;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = DirectExchangeConfig.ERROR_QUEUE)
public class ErrorLogConsumer {

    @RabbitHandler
    public void processErrorMessage(String logMessage) {
        System.out.println("Consumer [ERROR Log] nhận được: " + logMessage);
        // Xử lý logic ghi log lỗi vào file hoặc gửi cảnh báo
    }
}

Consumer cho Info Logs:

package com.example.rabbitmqdemo.consumer;

import com.example.rabbitmqdemo.config.DirectExchangeConfig;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = DirectExchangeConfig.INFO_QUEUE)
public class InfoLogConsumer {

    @RabbitHandler
    public void processInfoMessage(String logMessage) {
        System.out.println("Consumer [INFO Log] nhận được: " + logMessage);
        // Xử lý logic ghi log thông tin
    }
}

4.5. Mô Hình Định Tuyến Chủ Đề (Topic Exchange)

Topic Exchange là Exchange linh hoạt nhất, định tuyến tin nhắn dựa trên các mẫu (patterns) của Routing Key. Nó sử dụng các ký tự đại diện:

• * (dấu sao): khớp với một từ bất kỳ.
• # (dấu thăng): khớp với không hoặc nhiều từ.

Các từ được phân tách bằng dấu chấm (.).

Ví dụ mã nguồn (Java với Spring AMQP)

Cấu hình Topic Exchange và Queues

package com.example.rabbitmqdemo.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class TopicExchangeConfig {

    public static final String TOPIC_EXCHANGE_NAME = "systemAlertsExchange";
    public static final String CRITICAL_ALERTS_QUEUE = "criticalAlertsQueue";
    public static final String ALL_ALERTS_QUEUE = "allAlertsQueue";

    // Khởi tạo Topic Exchange
    @Bean
    public TopicExchange systemAlertsTopicExchange() {
        return new TopicExchange(TOPIC_EXCHANGE_NAME);
    }

    // Khởi tạo các hàng đợi
    @Bean
    public Queue criticalAlertsQueue() {
        return new Queue(CRITICAL_ALERTS_QUEUE, true);
    }

    @Bean
    public Queue allAlertsQueue() {
        return new Queue(ALL_ALERTS_QUEUE, true);
    }

    // Liên kết Critical Alerts Queue với pattern "alert.critical.*"
    // Ví dụ: alert.critical.database, alert.critical.network
    @Bean
    public Binding bindCriticalAlertsQueue(Queue criticalAlertsQueue, TopicExchange systemAlertsTopicExchange) {
        return BindingBuilder.bind(criticalAlertsQueue).to(systemAlertsTopicExchange).with("alert.critical.*");
    }

    // Liên kết All Alerts Queue với pattern "alert.#"
    // Ví dụ: alert.info, alert.critical.database, alert.warning.system
    @Bean
    public Binding bindAllAlertsQueue(Queue allAlertsQueue, TopicExchange systemAlertsTopicExchange) {
        return BindingBuilder.bind(allAlertsQueue).to(systemAlertsTopicExchange).with("alert.#");
    }
}

Nhà Sản Xuất (Producer)

package com.example.rabbitmqdemo.producer;

import com.example.rabbitmqdemo.config.TopicExchangeConfig;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class TopicAlertProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("/sendTopicAlert")
    public String sendTopicAlert(@RequestParam String routingKey, @RequestParam String message) {
        rabbitTemplate.convertAndSend(TopicExchangeConfig.TOPIC_EXCHANGE_NAME, routingKey, message);
        System.out.println("Đã gửi cảnh báo [" + routingKey + "]: " + message);
        return "Gửi cảnh báo thành công!";
    }
}

Người Tiêu Thụ (Consumers)

Consumer cho Critical Alerts:

package com.example.rabbitmqdemo.consumer;

import com.example.rabbitmqdemo.config.TopicExchangeConfig;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = TopicExchangeConfig.CRITICAL_ALERTS_QUEUE)
public class CriticalAlertsConsumer {

    @RabbitHandler
    public void processCriticalAlert(String alertMessage) {
        System.err.println("!!! CẢNH BÁO NGUY HIỂM !!!: " + alertMessage);
        // Gửi email hoặc SMS cho quản trị viên
    }
}

Consumer cho All Alerts:

package com.example.rabbitmqdemo.consumer;

import com.example.rabbitmqdemo.config.TopicExchangeConfig;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = TopicExchangeConfig.ALL_ALERTS_QUEUE)
public class AllAlertsConsumer {

    @RabbitHandler
    public void processAllAlerts(String alertMessage) {
        System.out.println("--- Ghi nhận tất cả cảnh báo: " + alertMessage);
        // Lưu trữ tất cả cảnh báo vào cơ sở dữ liệu để phân tích
    }
}

4.6. Đảm Bảo Độ Tin Cậy Của Tin Nhắn

Đảm bảo tin nhắn không bị mất mát là yếu tố quan trọng trong các hệ thống phân tán.

• Độ bền của hàng đợi và tin nhắn (Queue & Message Durability): Như đã đề cập ở mô hình Work Queues, việc khai báo Queue là durable và gửi tin nhắn dưới dạng persistent giúp bảo vệ tin nhắn khỏi việc mất mát khi RabbitMQ Server khởi động lại.
• Xác nhận từ người tiêu thụ (Consumer Acknowledgments): Chế độ xác nhận thủ công (manual acknowledgment) cho phép Consumer chỉ thông báo cho RabbitMQ rằng tin nhắn đã được xử lý thành công sau khi hoàn tất công việc. Nếu Consumer gặp lỗi hoặc chết trước khi xác nhận, tin nhắn sẽ được gửi lại cho một Consumer khác.
• Xác nhận từ nhà sản xuất (Publisher Confirms): Đây là cơ chế tin cậy cao hơn so với giao dịch AMQP truyền thống. Khi Producer gửi tin nhắn, RabbitMQ có thể gửi lại một xác nhận (ACK) hoặc lỗi (NACK) cho Producer để thông báo rằng tin nhắn đã được nhận và xử lý (ví dụ: đã được lưu vào đĩa hoặc định tuyến thành công).
  • Giao dịch AMQP (AMQP Transactions): RabbitMQ hỗ trợ cơ chế giao dịch (txSelect, txCommit, txRollback) để nhóm các hoạt động gửi tin nhắn. Tuy nhiên, cơ chế này có hiệu năng thấp vì nó đồng bộ hóa các hoạt động, làm giảm thông lượng.
  • Chế độ Confirm (Confirm Mode): Là phương pháp được khuyến nghị để đảm bảo Publisher tin cậy. Producer được cấu hình để gửi tin nhắn trong chế độ confirm. Sau khi tin nhắn được broker nhận và xử lý (ví dụ: đã lưu vào hàng đợi bền vững), broker sẽ gửi một ACK trở lại Producer. Nếu có lỗi, broker sẽ gửi NACK. Chế độ này không chặn và cho phép hiệu năng cao hơn nhiều so với giao dịch.