当前位置：首页 > article >正文

分布式事物在RocketMQ中的应用

article 2025/2/28 9:56:34

RocketMQ 4.3 版本之后提供了对分布式事务消息的支持，它采用了一种类似于两阶段提交（2PC）的机制，但又有所不同，可以实现最终一致性的分布式事务。RocketMQ 的事务消息主要用于解决生产者发送消息和本地事务的原子性问题。

应用场景：

典型的场景是电商中的下单流程：

用户下单，订单服务需要创建订单。
同时，需要发送一个消息到库存服务，通知其扣减库存。

如果订单创建成功，但消息发送失败，或者消息发送成功，但订单创建失败，都会导致数据不一致。RocketMQ 的事务消息可以解决这个问题。

RocketMQ 事务消息原理：

发送半消息（Half Message）：
- 生产者首先发送一个半消息（Half Message）到 RocketMQ Broker。
- 半消息对消费者不可见，它会被存储在一个特殊的内部 topic（RMQ_SYS_TRANS_HALF_TOPIC）中。
- Broker 会响应生产者，确认半消息已接收。
执行本地事务：
- 生产者在收到 Broker 对半消息的确认后，执行本地事务（例如，创建订单）。
- 本地事务的执行结果可能是提交、回滚或未知。
提交/回滚半消息：
- 根据本地事务的执行结果，生产者向 Broker 发送提交（Commit）或回滚（Rollback）请求。
  - 提交： Broker 将半消息标记为可投递，并将其从 RMQ_SYS_TRANS_HALF_TOPIC 转移到真正的目标 topic。消费者现在可以看到并消费这条消息。
  - 回滚： Broker 删除半消息。
  - 未知： 生产者不发送任何请求，Broker 会定期回查生产者（见下文）。
事务回查 (Transaction Check)：
- 如果 Broker 长时间没有收到生产者对半消息的提交或回滚请求（可能是生产者宕机、网络问题等原因），Broker 会主动向生产者发送事务回查请求。
- 生产者收到回查请求后，需要检查本地事务的状态（通常通过查询数据库），并再次向 Broker 发送提交或回滚请求。
- RocketMQ默认回查15次, 可以通过transactionCheckMax参数配置.
消费者消费消息：
- 消费者从目标 topic 消费消息，执行相应的业务逻辑（例如，扣减库存）。
- 消费者的消费逻辑需要保证幂等性，因为在某些情况下（例如，消费者处理消息时宕机），消息可能会被重复消费。

关键概念：

半消息 (Half Message): 一种特殊的消息，对消费者不可见，用于在生产者和 Broker 之间进行事务协调。
事务回查 (Transaction Check): Broker 主动向生产者查询半消息状态的机制，用于处理生产者宕机或网络问题导致的消息状态不一致。
事务消息生产者 (TransactionMQProducer): RocketMQ 提供的特殊生产者，用于发送事务消息。需要配置 TransactionListener。
TransactionListener: 生产者端的监听器接口, 包含两个方法：
- executeLocalTransaction: 执行本地事务，并返回本地事务的状态（LocalTransactionState.COMMIT_MESSAGE、LocalTransactionState.ROLLBACK_MESSAGE 或 LocalTransactionState.UNKNOW）。
- checkLocalTransaction: 用于事务回查，检查本地事务的状态，并返回本地事务的状态。

代码示例 (Java):

import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.concurrent.*;

public class TransactionProducer {

    public static void main(String[] args) throws Exception {
        // 1. 创建事务消息生产者
        TransactionMQProducer producer = new TransactionMQProducer("transaction_producer_group");
        producer.setNamesrvAddr("localhost:9876"); // 设置 NameServer 地址

        // 2. 设置事务监听器
        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("client-transaction-msg-check-thread");
                return thread;
            }
        });
        producer.setExecutorService(executorService);

        producer.setTransactionListener(new TransactionListener() {
            // 执行本地事务
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                System.out.println("Executing local transaction...");
                try {
                    // 执行本地事务 (例如，创建订单)
                    // ...
                    boolean success = true; // 假设本地事务执行成功

                    if (success) {
                        System.out.println("Local transaction succeeded. Commit message.");
                        return LocalTransactionState.COMMIT_MESSAGE; // 提交消息
                    } else {
                        System.out.println("Local transaction failed. Rollback message.");
                        return LocalTransactionState.ROLLBACK_MESSAGE; // 回滚消息
                    }
                } catch (Exception e) {
                    System.out.println("Local transaction exception. Unknown state.");
                    return LocalTransactionState.UNKNOW; // 未知状态 (Broker 会回查)
                }
            }

            // 事务回查
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.out.println("Checking local transaction state...");
                // 检查本地事务的状态 (例如，查询订单数据库)
                // ...
                boolean exists = true; // 假设订单已创建

                if (exists) {
                    System.out.println("Local transaction exists. Commit message.");
                    return LocalTransactionState.COMMIT_MESSAGE;
                } else {
                    System.out.println("Local transaction does not exist. Rollback message.");
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
            }
        });

        // 3. 启动生产者
        producer.start();

        // 4. 发送事务消息
        String[] tags = new String[]{"TagA", "TagB", "TagC"};
        for (int i = 0; i < 3; i++) {
            Message msg = new Message("TopicTest", tags[i % tags.length], "KEY" + i,
                    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));

            // 发送事务消息
            SendResult sendResult = producer.sendMessageInTransaction(msg, null);
            System.out.printf("%s%n", sendResult);

            Thread.sleep(10);
        }

        // 5. 关闭生产者 (实际应用中，生产者通常不需要立即关闭)
        // producer.shutdown();
    }
}