RocketMQ——事务消息详解编程语言

事务消息就是事务里参杂着发消息的动作

如果一个事务里牵扯到发消息的操作，那么一旦消息被消费了，想要回滚，这时就变的很难

所以在没有commit的时候，消息只会暂时发送存在broker不会被消费，当commit成功的时候，在MQ中会将这个消息设置为真正可用，这时comusmer才会消费消息，如果执行的是rollback，在MQ中就会把这个消息撤回（就是两阶段提交的机制）

下面是事务消息执行的顺序图

HFMessage半消息可以为一种向broker发送的标记，与真正的消息内容完全没有关系，仅用于在broker维持该消息的状态，到底消息是可用还是不可用。

请结合以下具体代码理解上述图文

import org.apache.rocketmq.client.producer.*; 
import org.apache.rocketmq.common.message.Message; 
import org.apache.rocketmq.common.message.MessageExt; 
import org.apache.rocketmq.remoting.common.RemotingHelper; 
/** 
* RocketMQ——事务消息 
* 
* @author Song X. 
* @date 2020/03/04 
*/ 
public class Producer {
 
public static void main(String[] args) throws Exception {
 
TransactionMQProducer producer = new TransactionMQProducer("TransactionGroup1"); 
producer.setNamesrvAddr("192.168.255.131:9876"); 
producer.setTransactionListener(new TransactionListener() {
 
/** 
*  这是producer执行的本地事务 
*  begin transaction: 
*      这里面的方法应该是同步的按顺序执行 
*      a() 
*      b() 
*      c() -> producer发消息 
*      d() 
*  end transaction 
* 上面的方法应使用try-catch，若发生异常则return ROLLBACK_MESSAGE 
* 若全部成功用COMMIT_MESSAGE 
* 
*      try{ 
*          a() 
*          b() 
*          c() 
*          d() 
*      } catch () { 
*          return LocalTransactionState.ROLLBACK_MESSAGE 
*      } 
*      return LocalTransactionState.COMMIT_MESSAGE 
* 
* 在这个方法中一搬不用UNKNOWN 
* 
* @param msg 就是producer send的消息 
* @param arg 
* @return  LocalTransactionState.COMMIT_MESSAGE 表示执行事务成功，确认提交 
*          LocalTransactionState.ROLLBACK_MESSAGE 表示回滚消息，broker端删除半消息(HFMessage) 
*          LocalTransactionState.UNKNOWN 表示为未知状态，等待broker回调查看 
*/ 
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
 
System.out.println("------executeLocalTransaction------"); 
System.out.println("msgBody: " + new String(msg.getBody())); 
//transactionId就是messageId 
System.out.println("msgTransactionID: " + msg.getTransactionId()); 
return LocalTransactionState.UNKNOW; 
} 
/** 
* 这是broker执行的回调函数 
* Broker端回调检查本地事务的执行状态，也就是检查executeLocalTransaction方法的执行情况 
* 
* if(...){ 
* 
*     ... //如果检查了XXX，表示事务成功了 
*     return LocalTransactionState.COMMIT_MESSAGE; 
* 
* } else if(...) { 
* 
*     ... //如果检查了XXX，表示可能executeLocalTransaction还在执行，等会回来再检查 
*     return LocalTransactionState.UNKNOWN 
* 
* } else { 
* 
*     //事务执行失败了，回滚，回滚的是HFMessage 
*     return LocalTransactionState.ROLLBACK_MESSAGE 
* 
* } 
* 
* @param msg 就是producer send的消息 
* @return LocalTransactionState.COMMIT_MESSAGE; 
*/ 
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
 
System.out.println("------checkLocalTransaction------"); 
System.out.println("msgBody: " + new String(msg.getBody())); 
//transactionId就是messageId 
System.out.println("msgTransactionID: " + msg.getTransactionId()); 
return LocalTransactionState.UNKNOW; 
} 
}); 
producer.start(); 
Message msg = new Message("myTransaction","TagA", "事务消息测试".getBytes( 
RemotingHelper.DEFAULT_CHARSET 
)); 
TransactionSendResult transactionSendResult = producer.sendMessageInTransaction(msg, null); 
System.out.println("transactionSendResult: " + transactionSendResult); 
//      producer.shutdown(); 
} 
} 

可在两个函数中都是用UNKNOWN观察checkLocalTransaction函数的回调运行机制