深入理解RocketMQ事务源码--TransactionMQProducer、TransactionListener

朴灿烈づ我的快乐病毒、 2022-12-28 11:09 164阅读 0赞

前言

上篇文章，介绍了RocketMQ消息类型支持事务消息，常见用在分布式系统中，保证数据的一致性。接下来一起去走进 TransactionMQProducer、TransactionListener俩个核心类

一.基于apache-rocketmq事务消息的实现

（1） pom.xml配置

  1. <dependency>
  2.             <groupId>org.springframework.boot</groupId>
  3.             <artifactId>spring-boot-starter</artifactId>
  4.         </dependency>
  6.         <dependency>
  7.             <groupId>org.springframework.boot</groupId>
  8.             <artifactId>spring-boot-starter-web</artifactId>
  9.         </dependency>
  11.         <!--lombok插件 -->
  12.         <dependency>
  13.             <groupId>org.projectlombok</groupId>
  14.             <artifactId>lombok</artifactId>
  15.             <scope>provided</scope>
  16.         </dependency>
  18.         <!--RocketMQ-->
  19.         <dependency>
  20.             <groupId>org.apache.rocketmq</groupId>
  21.             <artifactId>rocketmq-client</artifactId>
  22.             <version>4.3.0</version>
  23.         </dependency>

（2）事务消息生产者配置

  1. import org.apache.rocketmq.client.exception.MQClientException;
  2. import org.apache.rocketmq.client.producer.TransactionMQProducer;
  3. import org.springframework.stereotype.Component;
  5. @Component
  6. public class PayProducer {
  7.     /**
  8.      * 生产组,消费者通过订阅发布，在同一个组内
  9.      */
  10.     private String producerGroup = "pay_group";
  11.     /**
  12.      * 端口
  13.      */
  14.     private String nameServer = "127.0.0.1:9876";
  15.     //事务监听,其中rocketMQLisenter实现TransactionListener接口
  16.     private RocketMQLisenter rocketMQLisenter;
  17.     //事务消息生产者配置
  18.     private TransactionMQProducer producer;
  20.     public PayProducer() {
  21.         producer = new TransactionMQProducer(producerGroup);
  22.         rocketMQLisenter = new RocketMQLisenter();
  23.         // 指定nameServer地址,多个地址之间以 ; 隔开
  24.         producer.setNamesrvAddr(nameServer);
  25.         producer.setTransactionListener(rocketMQLisenter);
  26.         start();
  27.     }
  29.     public TransactionMQProducer getProducer() {
  30.         return producer;
  31.     }
  33.     /**
  34.      * 对象在使用之前必须调用一次,并且只能初始化一次
  35.      */
  36.     public void start() {
  37.         try {
  38.             this.producer.start();
  39.         } catch (MQClientException e) {
  40.             e.printStackTrace();
  41.         }
  42.     }
  44.     /**
  45.      * 一般在应用上下文,使用上下文监听器,进行关闭
  46.      */
  47.     public void shutdown() {
  48.         producer.shutdown();
  49.     }
  51. }

