4. 消息队列与最终一致性：可靠消息最终一致性方案

聊到分布式事务，很多人第一反应就是TCC或者XA。但说实话，在金融交易系统里，最终一致性才是真正扛得住高并发的方案。我这些年做交易系统，大部分场景用的都是消息队列来实现最终一致性。

什么叫最终一致性？说白了就是：我不要求你俩同时成功，但我保证你俩最终都会成功。如果中间出了岔子，系统会自动重试，直到完成为止。

4.1 可靠消息最终一致性方案

这个方案的套路其实很简单。我画个图你就明白了：

这个流程里最关键的一步是什么？是第一步。服务A在自己的数据库里，把业务操作和消息状态放在同一个本地事务里。要么都成功，要么都回滚。这就保证了消息不会丢。

4.2 本地消息表设计

本地消息表，说白了就是一张用来记录消息状态的数据库表。它跟业务表在同一个数据库里，共享同一个事务。

表结构长这样：

CREATE TABLE `local_message` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `biz_id` varchar(64) NOT NULL COMMENT '业务ID，用于幂等',
  `biz_type` varchar(32) NOT NULL COMMENT '业务类型',
  `message_body` text NOT NULL COMMENT '消息内容（JSON）',
  `status` tinyint(4) NOT NULL DEFAULT '0' COMMENT '0-待发送 1-已发送 2-已确认 3-已死亡',
  `retry_count` int(11) NOT NULL DEFAULT '0' COMMENT '重试次数',
  `max_retry` int(11) NOT NULL DEFAULT '3' COMMENT '最大重试次数',
  `next_retry_time` datetime DEFAULT NULL COMMENT '下次重试时间',
  `create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
  `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_biz_id` (`biz_id`),
  KEY `idx_status_next_retry` (`status`, `next_retry_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

嗯，这里要注意几个关键点：

• biz_id 唯一索引：防止重复插入，这是幂等的基础
• status 状态机：0→1→2 的流转必须严格
• next_retry_time 索引：定时任务扫描时，只查需要重试的记录

定时任务的逻辑也很简单：

// 伪代码：定时任务扫描并投递消息
public void scanAndSend() {
    // 1. 查询待发送且到重试时间的消息
    List<LocalMessage> messages = messageMapper.selectPending(
        LocalMessageStatus.PENDING, new Date());
    
    for (LocalMessage msg : messages) {
        try {
            // 2. 投递到MQ
            mqProducer.send(msg.getBizType(), msg.getMessageBody());
            
            // 3. 更新状态为已发送
            messageMapper.updateStatus(msg.getId(), 
                LocalMessageStatus.SENT);
        } catch (Exception e) {
            // 4. 更新重试次数和下次重试时间
            messageMapper.updateRetry(msg.getId(),
                msg.getRetryCount() + 1,
                calculateNextRetryTime(msg.getRetryCount() + 1));
            
            // 5. 超过最大重试次数，标记死亡
            if (msg.getRetryCount() + 1 >= msg.getMaxRetry()) {
                messageMapper.updateStatus(msg.getId(),
                    LocalMessageStatus.DEAD);
                // 发送告警通知
                alertService.notify("消息已死亡: " + msg.getId());
            }
        }
    }
}

你想想看，这个方案的好处是什么？简单、可靠、不依赖外部组件。只要你的数据库不丢数据，消息就绝对不会丢。

4.3 RocketMQ事务消息实现

RocketMQ的事务消息，其实就是把本地消息表的逻辑搬到了MQ服务端。我个人觉得，这是目前最优雅的最终一致性方案。

它的核心流程分三步：

1. 发送半消息：生产者发送一条「半消息」到MQ，此时消费者不可见
2. 执行本地事务：生产者执行本地业务操作
3. 提交/回滚：根据本地事务结果，告诉MQ是提交还是回滚

代码实现大概是这样的：

// 1. 实现 TransactionListener 接口
public class OrderTransactionListener implements TransactionListener {
    
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
            // 执行本地事务（比如创建订单）
            orderService.createOrder((OrderDTO) arg);
            // 成功，提交消息
            return LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            // 失败，回滚消息
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }
    
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // 回查：检查本地事务是否成功
        String orderId = msg.getKeys();
        boolean success = orderService.checkOrderExists(orderId);
        return success ? 
            LocalTransactionState.COMMIT_MESSAGE : 
            LocalTransactionState.ROLLBACK_MESSAGE;
    }
}

// 2. 发送事务消息
public void sendTransactionMessage(OrderDTO order) {
    Message msg = new Message("order-topic", 
        JSON.toJSONBytes(order));
    msg.setKeys(order.getOrderId());
    
    // 发送事务消息
    transactionProducer.sendMessageInTransaction(
        msg, order);
}

我记得有一次线上事故，就是因为回查逻辑写错了。订单明明创建成功了，回查却返回了ROLLBACK，导致消息被删了，下游没收到通知。嗯，从那以后我定了个规矩：回查逻辑必须跟本地事务用同一套数据源。

4.4 消息去重与幂等性

消息队列有个特点：at-least-once，也就是至少投递一次。这意味着你的消费者可能会收到重复消息。所以，幂等性是必须的。

什么叫幂等？说白了就是：同一个操作执行一次和执行多次，结果是一样的。

我常用的幂等方案有这几种：

方案	实现方式	适用场景	注意事项
唯一索引	数据库唯一键约束	插入类操作	需处理主键冲突异常
状态机	通过状态字段判断	状态流转类操作	状态变更需原子性
去重表	独立的去重记录表	通用场景	需定期清理历史数据
Redis分布式锁	SETNX + 过期时间	高并发场景	需考虑锁超时问题

我个人最推荐的是去重表方案。它的核心逻辑是：

public void consumeMessage(MessageDTO msg) {
    // 1. 尝试插入去重记录
    try {
        deduplicationMapper.insert(new DeduplicationRecord(
            msg.getBizId(), msg.getBizType()));
    } catch (DuplicateKeyException e) {
        // 已处理过，直接跳过
        log.info("重复消息，已跳过: {}", msg.getBizId());
        return;
    }
    
    // 2. 执行业务逻辑
    try {
        businessService.process(msg);
    } catch (Exception e) {
        // 业务失败，删除去重记录（允许重试）
        deduplicationMapper.deleteByBizId(msg.getBizId());
        throw e;
    }
}

还有一个细节：去重记录要设置过期时间。比如保留7天，超过的就删掉。不然表会越来越大，影响性能。

小结

最终一致性方案，说白了就是用空间换时间，用异步换性能。本地消息表适合中小型系统，RocketMQ事务消息适合大型分布式系统。不管用哪种，幂等性都是最后一道防线。

做交易系统这么多年，我最大的体会是：不要追求绝对的一致，要追求可控的最终一致。只要你的补偿机制够完善，最终一致性完全够用。