基于RocketMQ实现分布式事务的方法

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.*;

public class RocketMqTransactionDemo {
	public static void main(String[] args) throws Exception {
		// 创建事务消息生产者
		TransactionMQProducer producer = new TransactionMQProducer("transaction_producer");
		producer.setNamesrvAddr("127.0.0.1:9876");

		// 设置事务监听器
		TransactionListener transactionListener = new MyTransactionListener();
		producer.setTransactionListener(transactionListener);

		// 设置事务回查的线程池，可以不必设置，如果不设置也会默认生成一个
		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.start();

		// 发送事务消息
		Message message = new Message("transaction_topic", "test_tag", "test_key", "Hello RocketMQ".getBytes());
		producer.sendMessageInTransaction(message, null);

		// 关闭生产者
		producer.shutdown();
	}
}

/**
 * 事务监听器
 */
class MyTransactionListener implements TransactionListener {
	@Override
	public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
		// 执行本地事务操作
		System.out.println("执行本地事务操作，消息内容：" + new String(msg.getBody()));
		return LocalTransactionState.COMMIT_MESSAGE; // 提交事务，允许消费者消费该消息
		// return LocalTransactionState.ROLLBACK_MESSAGE;// 回滚事务，消息将被丢弃不允许消费。
		// return LocalTransactionState.UNKNOW;// 暂时无法判断状态，等待固定时间以后Broker端根据回查规则向生产者进行消息回查。
	}

	@Override
	public LocalTransactionState checkLocalTransaction(MessageExt msg) {
		// 检查本地事务状态
		System.out.println("检查本地事务状态，消息内容：" + new String(msg.getBody()));
		return LocalTransactionState.COMMIT_MESSAGE;
	}
}