RocketMQ源码分析之RocketMQ事务消息实现原理上篇(二阶段提交)
在阅读本文前,若您对RocketMQ技术感兴趣,请加入 RocketMQ技术交流群
根据上文的描述,发送事务消息的入口为:
TransactionMQProducer#sendMessageInTransaction: public TransactionSendResult sendMessageInTransaction(final Message msg, final Object arg) throws MQClientException { if (null == this.transactionListener) { // @1 throw new MQClientException("TransactionListener is null", null); } return this.defaultMQProducerImpl.sendMessageInTransaction(msg, transactionListener, arg); // @2 }
代码@1:如果transactionListener为空,则直接抛出异常。
代码@2:调用defaultMQProducerImpl的sendMessageInTransaction方法。
接下来重点分享sendMessageInTransaction方法
DefaultMQProducerImpl#sendMessageInTransaction public TransactionSendResult sendMessageInTransaction(final Message msg, final TransactionListener tranExecuter, final Object arg) throws MQClientException {
Step1:首先先阐述一下参数含义。
- final Message msg:消息
- TransactionListener tranExecuter:事务监听器
- Object arg:其他附加参数
DefaultMQProducerImpl#sendMessageInTransaction SendResult sendResult = null; MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TRANSACTION_PREPARED, "true"); MessageAccessor.putProperty(msg, MessageConst.PROPERTY_PRODUCER_GROUP, this.defaultMQProducer.getProducerGroup()); try { sendResult = this.send(msg); } catch (Exception e) { throw new MQClientException("send message Exception", e); }
Step2:在消息属性中增加两个属性:TRAN_MSG,其值为true,表示为事务消息;PGROUP:消息所属发送者组,然后以同步方式发送消息。在消息发送之前,会先检查消息的属性TRAN_MSG,如果存在并且值为true,则通过设置消息系统标记的方式,设置消息为MessageSysFlag.TRANSACTION_PREPARED_TYPE。
DefaultMQProducerImpl#sendKernelImpl final String tranMsg = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED); if (tranMsg != null && Boolean.parseBoolean(tranMsg)) { sysFlag |= MessageSysFlag.TRANSACTION_PREPARED_TYPE; } SendMessageProcessor#sendMessage String traFlag = oriProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED); if (traFlag != null && Boolean.parseBoolean(traFlag)) { if (this.brokerController.getBrokerConfig().isRejectTransactionMessage()) { response.setCode(ResponseCode.NO_PERMISSION); response.setRemark( "the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1() + "] sending transaction message is forbidden"); return response; } putMessageResult = this.brokerController.getTransactionalMessageService().prepareMessage(msgInner); } else { putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner); }
Step3:Broker端收到客户端发送消息请求后,判断消息类型。如果是事务消息,则调用TransactionalMessageService#prepareMessage方法,否则走原先的逻辑,调用MessageStore#putMessage方法将消息存入Broker服务端。
本节重点阐述事务消息的实现原理,故接下来将重点关注prepareMessage方法,如想了解RocketMQ消息存储相关,可以关注作者源码分析RocketMQ系列。
org.apache.rocketmq.broker.transaction.queue.TransactionalMessageServiceImpl#prepareMessage public PutMessageResult prepareMessage(MessageExtBrokerInner messageInner) { return transactionalMessageBridge.putHalfMessage(messageInner); }
step4:事务消息,将调用TransactionalMessageServiceImpl#prepareMessage方法,继而调用TransactionalMessageBridge#prepareMessage方法。
TransactionalMessageBridge#parseHalfMessageInner public PutMessageResult putHalfMessage(MessageExtBrokerInner messageInner) { return store.putMessage(parseHalfMessageInner(messageInner)); } private MessageExtBrokerInner parseHalfMessageInner(MessageExtBrokerInner msgInner) { MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_TOPIC, msgInner.getTopic()); MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msgInner.getQueueId())); msgInner.setSysFlag( MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(), MessageSysFlag.TRANSACTION_NOT_TYPE)); msgInner.setTopic(TransactionalMessageUtil.buildHalfTopic()); msgInner.setQueueId(0); msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties())); return msgInner; }
Step5:备份消息的原主题名称与原队列ID,然后取消消息的事务消息标签,重新设置消息的主题为:RMQ_SYS_TRANS_HALF_TOPIC,队列ID固定为0。然后调用MessageStore#putMessage方法将消息持久化,这里TransactionalMessageBridge桥接类,就是封装事务消息的相关流程,最终调用MessageStore完成消息的持久化。消息入库后,会继续回到DefaultMQProducerImpl#sendMessageInTransaction,上文的Step2后面,也就是通过同步将消息发送到消息服务端。
注:这是事务消息Prepare状态的处理逻辑,消息是存储在消息服务器了,但存储的并不是原主题,而是RMQ_SYS_TRANS_HALF_TOPIC,故此时消费端是无法消费shen
生产者发送的消息的。看到这里,如果对RocketMQ比较熟悉的话,肯定会有一个“定时任务”去取这个主题下的消息,然后则“合适”的时机将消息的主题恢复。
DefaultMQProducerImpl#sendMessageInTransaction switch (sendResult.getSendStatus()) { case SEND_OK: { try { if (sendResult.getTransactionId() != null) { msg.putUserProperty("__transactionId__", sendResult.getTransactionId()); } String transactionId = msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX); if (null != transactionId && !"".equals(transactionId)) { msg.setTransactionId(transactionId); } localTransactionState = tranExecuter.executeLocalTransaction(msg, arg); if (null == localTransactionState) { localTransactionState = LocalTransactionState.UNKNOW; } if (localTransactionState != LocalTransactionState.COMMIT_MESSAGE) { log.info("executeLocalTransactionBranch return {}", localTransactionState); log.info(msg.toString()); } } catch (Throwable e) { log.info("executeLocalTransactionBranch exception", e); log.info(msg.toString()); localException = e; } } break; case FLUSH_DISK_TIMEOUT: case FLUSH_SLAVE_TIMEOUT: case SLAVE_NOT_AVAILABLE: localTransactionState = LocalTransactionState.ROLLBACK_MESSAGE; break; default: break; }
Step6:如果消息发送成功,会回调TransactionListener#executeLocalTransaction方法,执行本地事务,并且返回本地事务状态为LocalTransactionState,枚举值如下:
- COMMIT_MESSAGE,
- ROLLBACK_MESSAGE,
- UNKNOW
注意:TransactionListener#executeLocalTransaction是在发送者成功发送PREPARED消息后,会执行本地事务方法,然后返回本地事务状态;如果PREPARED消息发送失败,则不会调用TransactionListener#executeLocalTransaction,并且本地事务消息,设置为LocalTransactionState.ROLLBACK_MESSAGE,表示消息需要被回滚。
DefaultMQProducerImpl#sendMessageInTransaction try { this.endTransaction(sendResult, localTransactionState, localException); } catch (Exception e) { log.warn("local transaction execute " + localTransactionState + ", but end broker transaction failed", e); }
step7:调用endTransaction方法结束事务(提交或回滚)。
DefaultMQProducerImpl#endTransaction EndTransactionRequestHeader requestHeader = new EndTransactionRequestHeader(); requestHeader.setTransactionId(transactionId); requestHeader.setCommitLogOffset(id.getOffset()); switch (localTransactionState) { case COMMIT_MESSAGE: requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_COMMIT_TYPE); break; case ROLLBACK_MESSAGE: requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_ROLLBACK_TYPE); break; case UNKNOW: requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_NOT_TYPE); break; default: break; } requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup()); requestHeader.setTranStateTableOffset(sendResult.getQueueOffset()); requestHeader.setMsgId(sendResult.getMsgId());
step8:组装结束事务请求,主要参数为:事务ID、事务操作(commitOrRollback)、消费组、消息队列偏移量、消息ID,fromTransactionCheck,从这里发出的请求,默认为false。Broker端的请求处理器为:EndTransactionProcessor。
step9:EndTransactionProcessor根据事务提交类型:TRANSACTION_COMMIT_TYPE(提交事务)、TRANSACTION_ROLLBACK_TYPE(回滚事务)、TRANSACTION_NOT_TYPE(忽略该请求)。
到目前为止,已详细梳理了RocketMQ事务消息的发送流程,更加准确的说是Prepare状态的消息发送流程。具体流程如图所示:
本文到这里,初步展示了事务消息的发送流程,总的说来,RocketMQ的事务消息发送使用二阶段提交思路,首先,在消息发送时,先发送消息类型为Prepread类型的消息,然后在将该消息成功存入到消息服务器后,会回调TransactionListener#executeLocalTransaction,执行本地事务状态回调函数,然后根据该方法的返回值,结束事务:
1、COMMIT_MESSAGE :提交事务。
2、ROLLBACK_MESSAGE:回滚事务。
3、UNKNOW:未知事务状态,此时消息服务器(Broker)收到EndTransaction命令时,将不对这种消息做处理,消息还处于Prepared类型,存储在主题为:RMQ_SYS_TRANS_HALF_TOPIC的队列中,然后消息发送流程将结束,那这些消息如何提交或回滚呢?
为了实现避免客户端需要再次发送提交、回滚命令,RocketMQ会采取定时任务将RMQ_SYS_TRANS_HALF_TOPIC中的消息取出,然后回到客户端,判断该消息是否需要提交或回滚,来完成事务消息的声明周期,该部分内容将在下节重点探讨。
若您对RocketMQ技术感兴趣,请加入作者所在的 RocketMQ技术交流群
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
报名 | Apache Dubbo™ 开发者沙龙@广州
Dubbo 诞生于 2008 年,是阿里巴巴开源的高性能分布式服务框架(A High Performance Java RPC Framework),使得应用可通过高性能的 RPC 实现服务的输出和输入功能,可以和 Spring 框架无缝集成。Dubbo 于 2018 年成为 Apache 基金会孵化项目。 2019 年,Apache Dubbo™ 开发者沙龙继续前行,将新一年的首站定在了羊城广州。 沙龙嘉宾阵容不仅延续了以往的高水准,我们在当天早上将首次开展小型的见面会,邀请 Dubbo 用户、社区核心成员面对面,一起坐下来聊聊社区发展的那些事儿。 活动主题:Apache Dubbo™ 开发者沙龙 活动时间:2019年1月19日(周六)13:00 - 18:00 活动地点:广州市越秀区广州大道中 289 号印报楼三层 A320、A321 活
- 下一篇
Python 连接数据库
用Python来编写网站,必须要能够通过python操作数据库,所谓操作数据库,就是通过python实现对数据的连接,以及对记录、字段的各种操作。 安装python-MySQLdb 要想通过python来操作数据库,还需要在已经安装了mysql的基础上安装一个称之为mysqldb的库,它是一个接口程序,python通过它对mysql数据实现各种操作。 在编程中,会遇到很多类似的接口程序,通过接口程序对另外一个对象进行操作,比较简单。 这里下载python-mysqldb:https://pypi.python.org/pypi/MySQL-python/ 下载之后就可以安装了。 我这里只能演示ubuntu下安装的过程。 sudo apt-get install python-mysqldb 在shell中输入上面的命令行,就安装了。 不管什么系统,安装不是难题。安装之后,怎么知道安装的结果呢? >>> import MySQLdb 在python的交互模式中,输入上面的指令,如果不报错,恭喜你,已经安装好了。如果报错,恭喜你,可以借着错误信息提高自己的计算机水平了,请...
相关文章
文章评论
共有0条评论来说两句吧...
文章二维码
点击排行
推荐阅读
最新文章
- CentOS7编译安装Gcc9.2.0,解决mysql等软件编译问题
- CentOS关闭SELinux安全模块
- Docker使用Oracle官方镜像安装(12C,18C,19C)
- Docker安装Oracle12C,快速搭建Oracle学习环境
- CentOS8,CentOS7,CentOS6编译安装Redis5.0.7
- CentOS6,7,8上安装Nginx,支持https2.0的开启
- CentOS8编译安装MySQL8.0.19
- CentOS7安装Docker,走上虚拟化容器引擎之路
- SpringBoot2配置默认Tomcat设置,开启更多高级功能
- CentOS8安装Docker,最新的服务器搭配容器使用