（3）事务监听器接口实现

  1. import com.alibaba.fastjson.JSONObject;
  2. import lombok.extern.slf4j.Slf4j;
  3. import org.apache.rocketmq.client.producer.LocalTransactionState;
  4. import org.apache.rocketmq.client.producer.TransactionListener;
  5. import org.apache.rocketmq.common.message.Message;
  6. import org.apache.rocketmq.common.message.MessageExt;
  8. @Slf4j
  9. public class RocketMQLisenter implements TransactionListener {
  10.     @Override
  11.     public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
  12.         log.info("=========本地事务开始执行=============");
  13.         String message = new String(msg.getBody());
  14.         int type = 0;
  15.         // 事务消息id，唯一 
  16.         String tid = msg.getTransactionId(); 
  17.         //todo 为了解决重复消息《==》幂等性，可以将tid放入redis，并设置状态,消费端，每次从redis中获取事务id，根据状态并判断是否被消费过
  18.         //模拟执行本地事务begin=======
  19.         /**
  20.          * 本地事务执行会有三种可能
  21.          * 1、commit 成功
  22.          * 2、Rollback 失败
  23.          * 3、网络等原因服务宕机收不到返回结果
  24.          */
  25.         log.info("本地事务执行参数----------------------");
  26.         //模拟执行本地事务end========
  27.         //TODO 实际开发下面不需要我们手动返回，而是根据本地事务执行结果自动返回
  28.         //1、二次确认消息，然后消费者可以消费
  29.         if (type == 0) {
  30.             return LocalTransactionState.COMMIT_MESSAGE;
  31.         }
  32.         //2、回滚消息，Broker端会删除半消息
  33.         if (type == 1) {
  34.             return LocalTransactionState.ROLLBACK_MESSAGE;
  35.         }
  36.         //3、Broker端会进行回查消息
  37.         if (type == 2) {
  38.             return LocalTransactionState.UNKNOW;
  39.         }
  40.         return LocalTransactionState.COMMIT_MESSAGE;
  41.     }
  43.     /**
  44.      * 当executeLocalTransaction ，返回LocalTransactionState.UNKNOW 调用此方法
  45.      * @param messageExt
  46.      * @return
  47.      */
  48.     public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
  49.         log.info("==========回查接口=========");
  50.         String key = messageExt.getKeys();
  51.         //TODO 1、必须根据key先去检查本地事务消息是否完成。
  52.         /**
  53.          * 因为有种情况就是：上面本地事务执行成功了，但是return LocalTransactionState.COMMIT_MESSAG的时候
  54.          * 服务挂了，那么最终 Brock还未收到消息的二次确定，还是个半消息 ，所以当重新启动的时候还是回调这个回调接口。
  55.          * 如果不先查询上面本地事务的执行情况 直接在执行本地事务，那么就相当于成功执行了两次本地事务了。
  56.          */
  57.         // TODO 2、这里返回要么commit 要么rollback。没有必要在返回 UNKNOW
  58.         return LocalTransactionState.COMMIT_MESSAGE;
  59.     }
  60. }

（4）消费者配置

  1. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
  2. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
  3. import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
  4. import org.apache.rocketmq.client.exception.MQClientException;
  5. import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
  6. import org.apache.rocketmq.common.message.Message;
  7. import org.springframework.stereotype.Component;
  8. import java.io.UnsupportedEncodingException;
  10. @Component
  11. public class PayConsumer {
  13.     private DefaultMQPushConsumer consumer;
  15.     private String consumerGroup = "pay_group";
  17.     public PayConsumer() throws MQClientException {
  18.         consumer = new DefaultMQPushConsumer(consumerGroup);
  19.         consumer.setNamesrvAddr(Topic.NAME_SERVER);
  20.         // 设置消费地点,从最后一个进行消费(其实就是消费策略)
  21.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
  22.         // 订阅主题的哪些Topic,及标签
  23.         consumer.subscribe("order_topic", "*");
  24.         // 注册监听器
  25.         consumer.registerMessageListener((MessageListenerConcurrently)
  26.                 (msgs, context) -> {
  27.                     try {
  28.                         // 获取Message
  29.                         Message msg = msgs.get(0);
  30.                         //获取事务id,并从redis获取事务的状态，判断是否消费过。
  31.                         String tid = msg.getTransactionId();
  32.                         System.out.printf("%s Receive New Messages: %s %n",
  33.                                 Thread.currentThread().getName(), new String(msgs.get(0).getBody()));
  34.                         String topic = msg.getTopic();
  35.                         String body = new String(msg.getBody(), "utf-8");
  36.                         // 标签
  37.                         String tags = msg.getTags();
  38.                         String keys = msg.getKeys();
  39.                         System.out.println("topic=" + topic + ", tags=" + tags + ",keys=" + keys + ", msg=" + body);
  40.                         return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  41.                     } catch (UnsupportedEncodingException e) {
  42.                         e.printStackTrace();
  43.                         return ConsumeConcurrentlyStatus.RECONSUME_LATER;
  44.                     }
  45.                 });
  46.         consumer.start();
  47.         System.out.println("Consumer Listener");
  48.     }
  50. }

（5）通过生产者发送事务消息

  1. import org.apache.rocketmq.client.exception.MQBrokerException;
  2. import org.apache.rocketmq.client.exception.MQClientException;
  3. import org.apache.rocketmq.client.producer.SendResult;
  4. import org.apache.rocketmq.common.message.Message;
  5. import org.apache.rocketmq.remoting.exception.RemotingException;
  6. import org.springframework.beans.factory.annotation.Autowired;
  7. import org.springframework.web.bind.annotation.GetMapping;
  8. import org.springframework.web.bind.annotation.ResponseBody;
  9. import org.springframework.web.bind.annotation.RestController;
  11. @RestController
  12. public class PayController {
  13.     @Autowired
  14.     private PayProducer payProducer;
  15.     private static final String topic = "order_topic";
  17.     @GetMapping("/pay")
  18.     @ResponseBody
  19.     public void callback(int type) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
  20.         // 创建消息  主题   二级分类   消息内容好的字节数组
  21.         Message message = new Message(topic, "taga", ("rocketMQ").getBytes());
  22.         //调用TransactionMQProducer的sendMessageInTransaction事务方法，并获取发送结果
  23.         SendResult send = payProducer.getProducer().sendMessageInTransaction(message,type);
  24.     }
  26. }

