Giới thiệu Hàng đợi Tin nhắn

Hàng đợi tin nhắn là giải pháp thay thế giao tiếp HTTP truyền thống, cho phép tách biệt người gửi (producer) và người nhận (consumer) thông qua cơ chế trung gian. Khi tin nhắn được đưa vào hàng đợi, hệ thống thông báo cho consumer xử lý, tăng khả năng mở rộng và giảm tải cao điểm.

Lựa chọn Hàng đợi Tin nhắn

• RabbitMQ: Phù hợp hệ thống doanh nghiệp, định tuyến phức tạp
• Kafka: Xử lý dữ liệu lớn, nhật ký hệ thống
• RocketMQ: Giao dịch thương mại điện tử, tài chính

Cài đặt RabbitMQ

# Cài đặt môi trường Erlang
sudo apt install erlang

# Cài đặt RabbitMQ
sudo apt install rabbitmq-server

# Kích hoạt bảng quản lý
sudo rabbitmq-plugins enable rabbitmq_management

# Tạo người dùng mới
sudo rabbitmqctl add_user admin admin
sudo rabbitmqctl set_user_tags admin administrator

Kiến trúc RabbitMQ

• Channel: Kênh giao tiếp sử dụng giao thức AMQP
• Exchange: Chuyển tiếp tin nhắn tới hàng đợi
• Queue: Nơi lưu trữ tin nhắn chờ xử lý
• Virtual Host: Cách ly môi trường hoạt động

Tích hợp với Spring Boot

@Configuration
public class RabbitConfig {
    
    @Bean
    public Queue exampleQueue() {
        return new Queue("demo_queue");
    }

    @Bean
    public TopicExchange exampleExchange() {
        return new TopicExchange("demo_exchange");
    }

    @Bean
    public Binding setupBinding(Queue q, TopicExchange e) {
        return BindingBuilder.bind(q).to(e).with("routing.demo");
    }
}

Producer Implementation

@Autowired
private RabbitTemplate rabbitTemplate;

public void guiTinNhan(String content) {
    rabbitTemplate.convertAndSend("demo_exchange", "routing.demo", content);
}

Consumer Implementation

@RabbitListener(queues = "demo_queue")
public void xuLyTinNhan(String msg) {
    System.out.println("Nhận tin: " + msg);
}

Dead Letter Queue

@Bean
public Queue mainQueue() {
    return QueueBuilder.durable("primary")
        .deadLetterExchange("dlx_exchange")
        .deadLetterRoutingKey("dead_routing")
        .ttl(10000)
        .maxLength(100)
        .build();
}

Work Queues Pattern

@Bean
public SimpleRabbitListenerContainerFactory containerFactory() {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setPrefetchCount(1);
    return factory;
}

@RabbitListener(queues = "task_queue", containerFactory = "containerFactory", concurrency = "5")
public void worker(String task) {
    // Xử lý tác vụ
}

Các Mô hình Exchange

Pub/Sub (fanout)

@Bean
public FanoutExchange pubSubExchange() {
    return new FanoutExchange("broadcast");
}

Routing chủ đề

@Bean
public TopicExchange topicExchange() {
    return new TopicExchange("topic_exchange");
}

@Bean
public Binding topicBinding(Queue q, TopicExchange e) {
    return BindingBuilder.bind(q).to(e).with("news.#");
}

Spring Cloud Stream

spring:
  cloud:
    stream:
      binders:
        rabbit_server:
          type: rabbit
          environment:
            spring:
              rabbitmq:
                host: localhost
      bindings:
        output:
          destination: stream_exchange

@Autowired
private StreamBridge streamBridge;

public void publish(String data) {
    streamBridge.send("output", data);
}

Spring Cloud Bus

spring:
  rabbitmq:
    host: localhost
management:
  endpoints:
    web:
      exposure:
        include: refresh