Java操作RocketMQ事务消息的教程

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;
    }
}