二十三、RocketMQ源码分析之事务消息实现原理下篇-消息服务器Broker提交回滚事务实现原理

文章永久连接：https://tech.souyunku.com/?p=6045

作者：唯有坚持不懈 | 出处：https://blog.csdn.net/prestigeding/article/details/78888290

   本文将重点分析RocketMQ Broker如何处理事务消息提交、回滚命令，其核心实现就是根据commitlogOffset找到消息，如果是提交动作，就恢复原消息的主题与队列，再次存入commitlog文件进而转到消息消费队列，供消费者消费，然后将原预处理消息存入一个新的主题RMQ_SYS_TRANS_OP_HALF_TOPIC，代表该消息已被处理；回滚消息与提交事务消息不同的是，提交事务消息会将消息恢复原主题与队列，再次存储在commitlog文件中。源码入口：
  EndTransactionProcessor#processRequest

OperationResult result = new OperationResult();
if (MessageSysFlag.TRANSACTION_COMMIT_TYPE == requestHeader.getCommitOrRollback()) {        // @1
result = this.brokerController.getTransactionalMessageService().commitMessage(requestHeader);    // @2
      if (result.getResponseCode() == ResponseCode.SUCCESS) {  // @3
          RemotingCommand res = checkPrepareMessage(result.getPrepareMessage(), requestHeader);    // @4
          if (res.getCode() == ResponseCode.SUCCESS) {
                MessageExtBrokerInner msgInner = endMessageTransaction(result.getPrepareMessage());     // @5
                msgInner.setSysFlag(MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(), requestHeader.getCommitOrRollback()));
                msgInner.setQueueOffset(requestHeader.getTranStateTableOffset());
                msgInner.setPreparedTransactionOffset(requestHeader.getCommitLogOffset());
                msgInner.setStoreTimestamp(result.getPrepareMessage().getStoreTimestamp());    // @6
                RemotingCommand sendResult = sendFinalMessage(msgInner);                              // @7
                if (sendResult.getCode() == ResponseCode.SUCCESS) {             
                    this.brokerController.getTransactionalMessageService().deletePrepareMessage(result.getPrepareMessage());    // @8
                }
                return sendResult;
           }
          return res;
     }
}

   代码@1：如果请求为提交事务，进入事务消息提交处理流程。
   代码@2：提交消息，别被这名字误导了，该方法主要是根据commitLogOffset从commitlog文件中查找消息返回OperationResult实例。

    ● private MessageExt prepareMessage ：消息对象。
    ● private int responseCode：查找结果。
    ● private String responseRemark ：错误提示。
   代码@3：如果成功查找到消息，则继续处理，否则返回给客户端，消息未找到错误信息。
   代码@4：验证消息必要字段。
    ● 验证消息的生产组与请求信息中的生产者组是否一致。
    ● 验证消息的队列偏移量（queueOffset）与请求信息中的偏移量是否一致。
    ● 验证消息的commitLogOffset与请求信息中的CommitLogOffset是否一致。
   代码@5:调用endMessageTransaction方法，该方法主要的目的就是恢复事务消息的真实的主题、队列，并设置事务ID。
   代码@6：设置消息的相关属性，这一步应该直接在endMessageTransaction中实现就好，统一恢复原消息的数量，特别关注的是取消了事务相关的系统标记。
   代码@7：发送最终消息，其实现原理非常简单，调用MessageStore将消息存储在commitlog文件中，此时的消息，会被转发到原消息主题对应的消费队列，被消费者消费。
   代码@8：删除预处理消息(prepare)，其实是将消息存储在主题为：RMQ_SYS_TRANS_OP_HALF_TOPIC的主题中，代表这些消息已经被处理（提交或回滚）。
   上述就是事务消息提交的流程，事务回滚类似，接下来大概分析一下事务消息回滚的流程。
   EndTransactionProcessor#processRequest

 else if (MessageSysFlag.TRANSACTION_ROLLBACK_TYPE == requestHeader.getCommitOrRollback()) {
       result = this.brokerController.getTransactionalMessageService().rollbackMessage(requestHeader);    // @1
       if (result.getResponseCode() == ResponseCode.SUCCESS) {
            RemotingCommand res = checkPrepareMessage(result.getPrepareMessage(), requestHeader);
            if (res.getCode() == ResponseCode.SUCCESS) {
                this.brokerController.getTransactionalMessageService().deletePrepareMessage(result.getPrepareMessage());   // @2
            }
           return res;
       }
}

    代码@1：回滚消息，其实内部就是根据commitlogOffset查找消息。
    代码@2：将消息存储在RMQ_SYS_TRANS_OP_HALF_TOPIC中，代表该消息已被处理，与提交事务消息不同的是，提交事务消息会将消息恢复原主题与队列，再次存储在commitlog文件中。
    事务消息在Broker服务端的提交回滚流程就介绍到这了。其核心实现就是根据commitlogOffset找到消息，如果是提交动作，就恢复原消息的主题与队列，再次存入commitlog文件进而转到消息消费队列，供消费者消费，然后将原预处理消息存入一个新的主题RMQ_SYS_TRANS_OP_HALF_TOPIC，代表该消息已被处理；回滚消息与提交事务消息不同的是，提交事务消息会将消息恢复原主题与队列，再次存储在commitlog文件中。

