RabbitMQ 消息队列实战详解

为什么使用消息队列？

• 解耦：生产者和消费者独立演化
• 异步：提升响应速度
• 削峰：缓冲突发流量
• 可靠：消息持久化不丢失

核心概念

• Producer：消息生产者
• Consumer：消息消费者
• Queue：消息队列
• Exchange：交换机，路由消息
• Binding：队列与交换机的绑定
• Routing Key：路由键

Exchange 类型

1. Direct Exchange

精确匹配 routing key

2. Fanout Exchange

广播到所有绑定队列

3. Topic Exchange

模式匹配（* 单词，# 多单词）

4. Headers Exchange

根据 header 属性匹配

Spring Boot 集成

生产者配置

@Configuration
public class RabbitConfig {
    @Bean
    public Queue orderQueue() {
        return new Queue("order.queue", true);
    }
    
    @Bean
    public TopicExchange exchange() {
        return new TopicExchange("order.exchange");
    }
    
    @Bean
    public Binding binding(Queue queue, TopicExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("order.#");
    }
}

发送消息

@Autowired
private RabbitTemplate rabbitTemplate;

public void createOrder(Order order) {
    rabbitTemplate.convertAndSend(
        "order.exchange",
        "order.created",
        order
    );
}

消费消息

@RabbitListener(queues = "order.queue")
public void handleOrder(Order order) {
    // 处理订单
    processOrder(order);
}

消息可靠性

1. 生产者确认

rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
    if (!ack) {
        log.error("消息发送失败：{}", cause);
    }
});

2. 消息持久化

// 队列持久化
new Queue("queue", true);

// 消息持久化
MessageProperties props = new MessageProperties();
props.setDeliveryMode(MessageDeliveryMode.PERSISTENT);

3. 消费者手动确认

@RabbitListener(queues = "order.queue")
public void handleOrder(Message message, Channel channel) {
    try {
        processOrder(message);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    } catch (Exception e) {
        // 拒绝消息，重新入队
        channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
    }
}

死信队列

@Bean
public Queue deadLetterQueue() {
    return new Queue("order.dlq", true);
}

@Bean
public Queue orderQueue() {
    Map<String, Object> args = new HashMap<>();
    args.put("x-dead-letter-exchange", "dlx.exchange");
    args.put("x-dead-letter-routing-key", "order.dead");
    return new Queue("order.queue", true, false, false, args);
}

延迟队列

// 设置消息 TTL
Map<String, Object> args = new HashMap<>();
args.put("x-message-ttl", 60000); // 60 秒
args.put("x-dead-letter-exchange", "dlx.exchange");

监控管理

• Management Plugin：Web 管理界面
• Prometheus + Grafana：监控指标
• 告警配置：队列积压、消费延迟

最佳实践

1. 消息体尽量小
2. 合理设置 prefetch
3. 处理幂等性
4. 监控队列长度
5. 定期清理死信

消息队列是分布式系统的核心组件，合理使用能显著提升系统可靠性！