二.TransactionMQProducer、TransactionListener源码阅读

1.TransactionMQProducer，可以看出继承DefaultMQProducer类，核心方法是sendMessageInTransaction(Message msg,Object org)

  1. public class TransactionMQProducer extends DefaultMQProducer {
  2.     private TransactionListener transactionListener;
  3.     private ExecutorService executorService;
  5.     public TransactionMQProducer() {
  6.     }
  8.     public TransactionMQProducer(String producerGroup) {
  9.         super(producerGroup);
  10.     }
  12.     public TransactionMQProducer(String producerGroup, RPCHook rpcHook) {
  13.         super(producerGroup, rpcHook);
  14.     }
  16.     public void start() throws MQClientException {
  17.         this.defaultMQProducerImpl.initTransactionEnv();
  18.         super.start();
  19.     }
  21.     public void shutdown() {
  22.         super.shutdown();
  23.         this.defaultMQProducerImpl.destroyTransactionEnv();
  24.     }
  26.     public TransactionSendResult sendMessageInTransaction(Message msg, Object arg) throws MQClientException {
  27.         if (null == this.transactionListener) {
  28.             throw new MQClientException("TransactionListener is null", (Throwable)null);
  29.         } else {
  30.             return this.defaultMQProducerImpl.sendMessageInTransaction(msg, this.transactionListener, arg);
  31.         }
  32.     }
  34.     public TransactionListener getTransactionListener() {
  35.         return this.transactionListener;
  36.     }
  38.     public void setTransactionListener(TransactionListener transactionListener) {
  39.         this.transactionListener = transactionListener;
  40.     }
  42.     public ExecutorService getExecutorService() {
  43.         return this.executorService;
  44.     }
  46.     public void setExecutorService(ExecutorService executorService) {
  47.         this.executorService = executorService;
  48.     }
  49. }

2.DefaultMQProducer.sendMessageInTransaction方法具体实现，

（1）可以看出先获取消息发送到broker的状态（可以理解成半状态，等待本地事务执行结果，执行commit或在rollback），没有成功则返回事务状态设置为LocalTransactionState.ROLLBACK_MESSAGE；发送成功，则获取消息事务的ID（每个消息都会由一个transactionId）

（2）接着调用TransactionListener.executeLocalTransaction方法（上面已经沾贴了事务监听器的实现代码），执行本地事务。

  1. /**
  2.  * 本地事务执行会有三种可能
  3.  * 1、commit 成功 ，对应 return LocalTransactionState.COMMIT_MESSAGE;
  4.  * 2、Rollback 失败，对应 return LocalTransactionState.ROLLBACK_MESSAGE;
  5.  * 3、网络等原因服务宕机收不到返回结果，对应return LocalTransactionState.UNKNOW;
  6.  */

（3）调用this.endTransaction(sendResult, localTransactionState, localException)方法，将本地事务执行的状态、消息发送的状态、异常信息，一起发送给broker，broker会根据提交的信息，确定是否确定投递消息、删除消息、或在回查本地事务。

