在《RocketMQ实战入门》里我们入门了基本的RocketMQ消息发布和消费，并封装了一个简单的util包，现在我们来看一下如何使用RocketMQ的事务消息来解决分布式事务问题。
事务消息基本流程
说明，RocketMQ来实现分布式事务主要基于的是BASE理论，即基本可用、软状态、最终一致性。属于刚性事务与柔性事务中的后者，性能较好，但取的是最终一致性。流程类似于2PC，但是个异步过程。
流程如下：
Producer发送半消息给RocketMQ，收到回复后开始执行自己本地的事务。半消息不会被消费。
执行本地事务并将执行的结果发送通知RocketMQ，MQ根据结果来确定之前的半消息是提交还是丢弃。
如果半消息提交，RocketMQ将负责确保该消息被Consumer消费，Consumer消费到消息即执行自己的事务。
异常情况：
如果Producer没有将本地事务结果发送给MQ，此时事务处于UNKNOW状态，则MQ每60s会通过Producer提供的回调接口来反查其事务执行结果来决定半消息是提交还是丢弃。默认会执行此回调15次，如果还是UNKNOW则丢弃半消息，并打印错误日志。
如果Producer发送完半消息之后挂了，MQ会去同一个group里的其他Producer调用回调接口反查事务执行结果。
代码编写
RocketMQ发送事务消息主要是使用TransactionMQProducer这个Producer。然后实现TransactionListener接口，并将该Listener绑定到Producer上。
TransactionMQProducer与普通消息的DefaultMQProducer比较类似，而TransactionListener接口需要实现executeLocalTransaction()和checkLocalTransaction()两个方法，分别写半消息发送成功后的本地事务逻辑以及供MQ回调的反查事务结果逻辑。
参考官方例子https://github.com/apache/rocketmq/blob/master/docs/cn/RocketMQ_Example.md，实际使用中一般应该是在executeLocalTransaction()根据业务逻辑执行情况直接返回明确的事务结果是Commit还是Rollback的。这里为了测试故意在executeLocalTransaction()返回事务结果UNKNOW，让MQ执行反查逻辑回调checkLocalTransaction()方法来获得事务结果。代码如下：(部分代码在前文《RocketMQ实战入门》中有)
TransactionProducer：
/**
* 事务消息Producer
* 用于发送事务消息，实现分布式事务
*
* 跟普通消息用的Producer不同，事务Producer没法全局用个单例，
* 因为需要绑定TransactionListener的业务处理逻辑。
* */
@Slf4j
@Component
public class TransactionMsgProducer {
@Value("${rocketmq.url}")
private String mqurl;
@Value("${rocketmq.accessKey}")
private String accessKey;
@Value("${rocketmq.secretKey}")
private String secretKey;
@Value("${rocketmq.producergroup.name}")
private String producerGroupName;
private TransactionMQProducer producer;
public TransactionSendResult publish(EventMessage eventMsg) {
try {
Message msg = new Message(eventMsg.getTopic(), eventMsg.getTag(), eventMsg.getMsgId(), JSON.toJSONString(eventMsg).getBytes("utf-8"));
TransactionSendResult result = producer.sendMessageInTransaction(msg, null);
return result;
} catch (Exception e) {
log.error("事务消息发送失败" + e.getMessage(), e);
e.printStackTrace();
}
return null;
}
@PostConstruct
public void init() {
producer = new TransactionMQProducer(producerGroupName + "-trans", getAclRPCHook());
producer.setNamesrvAddr(mqurl);
TransactionListener transactionListener = new MyTransactionListener();
producer.setTransactionListener(transactionListener);
try {
producer.start();
log.info("RocketMQ客户端事务producer初始化...");
} catch (MQClientException e) {
e.printStackTrace();
}
}
@PreDestroy
public void shutdown() {
if(producer != null)
producer.shutdown();
}
private RPCHook getAclRPCHook() {
return new AclClientRPCHook(new SessionCredentials(accessKey, secretKey));
}
}
TransactionListener：
/**
* 半消息发送成功则执行executeLocalTransaction
* RocketMQ回调checkLocalTransaction来确认Producer本地的事务执行状态
* */
@Slf4j
public class MyTransactionListener implements TransactionListener{
private AtomicInteger transactionIndex = new AtomicInteger(0);
private ConcurrentHashMap<String, Integer> localTransactions = new ConcurrentHashMap<>();
@Override
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
log.info("Producer开始执行本地事务...");
int value = transactionIndex.getAndIncrement();
localTransactions.put(msg.getTransactionId(), value % 3);
log.info("事务ID:{},执行状态{}, 但统一返回RocketMQ事务状态为UNKNOW", msg.getTransactionId(), value % 3);
return LocalTransactionState.UNKNOW;
}
@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
Integer state = localTransactions.get(msg.getTransactionId());
log.info("事务ID:{}执行后返回MQ状态为UNKNOW，触发RocketMQ回调查询事务状态机制，"
+ "MQ调用Producer提供的回调接口查询事务状态为{}", msg.getTransactionId(), state);
if(null != state) {
switch(state) {
case 0:
log.info("从肥兔子账户-300操作还没执行完");
return LocalTransactionState.UNKNOW;
case 1:
log.info("已经成功从肥兔子账户-300");
return LocalTransactionState.COMMIT_MESSAGE;
case 2:
log.info("从肥兔子账户-300失败");
return LocalTransactionState.ROLLBACK_MESSAGE;
}
}
return LocalTransactionState.COMMIT_MESSAGE;
}
}
消费者，也即分布式事务的其他参与者：
@Slf4j
@Component
@MsgConsumer(consumerGroup = "testapp-consumer-group", tag = "douchuzi", topic = "test-trans-topic")
public class TestConsumer implements MQMsgHandler{
@Override
public void handleMsg(List<EventMessage> msgs) {
for(EventMessage msg : msgs) {
log.info(JSON.toJSONString(msg));
}
}
}
测试代码：
EventMessage msg = new EventMessage();
msg.setTopic("test-trans-topic");
msg.setTag("douchuzi");
msg.setMsgId("123");
msg.setProducerGroup("testapp-producer-group");
msg.setMsgBody("给豆畜子账户+300块");
msg.setPublishTime(LocalDateTime.now().toString());
TransactionSendResult result = transactionMsgProducer.publish(msg);
log.info("肥兔子给豆畜子转账的事务消息发送完毕，"
+ "消息ID:{}, 事务ID{},消息发送状态:{}，分布式事务开始！",
result.getMsgId(),
result.getTransactionId(),
result.getSendStatus());
测试执行结果分3种情况，执行3次一个循环。
反查本地事务UNKNOW：
2022-02-01 22:48:37.172 INFO 12388 --- [nio-8080-exec-2] c.w.controller.MyTransactionListener : Producer开始执行本地事务...
2022-02-01 22:48:37.173 INFO 12388 --- [nio-8080-exec-2] c.w.controller.MyTransactionListener : 事务ID:7F00000130644E0E2F2A04E502240000,执行状态0, 但统一返回RocketMQ事务状态为UNKNOW
2022-02-01 22:48:37.174 INFO 12388 --- [nio-8080-exec-2] com.wangan.controller.WanganController : 肥兔子给豆畜子转账的事务消息发送完毕，消息ID:7F00000130644E0E2F2A04E502240000, 事务IDnull,消息发送状态:SEND_OK，分布式事务开始！
2022-02-01 22:49:05.702 INFO 12388 --- [pool-1-thread-1] c.w.controller.MyTransactionListener : 事务ID:7F00000130644E0E2F2A04E502240000执行后返回MQ状态为UNKNOW，触发RocketMQ回调查询事务状态机制，MQ调用Producer提供的回调接口查询事务状态为0
2022-02-01 22:49:05.702 INFO 12388 --- [pool-1-thread-1] c.w.controller.MyTransactionListener : 从肥兔子账户-300操作还没执行完
2022-02-01 22:50:05.700 INFO 12388 --- [pool-1-thread-1] c.w.controller.MyTransactionListener : 事务ID:7F00000130644E0E2F2A04E502240000执行后返回MQ状态为UNKNOW，触发RocketMQ回调查询事务状态机制，MQ调用Producer提供的回调接口查询事务状态为0
2022-02-01 22:50:05.700 INFO 12388 --- [pool-1-thread-1] c.w.controller.MyTransactionListener : 从肥兔子账户-300操作还没执行完
2022-02-01 22:51:05.697 INFO 12388 --- [pool-1-thread-1] c.w.controller.MyTransactionListener : 事务ID:7F00000130644E0E2F2A04E502240000执行后返回MQ状态为UNKNOW，触发RocketMQ回调查询事务状态机制，MQ调用Producer提供的回调接口查询事务状态为0
2022-02-01 22:51:05.697 INFO 12388 --- [pool-1-thread-1] c.w.controller.MyTransactionListener : 从肥兔子账户-300操作还没执行完
发送完半消息之后执行本地事务并回复MQ事务执行结果unknow，这样MQ会反查回调接口来确认事务结果，1分钟查1次，默认15次仍无法确认事务结果则丢弃半消息，这里为了测试方便改成了3次（修改broker.conf里的transactionCheckMax=3）。
反查本地事务COMMIT_MESSAGE：
2022-02-01 22:53:27.323 INFO 12388 --- [nio-8080-exec-5] c.w.controller.MyTransactionListener : Producer开始执行本地事务...
2022-02-01 22:53:27.323 INFO 12388 --- [nio-8080-exec-5] c.w.controller.MyTransactionListener : 事务ID:7F00000130644E0E2F2A04E96F880001,执行状态1, 但统一返回RocketMQ事务状态为UNKNOW
2022-02-01 22:53:27.323 INFO 12388 --- [nio-8080-exec-5] com.wangan.controller.WanganController : 肥兔子给豆畜子转账的事务消息发送完毕，消息ID:7F00000130644E0E2F2A04E96F880001, 事务IDnull,消息发送状态:SEND_OK，分布式事务开始！
2022-02-01 22:54:05.708 INFO 12388 --- [pool-1-thread-1] c.w.controller.MyTransactionListener : 事务ID:7F00000130644E0E2F2A04E96F880001执行后返回MQ状态为UNKNOW，触发RocketMQ回调查询事务状态机制，MQ调用Producer提供的回调接口查询事务状态为1
2022-02-01 22:54:05.708 INFO 12388 --- [pool-1-thread-1] c.w.controller.MyTransactionListener : 已经成功从肥兔子账户-300
2022-02-01 22:54:05.718 INFO 12388 --- [MessageThread_6] com.wangan.controller.TestConsumer : {"msgBody":"给豆畜子账户+300块","msgId":"123","producerGroup":"testapp-producer-group","publishTime":"2022-02-01T22:53:27.304","tag":"douchuzi","topic":"test-trans-topic"}
MQ反查事务结果，获知事务成功，MQ提交半消息，消息即可被消费者正常消费并执行其事务。分布式事务完成。
反查本地事务ROLLBACK_MESSAGE：
2022-02-01 22:54:59.306 INFO 12388 --- [nio-8080-exec-8] c.w.controller.MyTransactionListener : Producer开始执行本地事务...
2022-02-01 22:54:59.306 INFO 12388 --- [nio-8080-exec-8] c.w.controller.MyTransactionListener : 事务ID:7F00000130644E0E2F2A04EAD6DD0002,执行状态2, 但统一返回RocketMQ事务状态为UNKNOW
2022-02-01 22:54:59.307 INFO 12388 --- [nio-8080-exec-8] com.wangan.controller.WanganController : 肥兔子给豆畜子转账的事务消息发送完毕，消息ID:7F00000130644E0E2F2A04EAD6DD0002, 事务IDnull,消息发送状态:SEND_OK，分布式事务开始！
2022-02-01 22:56:05.718 INFO 12388 --- [pool-1-thread-1] c.w.controller.MyTransactionListener : 事务ID:7F00000130644E0E2F2A04EAD6DD0002执行后返回MQ状态为UNKNOW，触发RocketMQ回调查询事务状态机制，MQ调用Producer提供的回调接口查询事务状态为2
2022-02-01 22:56:05.718 INFO 12388 --- [pool-1-thread-1] c.w.controller.MyTransactionListener : 从肥兔子账户-300失败
反查事务结果为事务失败，MQ删除半消息。
总结
RocketMQ事务消息可以实现柔性分布式事务，确保最终一致性，不锁定数据库资源，适合高并发场景。
