本文将重点分析RocketMQ Broker如何处理事务消息提交、回滚命令,根据前面的介绍,其入口EndTransactionProcessor#proce***equest:

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;br/>}
}
代码@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#proce***equest
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;br/>}
}
代码@1:回滚消息,其实内部就是根据commitlogOffset查找消息。
代码@2:将消息存储在RMQ_SYS_TRANS_OP_HALF_TOPIC中,代表该消息已被处理,与提交事务消息不同的是,提交事务消息会将消息恢复原主题与队列,再次存储在commitlog文件中。

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

转载于:https://blog.51cto.com/14031893/2340243

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

  1. RocketMQ源码分析之RocketMQ事务消息实现原理上篇(二阶段提交)

    在阅读本文前,若您对RocketMQ技术感兴趣,请加入 RocketMQ技术交流群 根据上文的描述,发送事务消息的入口为: TransactionMQProducer#sendMessageInTra ...

  2. RocketMQ源码分析之从官方示例窥探:RocketMQ事务消息实现基本思想

    RocketMQ4.3.0版本开始支持事务消息,后续分享将开始将剖析事务消息的实现原理.首先从官方给出的Demo实例入手,以此通往RocketMQ事务消息的世界中. 官方版本未发布之前,从apache ...

  3. RocketMQ源码分析之延迟消息

    文章目录 前言 一.延迟消息 1.特点 2.使用场景 3.demo 二.发送延迟消息 三.broker端存储延迟消息 四.总结 1.延迟消息工作原理 2.延迟消息在消费者消费重试中的应用 前言 本篇文 ...

  4. RocketMQ源码分析之request-reply特性

    1.什么是request-reply?   RocketMQ4.6.0版本中增加了request-reply新特性,该特性允许producer在发送消息后同步或者异步等待consumer消费完消息并返 ...

  5. 《RocketMQ源码分析》NameServer如何处理Broker的连接

    <RocketMQ源码分析>NameServer如何处理Broker的连接 NameServer 介绍 NameServer 功能 动态路由发现和注册 服务剔除 创建NameServerC ...

  6. RocketMQ源码(八)Broker asyncSendMessage处理消息以及自动创建Topic

    此前已经梳理了RocketMQ的broker接收Producer消息的入口源码RocketMQ(七)broker接收消息入口源码_代码---小白的博客-CSDN博客 在文章的最后我们到了SendMes ...

  7. RocketMQ 源码分析 事务消息

    为什么80%的码农都做不了架构师?>>>    1. 概述 必须必须必须 前置阅读内容: <事务消息(阿里云)> 2. 事务消息发送 2.1 Producer 发送事务消 ...

  8. rocketmq源码分析 -生产者

    概念 生产者producer,用于生产消息,在rocketmq中对应着MQProducer接口. 组件 Producer 消息生产者.在rocketmq中,生产者对应MQProducer接口: pub ...

  9. RocketMQ 源码分析 —— 集成 Spring Boot

    点击上方"芋道源码",选择"设为星标" 做积极的人,而不是积极废人! 源码精品专栏 原创 | Java 2020 超神之路,很肝~ 中文详细注释的开源项目 RP ...

最新文章

  1. windows桌面待办事项_想在手机桌面上安装一个便利贴,下载什么便签软件好?
  2. Linux下 Nginx 启动 重启 关闭
  3. Error: could not open `C:\Java\jre7\lib\i386\jvm.cfg
  4. CentOs6.5下安装svn
  5. 射电天文偏振线的绘制
  6. python排版word文档 效率_5秒搞定Word排版,效率颠覆你的想象!
  7. gear操作系统 基于web的linux开放平台,支持Linux操作系统 开源飞机模拟器FlightGear...
  8. flutter:实现扫码枪获取数据
  9. 51单片机电子制作DIY-----交通灯控制器
  10. easy_ui之搭建后台界面(一)
  11. 洛谷 P5708 【深基2.习2】三角形面积(C)
  12. 配置IIS,以及添加网站
  13. 当年我们一起追过的Java,Java SE 个人笔记
  14. 五位本科生4个月造出芯片毕业!新的后续来了……
  15. g40-45支持虚拟化技术_CPU支持虚拟化,虚拟化有什么用怎么开启
  16. 【九度OJ】查找第K小数
  17. Aztec协议上线:V神「匿名以太坊梦」成真
  18. SpringBoot Tomcat 配置https 且443端口也是https(若依为例对接微信小程序的https,小程序也可以访问)
  19. Android 源码 Camera2 预览流程分析一
  20. 去掉json的双引号_JSON 去掉转义字符,value去掉双引号

热门文章

  1. C++ Set常用用法
  2. UA MATH565C 随机微分方程V Stationary Measure
  3. Winrar从命令行查看压缩文件内容
  4. Lisp和前缀表示法入门
  5. VS2010重构学习总结
  6. 图解VC++版PE文件解析器源码分析
  7. 安卓指令和命令学习总结
  8. HTML5实现屏幕手势解锁(转载)
  9. Wannafly挑战赛22
  10. 从零开始单排学设计模式「简单工厂设计模式」黑铁 III