📝引言：

在微服务架构和分布式系统中，消息队列（如 RabbitMQ、Kafka、RocketMQ） 成为了各大企业的“通信纽带”。然而，很多开发者掉以轻心，以为“用了消息队列就万无一失”。

事实却是：消息丢失、重复消费、顺序错乱、幂等难题频发，影响系统稳定性。

本篇文章，我们将从架构设计的高度出发，深入剖析“如何保证消息可靠性”。不仅提供完整的理论体系，还配套完整代码示例，助你真正掌握“消息可靠投递的核心技术”。

📌文章大纲：

为什么消息可靠性是大问题？

消息投递的全链路风险分析

消息可靠性的三大保障措施

实战：基于 RabbitMQ 的可靠消息投递 + 本地事务表方案

延伸：分布式事务下的最终一致性与 TCC 对比

📍一、为什么消息可靠性是大问题？

举个例子：
用户下单 -> 订单系统写入数据库成功 -> 通过 MQ 通知库存系统扣减库存。

问题：订单写入成功，MQ 没发送成功怎么办？

👉结果就是：订单有了，库存却没扣，形成“脏数据”或“超卖”。

📍二、消息投递的全链路风险分析

从消息生产到消费，链路如下图：

生产者 --> 消息中间件（MQ） --> 消费者

环节	风险点	对应问题
生产者	本地事务成功但 MQ 消息发送失败	数据不一致
MQ	异常宕机或队列满了	消息丢失
消费者	消费失败或处理超时	消息重复或失败

📍三、如何保障消息可靠性？三大核心措施

消息发送确认机制（Confirm + Return）

持久化消息（Durable Queue + Persistent Message）

消费幂等性设计（幂等Token/数据库幂等索引）

📍四、实战：RabbitMQ + 本地事务表解决方案

🛠️ 场景说明：
用户下单，订单系统保存订单后发送 MQ 消息，确保两者要么都成功，要么都失败（保证最终一致性）。

📂数据库表设计

CREATE TABLE order_info (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  user_id BIGINT,
  status VARCHAR(20),
  create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE message_log (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  msg_id VARCHAR(64),
  msg_body TEXT,
  status VARCHAR(20), -- 发送中/已发送/失败
  retry_count INT DEFAULT 0,
  next_retry TIMESTAMP,
  create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

💻核心代码

1. 发送前记录消息日志

@Transactional
public void createOrder(Order order) {
    // 保存订单
    orderMapper.insert(order);

    // 保存消息日志
    MessageLog log = new MessageLog();
    log.setMsgId(UUID.randomUUID().toString());
    log.setMsgBody(JSON.toJSONString(order));
    log.setStatus("SENDING");
    log.setNextRetry(LocalDateTime.now().plusMinutes(1));
    messageLogMapper.insert(log);

    // 异步发送消息
    rabbitTemplate.convertAndSend("order.exchange", "order.create", log.getMsgBody(), message -> {
        message.getMessageProperties().setMessageId(log.getMsgId());
        return message;
    });
}

2. 发送确认回调

rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
    if (ack) {
        // 修改日志状态为“已发送”
        messageLogMapper.updateStatus(correlationData.getId(), "SENT");
    } else {
        // 重试逻辑
        log.error("消息发送失败：{}", cause);
    }
});

3. 消费端保证幂等性处理

@RabbitListener(queues = "order.queue")
public void receiveOrder(String msg) {
    Order order = JSON.parseObject(msg, Order.class);
    
    // 幂等处理：检查是否已经处理
    if (orderService.isProcessed(order.getId())) {
        return;
    }

    // 扣减库存/发通知等
    orderService.process(order);
}

🔁延伸内容：与分布式事务（TCC/2PC）的对比

方式	一致性保障	性能	应用场景
MQ事务表	最终一致性	高	订单、支付、积分通知
TCC	强一致	中	银行转账、库存扣减
2PC	较强一致	低	传统数据库场景（不推荐）

✅总结

保障消息可靠性，不止是写个 RabbitMQ 的发送方法就够了，它是系统级别的稳定性设计。需要配合本地事务日志、确认机制、幂等控制，打造一个可用、可控、可恢复的消息链路。