RocketMQ源码分析之RocketMQ事务消息实现原下篇(事务提交或回滚)
摘要: 事务消息提交或回滚的实现原理就是根据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实例:
- 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文件中。

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
随行付微服务之基于Zuul自研服务网关
随行付微服务之服务网关 微服务是时下最流行的架构之一,作为微服务不可或缺的一部分,API网关的作用至关重要。本文将对随行付微服务的API网关实践进行介绍。 API网关的作用 我们知道,在一个微服务系统中,整个系统被划分为许多小模块,客户端想要调用服务,可能需要维护很多ip+port信息,管理十分复杂。API网关作为整个系统的统一入口,所有请求由网关接收并路由转发给内部的微服务。对于客户端而言,系统相当于一个黑箱,客户端不需要关心其内部结构。 随着业务的发展,服务端可能需要对微服务进行重新划分等操作,由于网关将客户端和具体服务隔离,因此可以在尽量不改动客户端的情况下进行。网关可以完成权限验证、限流、安全、监控、缓存、服务路由、协议转换、服务编排、灰度发布等功能剥离出来,讲这些非业务功能统一解决、统一机制处理。 Zuul原理简介 随行付微服务API网关基于Netflix的Zuul实现。Netflix是实践微服务最成功的公司之一,他们创建并开源了一系列微服务相关的框架,Zuul便是用来实现网关功能的框架。Zuul的整体架构图如下: Zuul基于Servlet开发,ZuulServlet是整个...
- 下一篇
RocketMQ源码分析之RocketMQ事务消息实现原理中篇----事务消息状态回查
上节已经梳理了RocketMQ发送事务消息的流程(基于二阶段提交),本节将继续深入学习事务状态消息回查,我们知道,第一次提交到消息服务器时消息的主题被替换为RMQ_SYS_TRANS_HALF_TOPIC,本地事务执行完后如果返回本地事务状态为UN_KNOW时,第二次提交到服务器时将不会做任何操作,也就是说此时消息还存在与RMQ_SYS_TRANS_HALF_TOPIC主题中,并不能被消息消费者消费,那这些消息最终如何被提交或回滚呢? 原来RocketMQ使用TransactionalMessageCheckService线程定时去检测 RMQ_SYS_TRANS_HALF_TOPIC主题中的消息,回查消息的事务状态。TransactionalMessageCheckService的检测频率默认1分钟,可通过在broker.conf文件中设置transactionCheckInterval的值来改变默认值,单位为毫秒。 接下来将深入分析该线程的实现原理,从而解开事务消息回查机制。 TransactionalMessageCheckService#onWaitEnd protected...
相关文章
文章评论
共有0条评论来说两句吧...
文章二维码
点击排行
推荐阅读
最新文章
- CentOS关闭SELinux安全模块
- Hadoop3单机部署,实现最简伪集群
- Windows10,CentOS7,CentOS8安装MongoDB4.0.16
- Springboot2将连接池hikari替换为druid,体验最强大的数据库连接池
- Docker使用Oracle官方镜像安装(12C,18C,19C)
- CentOS8安装MyCat,轻松搞定数据库的读写分离、垂直分库、水平分库
- CentOS8安装Docker,最新的服务器搭配容器使用
- CentOS6,7,8上安装Nginx,支持https2.0的开启
- SpringBoot2整合Redis,开启缓存,提高访问速度
- CentOS7编译安装Cmake3.16.3,解决mysql等软件编译问题