public enum LocalTransactionState {
    COMMIT_MESSAGE,
    ROLLBACK_MESSAGE,
    UNKNOW,
}

package com.morris.rocketmq.transaction;
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.*;
import java.util.concurrent.atomic.AtomicInteger;
import static com.morris.rocketmq.util.Contant.NAME_SERVER_ADDRESS;
/**
 * 事務(wù)消息生產(chǎn)者
 */
public class TransactionProducer {
    public static void main(String[] args) throws Exception {
        TransactionMQProducer producer = new TransactionMQProducer("transaction-producer-demo");
        producer.setNamesrvAddr(NAME_SERVER_ADDRESS);
        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<>(2000), r -> {
            Thread thread = new Thread(r);
            thread.setName("client-transaction-msg-check-thread");
            return thread;
        });
        producer.setExecutorService(executorService);
        // 指定事務(wù)會(huì)查的實(shí)現(xiàn)類
        producer.setTransactionListener(new TransactionListener() {
            private final AtomicInteger transactionIndex = new AtomicInteger(0);
            private final ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                int value = transactionIndex.getAndIncrement();
                System.out.println(Thread.currentThread().getName()+  "-executeLocalTransaction:" + new String(msg.getBody()) + ",value=" + value);
                int status = value % 3;
                localTrans.put(msg.getTransactionId(), status);
                return LocalTransactionState.UNKNOW;
            }
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.out.println(Thread.currentThread().getName()+  "-checkLocalTransaction:" + new String(msg.getBody()));
                Integer status = localTrans.get(msg.getTransactionId());
                if (null != status) {
                    switch (status) {
                        case 0:
                            return LocalTransactionState.COMMIT_MESSAGE;
                        case 1:
                            return LocalTransactionState.UNKNOW;
                        case 2:
                            return LocalTransactionState.ROLLBACK_MESSAGE;
                    }
                }
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });
        producer.start();
        for(int i = 0; i < 10; i++) {
            Message message = new Message("TransactionTopic", ("transactionDemo" + i).getBytes());
            // 發(fā)送事務(wù)消息
            producer.sendMessageInTransaction(message, i);
            System.out.println(message);
        }
    }
}

Message message = new Message();
message.putUserProperties(MessageConst.PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS, "60");

package com.morris.rocketmq.transaction;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import static com.morris.rocketmq.util.Contant.NAME_SERVER_ADDRESS;
/**
 * 事務(wù)消息消費(fèi)者
 */
public class TranscationConsumer {
    public static void main(String[] args) throws Exception {
        // 實(shí)例化消息生產(chǎn)者,指定組名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("transaction-consumer-group");
        // 指定Namesrv地址信息.
        consumer.setNamesrvAddr(NAME_SERVER_ADDRESS);
        // 訂閱Topic
        consumer.subscribe("TransactionTopic", "*");
        //負(fù)載均衡模式消費(fèi)
        consumer.setMessageModel(MessageModel.CLUSTERING);
        // 注冊(cè)回調(diào)函數(shù)，處理消息
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            try {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
            } catch (Exception e) {
                e.printStackTrace();
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        //啟動(dòng)消息者
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}