3、消息队列与最终一致性：RocketMQ事务消息原理、Kafka幂等性与事务、本地消息表模式

聊到分布式系统的一致性，大家第一反应往往是两阶段提交（2PC）、TCC 这些强一致性方案。但说实话，我在实际项目中很少用它们——太笨重了，性能损耗大，而且对底层存储要求极高。真正在生产环境里扛大梁的，反而是「最终一致性」。

最终一致性，说白了就是：我不保证你写进去立刻就能读到，但我保证你最终一定能读到。这个「最终」可能是几毫秒，也可能是几秒。嗯，这里要注意，如果你的业务要求用户付款后立刻看到余额变化，那最终一致性可能不适合你。但如果是发一条通知、更新一个积分、同步一份报表——那它简直是神器。

这一章，我带你看看三种最主流的最终一致性实现方式：RocketMQ 的事务消息、Kafka 的幂等性与事务，以及经典的本地消息表模式。这三种我都踩过坑，咱们一个一个说。

3.1 RocketMQ 事务消息：两阶段提交的轻量替代

先说说 RocketMQ 的事务消息。我记得第一次看到这个特性时，心里想的是：「这不就是消息队列版的 2PC 吗？」后来仔细研究才发现，它比 2PC 聪明得多。

RocketMQ 事务消息的核心思路是：把本地事务和消息发送绑定在一起。要么都成功，要么都回滚。但它不像 2PC 那样锁住所有资源，而是通过「半消息」和「回查机制」来保证最终一致性。

3.1.1 半消息与回查机制

什么叫半消息？就是消息已经发到 Broker 了，但消费者暂时看不到。只有等本地事务执行成功，生产者再发一个 commit 指令，Broker 才会把消息变成「可消费」状态。

流程大概是这样的：

1. 生产者发送半消息到 Broker
2. Broker 存储半消息，返回确认
3. 生产者执行本地事务（比如写数据库）
4. 根据本地事务结果，发送 commit 或 rollback 给 Broker
5. 如果 Broker 长时间没收到 commit/rollback，会主动回查生产者

这里有个关键点：回查机制。我曾经在一个项目中遇到过网络抖动，commit 消息丢了。如果没有回查，那条半消息就永远卡在 Broker 里了。RocketMQ 的回查机制会定期询问生产者：「你那个事务到底怎么样了？」生产者根据本地事务表的状态来回复。

3.1.2 代码示例：RocketMQ 事务消息生产者

// 创建事务消息生产者
TransactionMQProducer producer = new TransactionMQProducer("transaction_producer_group");
producer.setNamesrvAddr("localhost:9876");
producer.start();

// 设置事务监听器
producer.setTransactionListener(new TransactionListener() {
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 执行本地事务，比如插入订单表
        try {
            orderService.createOrder((Order) arg);
            return LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // 回查：根据消息中的业务 key 查询本地事务状态
        String orderId = msg.getKeys();
        if (orderService.isOrderExists(orderId)) {
            return LocalTransactionState.COMMIT_MESSAGE;
        }
        return LocalTransactionState.UNKNOW;
    }
});

// 发送半消息
Message msg = new Message("order_topic", "create", orderId.getBytes());
SendResult result = producer.sendMessageInTransaction(msg, order);

3.2 Kafka 幂等性与事务：从消息不丢到精确一次

Kafka 的最终一致性方案和 RocketMQ 思路不太一样。Kafka 更强调「幂等性」和「精确一次语义」（Exactly-Once Semantics，简称 EOS）。

为什么需要幂等性？你想想看，生产者发送消息时如果网络超时，它会重试。重试可能导致 Broker 收到多条相同的消息。如果消费者不处理幂等，那数据就重复了。

3.2.1 幂等性生产者

Kafka 的幂等性其实很简单：每个生产者有一个唯一的 producer ID（PID），每条消息带一个序列号。Broker 会检查序列号，如果发现重复，直接丢弃。

开启方式就一行配置：

props.put("enable.idempotence", true);

就这么简单。但要注意，幂等性只保证单个分区内的消息不重复，跨分区它管不了。如果你需要跨分区的精确一次，那就得上事务了。

3.2.2 Kafka 事务：跨分区原子写入

