大橙子网站建设,新征程启航
为企业提供网站建设、域名注册、服务器等服务
这篇文章将为大家详细讲解有关如何进行RocketMQ事务消息实现,文章内容质量较高,因此小编分享给大家做个参考,希望大家阅读完这篇文章后对相关知识有一定的了解。
成都创新互联公司专注为客户提供全方位的互联网综合服务,包含不限于成都网站建设、成都网站设计、波密网络推广、小程序设计、波密网络营销、波密企业策划、波密品牌公关、搜索引擎seo、人物专访、企业宣传片、企业代运营等,从售前售中售后,我们都将竭诚为您服务,您的肯定,是我们最大的嘉奖;成都创新互联公司为所有大学生创业者提供波密建站搭建服务,24小时服务热线:18980820575,官方网址:www.cdcxhl.com
摘要: 事务消息提交或回滚的实现原理就是根据commitlogOffset找到消息,如果是提交动作,就恢复原消息的主题与队列,再次存入commitlog文件进而转到消息消费队列,供消费者消费,然后将原预处理消息存入一个新的主题RMQ_SYS_TRANS_OP_HALF_TOPIC,代表该消息已被处理;回滚消息与提交事务消息不同的是,提交事务消息会将消息恢复原主题与队列,再次存储在commitlog文件中。
若您对RocketMQ技术感兴趣,请加入 RocketMQ技术交流群
小编将重点分析RocketMQ Broker如何处理事务消息提交、回滚命令,根据前面的介绍,其入口EndTransactionProcessor#processRequest:
OperationResult result = new OperationResult();if (MessageSysFlag.TRANSACTION_COMMIT_TYPE == requestHeader.getCommitOrRollback()) { // @1result = 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实例:
cdn.com/14618120dcde45cd59aba1c5cb3ccd5cee5c9e61.png">
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文件中。
关于如何进行RocketMQ事务消息实现就分享到这里了,希望以上内容可以对大家有一定的帮助,可以学到更多知识。如果觉得文章不错,可以把它分享出去让更多的人看到。