更多详细信息，请查看事务消息

  1. public TransactionSendResult sendMessageInTransaction(Message msg, TransactionListener tranExecuter, Object arg) throws MQClientException {
  2.         if (null == tranExecuter) {
  3.             throw new MQClientException("tranExecutor is null", (Throwable)null);
  4.         } else {
  5.             Validators.checkMessage(msg, this.defaultMQProducer);
  6.             SendResult sendResult = null;
  7.             MessageAccessor.putProperty(msg, "TRAN_MSG", "true");
  8.             MessageAccessor.putProperty(msg, "PGROUP", this.defaultMQProducer.getProducerGroup());
  10.             try {
  11.                 sendResult = this.send(msg);
  12.             } catch (Exception var10) {
  13.                 throw new MQClientException("send message Exception", var10);
  14.             }
  16.             LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;
  17.             Throwable localException = null;
  18.             switch(sendResult.getSendStatus()) {
  19.             case SEND_OK:
  20.                 try {
  21.                     if (sendResult.getTransactionId() != null) {
  22.                         msg.putUserProperty("__transactionId__", sendResult.getTransactionId());
  23.                     }
  25.                     String transactionId = msg.getProperty("UNIQ_KEY");
  26.                     if (null != transactionId && !"".equals(transactionId)) {
  27.                         msg.setTransactionId(transactionId);
  28.                     }
  30.                     localTransactionState = tranExecuter.executeLocalTransaction(msg, arg);
  31.                     if (null == localTransactionState) {
  32.                         localTransactionState = LocalTransactionState.UNKNOW;
  33.                     }
  35.                     if (localTransactionState != LocalTransactionState.COMMIT_MESSAGE) {
  36.                         this.log.info("executeLocalTransactionBranch return {}", localTransactionState);
  37.                         this.log.info(msg.toString());
  38.                     }
  39.                 } catch (Throwable var9) {
  40.                     this.log.info("executeLocalTransactionBranch exception", var9);
  41.                     this.log.info(msg.toString());
  42.                     localException = var9;
  43.                 }
  44.                 break;
  45.             case FLUSH_DISK_TIMEOUT:
  46.             case FLUSH_SLAVE_TIMEOUT:
  47.             case SLAVE_NOT_AVAILABLE:
  48.                 localTransactionState = LocalTransactionState.ROLLBACK_MESSAGE;
  49.             }
  51.             try {
  52.                 this.endTransaction(sendResult, localTransactionState, localException);
  53.             } catch (Exception var8) {
  54.                 this.log.warn("local transaction execute " + localTransactionState + ", but end broker transaction failed", var8);
  55.             }
  57.             TransactionSendResult transactionSendResult = new TransactionSendResult();
  58.             transactionSendResult.setSendStatus(sendResult.getSendStatus());
  59.             transactionSendResult.setMessageQueue(sendResult.getMessageQueue());
  60.             transactionSendResult.setMsgId(sendResult.getMsgId());
  61.             transactionSendResult.setQueueOffset(sendResult.getQueueOffset());
  62.             transactionSendResult.setTransactionId(sendResult.getTransactionId());
  63.             transactionSendResult.setLocalTransactionState(localTransactionState);
  64.             return transactionSendResult;
  65.         }
  66.     }
  68.   //将本地事务执行情况，及消息发送状态，异常信息，一起发送给broker,broker根据提交的信息，确定是否确定投递消息、删除消息、或在回查本地事务
  70.     public void endTransaction(SendResult sendResult, LocalTransactionState localTransactionState, Throwable localException) throws RemotingException, MQBrokerException, InterruptedException, UnknownHostException {
  71.         MessageId id;
  72.         if (sendResult.getOffsetMsgId() != null) {
  73.             id = MessageDecoder.decodeMessageId(sendResult.getOffsetMsgId());
  74.         } else {
  75.             id = MessageDecoder.decodeMessageId(sendResult.getMsgId());
  76.         }
  78.         String transactionId = sendResult.getTransactionId();
  79.         String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(sendResult.getMessageQueue().getBrokerName());
  80.         EndTransactionRequestHeader requestHeader = new EndTransactionRequestHeader();
  81.         requestHeader.setTransactionId(transactionId);
  82.         requestHeader.setCommitLogOffset(id.getOffset());
  83.         switch(localTransactionState) {
  84.         case COMMIT_MESSAGE:
  85.             requestHeader.setCommitOrRollback(8);
  86.             break;
  87.         case ROLLBACK_MESSAGE:
  88.             requestHeader.setCommitOrRollback(12);
  89.             break;
  90.         case UNKNOW:
  91.             requestHeader.setCommitOrRollback(0);
  92.         }
  94.         requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
  95.         requestHeader.setTranStateTableOffset(sendResult.getQueueOffset());
  96.         requestHeader.setMsgId(sendResult.getMsgId());
  97.         String remark = localException != null ? "executeLocalTransactionBranch exception: " + localException.toString() : null;
  98.         this.mQClientFactory.getMQClientAPIImpl().endTransactionOneway(brokerAddr, requestHeader, remark, (long)this.defaultMQProducer.getSendMsgTimeout());
  99.     }

发表评论

表情：
评论列表 （有 0 条评论，164人围观）

还没有评论，来说两句吧...

相关阅读