Kafka 的事务机制，说白了就是让一批消息要么全部写入成功，要么全部不可见。它引入了一个叫 Transaction Coordinator 的组件，专门管理事务状态。

流程大致是：

1. 生产者向 Coordinator 注册事务
2. 生产者发送消息到多个分区（这些消息暂时标记为「未提交」）
3. 生产者提交事务，Coordinator 将事务标记为「已提交」
4. 消费者通过隔离级别配置，决定是否读取未提交的消息

3.2.3 代码示例：Kafka 事务生产者

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("transactional.id", "my-transactional-id");
props.put("enable.idempotence", true);

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

// 初始化事务
producer.initTransactions();

try {
    // 开始事务
    producer.beginTransaction();

    // 发送多条消息到不同分区
    producer.send(new ProducerRecord<>("topic1", "key1", "value1"));
    producer.send(new ProducerRecord<>("topic2", "key2", "value2"));

    // 提交事务
    producer.commitTransaction();
} catch (Exception e) {
    // 回滚事务
    producer.abortTransaction();
}

3.3 本地消息表模式：最朴素但最可靠

聊完两个消息队列的方案，咱们再来看一个「老派」但极其可靠的方式——本地消息表模式。说实话，我在很多传统企业项目里，反而觉得这个模式最稳。

它的思路特别朴素：把消息和业务数据放在同一个数据库里，通过本地事务保证一致性。然后另起一个定时任务，把消息发到 MQ。

3.3.1 核心流程

1. 业务操作和消息插入在同一个本地事务中
2. 定时任务扫描消息表，把状态为「待发送」的消息发到 MQ
3. MQ 消费者处理成功后，回调修改消息状态为「已发送」
4. 如果发送失败，定时任务会重试

你想想看，这个模式的好处是什么？它不依赖任何分布式事务协调器，只要你的数据库支持事务就行。MySQL、PostgreSQL、Oracle 都可以。

3.3.2 表结构设计

字段名	类型	说明
id	bigint	主键，自增
business_key	varchar(64)	业务唯一键，用于幂等
message_body	text	消息内容（JSON）
status	tinyint	0-待发送，1-已发送，2-已确认
retry_count	int	重试次数
create_time	datetime	创建时间

3.3.3 代码示例：本地消息表实现

@Transactional
public void createOrderAndSendMessage(Order order) {
    // 1. 插入订单
    orderDao.insert(order);

    // 2. 插入消息记录（同一个事务）
    MessageRecord record = new MessageRecord();
    record.setBusinessKey(order.getOrderId());
    record.setMessageBody(JSON.toJSONString(order));
    record.setStatus(0); // 待发送
    messageDao.insert(record);
}

// 定时任务：扫描并发送消息
@Scheduled(fixedDelay = 5000)
public void scanAndSend() {
    List<MessageRecord> records = messageDao.selectByStatus(0);
    for (MessageRecord record : records) {
        try {
            // 发送到 MQ
            mqProducer.send("order_topic", record.getMessageBody());
            // 更新状态
            messageDao.updateStatus(record.getId(), 1);
        } catch (Exception e) {
            // 记录失败，下次重试
            log.error("发送失败，消息ID: {}", record.getId(), e);
        }
    }
}

3.4 三种方案对比

说了这么多，咱们来做个对比。我根据自己的项目经验，整理了一张表：

方案	一致性保证	性能影响	复杂度	适用场景
RocketMQ 事务消息	最终一致，支持回查	中等（半消息+回查开销）	中等	需要强事务保证，且使用 RocketMQ
Kafka 幂等性+事务	精确一次（分区内/跨分区）	较高（事务协调器开销）	较高	大数据场景，跨分区原子写入
本地消息表模式	最终一致，依赖数据库	低（仅数据库开销）	低	传统企业项目，数据库为主

我个人习惯是：如果团队已经用了 RocketMQ，优先用它的事务消息；如果是 Kafka 生态，幂等性基本必开，事务按需开启；如果是老项目或者数据库为主的场景，本地消息表模式最稳妥。

3.5 本章知识体系图

下面这张图，帮你把三种方案的核心逻辑串起来：

最后说一句：没有银弹。这三种方案我都用过，也都在生产环境出过问题。关键是理解它们的原理和局限，然后根据你的业务场景做选择。嗯，这一章的内容就到这儿，希望对你有所帮助。

---