RocketMQ事务消息实现教程详解

本篇文章向大家介绍《Java实现RocketMQ事务消息教程》，主要包括，具有一定的参考价值，需要的朋友可以参考一下。

RocketMQ事务消息通过“半消息”和“事务回查”机制解决分布式系统中的数据一致性问题。1. 事务生产者（TransactionMQProducer）发送半消息，消费者不可见；2. Broker回调executeLocalTransaction方法执行本地事务，决定提交、回滚或未知状态；3. 若状态为未知，Broker通过checkLocalTransaction方法进行回查，确保最终一致性；4. 本地事务状态需持久化存储，保障回查准确性；5. 消费者需实现幂等性，防止重复消费；6. Producer Group需唯一且稳定，确保回查机制正常运作。整个流程通过异步方式实现最终一致性，降低分布式事务复杂度，适用于高并发场景。

RocketMQ事务消息是一种非常实用的机制，它主要用来解决分布式系统中的数据一致性问题。简单来说，它通过“半消息”和“事务回查”的巧妙设计，确保了本地事务的执行结果能与消息的发送状态保持同步，最终实现业务数据的最终一致性。

解决方案

要操作RocketMQ事务消息，核心在于使用TransactionMQProducer和实现一个LocalTransactionListener接口。这听起来有点绕，但实际操作起来逻辑还是挺清晰的。