干货推荐

本站推荐：精选优质专栏

专栏汇总：RabbitMQ 源码解析

专栏汇总：Dubbo 源码分析

专栏汇总：Tomcat源码分析

附录：RocketMQ源码分析，系列文章

• 一、RocketMQ源码分析之NameServer
• 二、RocketMQ源码分析之Broker概述与同步消息发送原理与高可用设计及思考
• 三、RocketMQ源码分析之CommitLog消息存储机制
• 四、RocketMQ源码分析之消息消费概述
• 五、RocketMQ源码分析消息消费机制—-消费者拉取消息机制
• 六、RocketMQ源码分析消息消费机制—-消费端消息负载均衡机制与重新分布
• 七、RocketMQ源码分析之消息消费重试机制
• 八、RocketMQ源码分析之消息ACK机制（消费进度）
• 九、RocketMQ源码分析之消费队列、Index索引文件存储结构与存储机制-上篇
• 十、RocketMQ源码分析之消费队列、Index索引文件存储结构与存储机制-下篇
• 十一、RocketMQ源码分析刷盘机制
• 十二、RocketMQ源码分析消息过滤机制上篇—–消息消费服务端过滤与TAG模式过滤实现
• 十三、RocketMQ源码分析消息过滤机制下篇-FilterServer、ClassFilter模式详解
• 十四、RocketMQ源码分析消息拉取拉模式PULL
• 十五、RocketMQ源码分析消息PULL-长轮询模式
• 十六、RocketMQ源码分析顺序消息消费实现原理
• 十七、RocketMQ源码分析文件清除机制
• 十八、源码研究RocketMQ主从同步机制(HA)
• 十九、RocketMQ 主从同步读写分离机制
• 二十、RocketMQ源码分析之从官方示例窥探RocketMQ事务消息实现基本思想
• 二十一、RocketMQ源码分析之RocketMQ事务消息实现原理上篇
• 二十二、RocketMQ源码分析之RocketMQ事务消息实现原理中篇—-事务消息状态回查
• 【当前读到】二十三、RocketMQ源码分析之事务消息实现原理下篇-消息服务器Broker提交回滚事务实现原理
• 二十四、RocketMQ事务消息实战
• 二十五、RocketMQ实战：生产环境中，autoCreateTopicEnable为什么不能设置为true
• 二十六、RocketMQ 消息发送system busy、broker busy原因分析与解决方案
• 二十七、RocketMQ HA机制(主从同步)
• 二十八、RocketMQ ACL使用指南
• 二十九、RocketMQ源码分析 ACL实现机制
• 三十、RocketMQ消息轨迹-设计篇
• 三十一、RocketMQ源码分析消息轨迹
• 三十二、RocketMQ一个新的消费组初次启动时从何处开始消费呢？
• 三十三、RocketMQ 多副本前置篇：初探raft协议
• 三十四、源码分析 RocketMQ DLedger 多副本之 Leader 选主
• 三十五、源码分析 RocketMQ DLedger 多副本存储实现
• 三十六、RocketMQ 主题扩分片后遇到的坑