一、RocketMq事务消息流程：
1、首先会向broker发送一个预请求消息，消费者不可见
2、回调执行本地事务（比如操作数据库）
3、事务执行成功后，再次发送消息给broker，告诉broker事务执行成功这个消息要提交，让消费者可见。如果本地事务执行超时，会返回一个unknow，broker会发送一个消息回查，检查消息是否执行成功。
二、RocketMq事务消息实例：
1、引入rocketMq相关的依赖：
<dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-client</artifactId> <version>4.4.0</version> </dependency>
2、创建一个TransactionProducer类：
public class TransactionProducer {
public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException, UnsupportedEncodingException {
//创建生产者并制定组名
TransactionMQProducer producer = new TransactionMQProducer("rocketMQ_transaction_producer_group");
//2.指定Nameserver地址
producer.setNamesrvAddr("192.168.***.***:9876");
//3、指定消息监听对象用于执行本地事务和消息回查
TransactionListener listener = new TransactionListenerImol();
producer.setTransactionListener(listener);
//4、线程池
ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread thread = newThread(r);
thread.setName("client-tanscation-msg-check-thread");
return thread;
}
});
producer.setExecutorService(executorService);
//5、启动producer
producer.start();
//6.创建消息对象，指定主题Topic、Tag和消息体 String topic, String tags, String keys, byte[] body
Message message = new Message("Topic_transaction_demo", //主题
"Tags", //主要用于消息过滤
"Key_1", //消息唯一值
("hello-transaction").getBytes(RemotingHelper.DEFAULT_CHARSET));
//7、发送事务消息
TransactionSendResult result = producer.sendMessageInTransaction(message, "hello-transaction");
producer.shutdown();
}
}
3、发送事务消息还需要一个事务监听对象，它实现TransactionListener 接口，其中有两个方法作用分别是执行本地事务和消息回查：
public class TransactionListenerImol implements TransactionListener {
//存储事务状态信息 key:事务id value：当前事务执行的状态
private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();
//执行本地事务
@Override
public LocalTransactionState executeLocalTransaction(Message message, Object o) {
//事务id
String transactionId = message.getTransactionId();
//0:执行中，状态未知 1：执行成功 2：执行失败
localTrans.put(transactionId, 0);
//业务执行，本地事务，service
System.out.println("hello-demo-transaction");
try {
System.out.println("正在执行本地事务---");
Thread.sleep(60000*2);
System.out.println("本地事务执行成功---");
localTrans.put(transactionId, 1);
} catch (InterruptedException e) {
e.printStackTrace();
localTrans.put(transactionId, 2);
return LocalTransactionState.ROLLBACK_MESSAGE;
}
return LocalTransactionState.COMMIT_MESSAGE;
}
//消息回查
@Override
public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
//获取对应事务的状态信息
String transactionId = messageExt.getTransactionId();
//获取对应事务id执行状态
Integer status = localTrans.get(transactionId);
//消息回查
System.out.println("消息回查---transactionId:" + transactionId + "状态:" + status);
switch (status) {
case 0:
return LocalTransactionState.UNKNOW;
case 1:
return LocalTransactionState.COMMIT_MESSAGE;
case 2:
return LocalTransactionState.ROLLBACK_MESSAGE;
}
return LocalTransactionState.UNKNOW;
}
}
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持脚本之家。