首先，你需要一个TransactionMQProducer来发送消息。这个生产者不同于普通的DefaultMQProducer，它专门用于处理事务。

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class TransactionProducerExample {

    public static void main(String[] args) throws MQClientException, InterruptedException {
        // 1. 初始化TransactionMQProducer
        TransactionMQProducer producer = new TransactionMQProducer("my_transaction_producer_group");
        producer.setNamesrvAddr("localhost:9876"); // 你的NameServer地址

        // 2. 设置事务监听器
        // 这里的线程池配置，说实话，挺重要的，它决定了并发执行本地事务的能力
        ExecutorService executorService = new ThreadPoolExecutor(
            2, // 核心线程数
            5, // 最大线程数
            100, // 线程空闲时间
            TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(2000), // 阻塞队列
            r -> {
                Thread thread = new Thread(r, "transaction-msg-check-thread");
                return thread;
            });
        producer.setExecutorService(executorService);
        producer.setTransactionListener(new MyTransactionListener()); // 后面会实现这个监听器

        // 3. 启动生产者
        producer.start();
        System.out.println("Transaction Producer Started.");

        // 4. 发送事务消息
        for (int i = 0; i < 1; i++) { // 示例只发送一条
            try {
                Message msg = new Message(
                    "TopicTest", // 主题
                    "TagA", // 标签
                    "KEY" + i, // 业务唯一Key，非常重要，用于回查
                    ("Hello RocketMQ Transaction " + i).getBytes());

                // 发送半消息，并传入一个args，这个args在executeLocalTransaction方法里能拿到
                // 我个人习惯把业务ID或者其他上下文信息放在这里
                producer.sendMessageInTransaction(msg, null); 
                System.out.printf("Sent transaction message: %s%n", msg);
                Thread.sleep(10); // 稍微停顿一下
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        // 实际应用中，生产者不会立即关闭，这里为了示例方便
        // producer.shutdown();
    }
}

// 事务监听器实现
class MyTransactionListener implements TransactionListener {

    // 这是一个模拟的本地事务状态存储，实际中应该是数据库或其他持久化存储
    // 我用Map是为了简化，但实际业务中，你要么查DB，要么查Redis，确保状态的准确性
    private final ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();
    private final AtomicInteger transactionIndex = new AtomicInteger(0);

    /**
     * 执行本地事务
     * 当生产者发送半消息成功后，Broker会回调这个方法，让你执行本地事务。
     * 这里的逻辑至关重要，它决定了消息最终是提交还是回滚。
     */
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 模拟本地事务执行
        int value = transactionIndex.getAndIncrement();
        int status = value % 3; // 0: COMMIT, 1: ROLLBACK, 2: UNKNOW (需要回查)

        // 假设我们把消息的Key作为本地事务的唯一标识
        String transId = msg.getKeys(); 
        localTrans.put(transId, status); // 存储本地事务状态

        System.out.printf("--- executeLocalTransaction: %s, Local Status: %d%n", transId, status);

        // 根据本地事务执行结果返回状态
        if (status == 0) {
            // 本地事务成功，提交消息
            System.out.println("Local transaction SUCCESS, COMMIT message.");
            return LocalTransactionState.COMMIT_MESSAGE;
        } else if (status == 1) {
            // 本地事务失败，回滚消息
            System.out.println("Local transaction FAILED, ROLLBACK message.");
            return LocalTransactionState.ROLLBACK_MESSAGE;
        } else {
            // 本地事务状态未知（可能正在处理中，或者出现异常），需要Broker进行回查
            System.out.println("Local transaction UNKNOWN, Broker will check later.");
            return LocalTransactionState.UNKNOW;
        }
    }

    /**
     * 检查本地事务状态
     * 当Broker发现有事务消息长时间处于UNKNOW状态时，会回调这个方法，
     * 让你再次确认本地事务的真实状态。
     * 这个方法是事务消息可靠性的最后一道防线。
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // 根据消息的Key，查询本地事务的真实状态
        // 这里的查询逻辑必须非常严谨，直接查询你的业务数据库，确保状态的准确性。
        String transId = msg.getKeys();
        Integer status = localTrans.get(transId); // 实际中这里应该去查数据库

        System.out.printf("+++ checkLocalTransaction: %s, Current Status: %s%n", transId, status);

        if (null != status) {
            if (status == 0) {
                System.out.println("Checker confirms: Local transaction SUCCESS, COMMIT message.");
                return LocalTransactionState.COMMIT_MESSAGE;
            } else if (status == 1) {
                System.out.println("Checker confirms: Local transaction FAILED, ROLLBACK message.");
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        }
        // 如果查不到状态，或者状态还是UNKNOW，就继续返回UNKNOW，等待下次回查
        // 实际场景中，如果长时间查不到，可能需要人工介入或报警
        System.out.println("Checker still UNKNOWN, waiting for next check.");
        return LocalTransactionState.UNKNOW;
    }
}

RocketMQ事务消息在分布式系统里究竟扮演什么角色？

在分布式系统里，数据一致性一直是个老大难的问题。你想啊，一个用户下单，既要创建订单，又要扣减库存，这俩操作可能在不同的服务甚至不同的数据库里。如果订单创建成功了，库存没扣成功，或者反过来，那不就乱套了吗？传统的2PC（两阶段提交）虽然能保证强一致性，但性能和可用性实在不尽如人意，而且对业务侵入性也大。

这时候，RocketMQ事务消息就登场了。它扮演的角色，说白了，就是个“协调员”和“兜底保障”。它不像2PC那么重，而是利用了消息队列的异步特性，实现了一种“最终一致性”的方案。它通过先发送一个“半消息”（意思是这消息还没确定是发出去还是不发），然后让你去执行本地事务，根据本地事务的结果再决定是提交这个消息还是回滚。万一本地事务执行完，但通知消息队列出错了，或者本地事务状态不明确，RocketMQ还会主动来“回查”，问你本地事务到底怎么样了。这样一来，即使网络抖动、服务崩溃，也能最大限度地保证消息和本地事务的一致性，大大降低了分布式事务的复杂性。我个人觉得，它在追求高性能和高可用性的互联网业务中，真的是个非常优雅的解决方案。

实现RocketMQ事务消息，具体需要哪些关键步骤和组件？

实现RocketMQ事务消息，其实是围绕着一个核心流程展开的，涉及几个关键组件的协同工作。

首先，是事务生产者（Transaction Producer）。它不是直接发送普通消息，而是先发送一个“半消息”到RocketMQ Broker。这个半消息消费者是看不到的，它就像一个待定的订单，等着你确认。

接着，是本地事务执行器。当半消息发送成功后，Broker会回调生产者的executeLocalTransaction方法。在这个方法里，你需要执行你的核心业务逻辑，比如更新数据库、调用其他服务等等，这些操作共同构成了你的“本地事务”。这个本地事务的成功与否，直接决定了半消息的命运。

然后，是本地事务状态回查器。这是TransactionListener接口中的checkLocalTransaction方法。这个组件非常关键，它是整个事务消息机制的“兜底”保障。如果生产者在执行完本地事务后，因为某些原因（比如网络中断、程序崩溃）没能及时通知Broker事务的最终状态（提交或回滚），或者executeLocalTransaction返回了UNKNOW状态，Broker就会主动来调用这个方法，询问你那个半消息对应的本地事务到底成功了没。你在这个方法里，需要去查询你的业务系统，获取本地事务的真实状态，然后告诉Broker。

最后，当本地事务确认成功并提交消息后，消息才对消费者（Consumer）可见。消费者会像处理普通消息一样去消费它。如果本地事务失败或者被回滚，那么半消息就会被删除，消费者永远也看不到这条消息。

整个流程可以概括为：发半消息 -> 执行本地事务 -> 根据结果提交/回滚消息 ->（必要时）Broker回查 -> 最终消费。 每一个环节都环环相扣，确保了数据的一致性。

实践RocketMQ事务消息时，有哪些不容忽视的细节和“雷区”？

说实话，RocketMQ事务消息虽然好用，但实际落地的时候，还是有些“坑”需要注意，否则可能会遇到一些意想不到的问题。

一个常见的“雷区”是checkLocalTransaction方法的设计。这个方法是事务回查的关键，它必须能够准确、幂等地查询到本地事务的最终状态。我见过不少项目，这里只是简单地返回UNKNOW或者依赖一些不靠谱的内存状态，一旦生产者重启或者数据丢失，就彻底乱了。正确的做法是，这里应该去查询你的持久化存储（比如数据库），根据消息的Key或者其他业务唯一标识，来判断本地事务是否真的成功。如果业务流程复杂，可能还需要考虑事务的中间状态，甚至引入超时机制。

再来就是本地事务执行的超时与异常处理。executeLocalTransaction方法里，如果你的业务代码执行时间过长，或者抛出了未捕获的异常，这都会影响事务消息的最终状态。RocketMQ Broker对事务回查是有时间间隔和次数限制的。如果你的本地事务长时间不返回状态，或者频繁抛异常，可能导致Broker反复回查，甚至最终判断为失败并回滚消息，这显然不是我们希望的。所以，本地事务的代码要健壮，异常要妥善处理，并确保能在合理的时间内返回状态。

还有一点，虽然不直接是事务消息的问题，但紧密相关——消费者的幂等性。即使事务消息确保了生产端和消息的一致性，但由于网络重传、Broker故障恢复等原因，消费者仍然有可能收到重复的消息。所以，你的消费者必须设计成幂等的，即多次消费同一条消息，业务结果也是一致的。这通常通过业务唯一ID和状态机的设计来实现。

最后，Producer Group的命名和管理也常常被忽视。事务回查机制是基于Producer Group来识别生产者的。如果你的Producer Group不唯一，或者在不同实例上使用了不同的组名，Broker可能无法正确地找到对应的生产者进行回查，导致事务消息一直处于UNKNOW状态，最终可能被回滚。所以，Producer Group务必保持唯一且稳定。同时，消息的Key也应该尽可能地使用业务唯一ID，这对于后续的问题排查和状态核对非常有帮助。这些细节，虽然看似不起眼，但往往是决定系统稳定性的关键所在。

今天关于《RocketMQ事务消息实现教程详解》的内容介绍就到此结束，如果有什么疑问或者建议，可以在golang学习网公众号下多多回复交流；文中若有不正之处，也希望回复留言以告知！