文档章节

RocketMQ源码分析之RocketMQ事务消息实现原下篇(事务提交或回滚)

阿里云云栖社区
 阿里云云栖社区
发布于 01/08 16:26
字数 1041
阅读 218
收藏 11

摘要: 事务消息提交或回滚的实现原理就是根据commitlogOffset找到消息,如果是提交动作,就恢复原消息的主题与队列,再次存入commitlog文件进而转到消息消费队列,供消费者消费,然后将原预处理消息存入一个新的主题RMQ_SYS_TRANS_OP_HALF_TOPIC,代表该消息已被处理;回滚消息与提交事务消息不同的是,提交事务消息会将消息恢复原主题与队列,再次存储在commitlog文件中。

本文将重点分析RocketMQ Broker如何处理事务消息提交、回滚命令,根据前面的介绍,其入口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实例:
clipboard

  • 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文件中。

作者: 丁威
原文链接
本文为云栖社区原创内容,未经允许不得转载。

© 著作权归作者所有

共有 人打赏支持
阿里云云栖社区
粉丝 85
博文 896
码字总数 1954512
作品 0
朝阳
私信 提问
加载中

评论(1)

9
92年的java
http://www.jujingyun.com 北京网站建设
http://www.shjuntang.com 上海装潢
RocketMQ源码分析之RocketMQ事务消息实现原下篇(事务提交或回滚)

摘要: 事务消息提交或回滚的实现原理就是根据commitlogOffset找到消息,如果是提交动作,就恢复原消息的主题与队列,再次存入commitlog文件进而转到消息消费队列,供消费者消费,然后将原预...

zhaowei121
01/08
0
0
RocketMQ源码分析之从官方示例窥探:RocketMQ事务消息实现基本思想

摘要: RocketMQ源码分析之从官方示例窥探RocketMQ事务消息实现基本思想。 RocketMQ4.3.0版本开始支持事务消息,后续分享将开始将剖析事务消息的实现原理。首先从官方给出的Demo实例入手,以...

阿里云云栖社区
01/08
0
0
RocketMQ源码分析之RocketMQ事务消息实现原理上篇(二阶段提交)

摘要: 初步展示了事务消息的发送流程,总的说来,RocketMQ的事务消息发送使用二阶段提交思路,首先,在消息发送时,先发送消息类型为Prepread类型的消息,然后在将该消息成功存入到消息服务...

zhaowei121
01/08
0
0
RocketMQ源码分析之RocketMQ事务消息实现原理中篇----事务消息状态回查

摘要: 本文详细分析了RocketMQ事务消息实现原理中的事务状态回查实现,RocketMQ会默认一分钟的频率处理消息状态为Prepare的消息,通过调用消息生产者的事务状态查询接口得知消息的事务状态,...

zhaowei121
01/08
0
0
rocketmq事务消息入门介绍

说明 周五的时候发了篇:Rocketmq4.3支持事务啦!!!,趁着周末的时候把相关内容看了下,下面的主要内容就是关于RocketMQ事务相关内容介绍了。 说明: 今天这篇仅仅是入门介绍,并没有涉及到...

匠心零度
2018/10/29
0
0

没有更多内容

加载失败,请刷新页面

加载更多

大数据教程(11.9)hive操作基础知识

上一篇博客分享了hive的简介和初体验,本节博主将继续分享一些hive的操作的基础知识。 DDL操作 (1)创建表 #建表语法CREATE [EXTERNAL] TABLE [IF NOT EXISTS] table_name [(col_name ...

em_aaron
今天
0
0
OSChina 周四乱弹 —— 我家猫真会后空翻

Osc乱弹歌单(2019)请戳(这里) 【今日歌曲】 @我没有抓狂 :#今天听这个# 我艇牛逼,百听不厌,太好听辣 分享 Led Zeppelin 的歌曲《Stairway To Heaven》 《Stairway To Heaven》- Led Z...

小小编辑
今天
1
0
node调用dll

先安装python2.7 安装node-gyp cnpm install node-gyp -g 新建一个Electron-vue项目(案例用Electron-vue) vue init simulatedgreg/electron-vue my-project 安装electron-rebuild cnpm ins......

Chason-洪
今天
3
0
scala学习(一)

学习Spark之前需要学习Scala。 参考学习的书籍:快学Scala

柠檬果过
今天
3
0
通俗易懂解释网络工程中的技术,如STP,HSRP等

导读 在面试时,比如被问到HSRP的主备切换时间时多久,STP几个状态的停留时间,自己知道有这些东西,但在工作中不会经常用到,就老是记不住,觉得可能还是自己基础不够牢固,知识掌握不够全面...

问题终结者
昨天
4
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部