美文网首页Java高级进阶
rocketmq事务消息入门介绍

rocketmq事务消息入门介绍

作者: Java大生 | 来源:发表于2018-09-26 15:51 被阅读4次

    引出分布式事务相关内容

    这里主要是想说明下,是什么背景下面产生了此类问题。

    首先我们来说说事务,说道事务,首先让我想到的就是我大学的时候,老师经常举例转账的事情,例子是这样的:

    银行转账!张三转100块到李四的账户,这其实需要两条SQL语句:

    给张三的账户减去100元

    给李四的账户加上100元

    如果在第一条SQL语句执行成功后,在执行第二条SQL语句之前,程序被中断了(可能是抛出了某个异常,也可能是其他什么原因),那么李四的账户没有加上100元,而张三却减去了100元。这肯定是不行的!你现在可能已经知道什么是事务了吧!事务中的多个操作,要么完全成功,要么完全失败!不可能存在成功一半的情况!也就是说给张三的账户减去100元如果成功了,那么给李四的账户加上100元的操作也必须是成功的;否则给张三减去100元,以及给李四加上100元都是失败的!

    如果在单个服务里面,我们使用spring的话,我们使用一个注解就解决了这个问题(@Transactional注解就解决了,基本上就没有下文什么事情了。)

    随着各各公司服务越来越复杂,数据量越来越多,慢慢的公司就引入了微服务以及分库分表等技术,这些技术的确非常的有用,但是在解决事务方面带来了一下问题。

    强调下:分布式事务,我们一般都是强调的最终一致性,而不是强一致性!!!

    主要概括为3类:

    基于单个JVM,数据库分库分表了(跨多个数据库)。

    基于多JVM,服务拆分了(不跨数据库)。

    基于多JVM,服务拆分了 并且数据库分库分表了。

    正是为了解决上面的3类问题,才引入了分布式事务相关的技术来解决这些问题。

    Half(Prepare) Message

    指的是暂不能投递的消息,发送方已经将消息成功发送到了 MQ 服务端,但是服务端未收到生产者对该消息的二次确认,此时该消息被标记成“暂不能投递”状态,处于该种状态下的消息即半消息。

    消息回查

    由于网络闪断、生产者应用重启等原因,导致某条事务消息的二次确认丢失,MQ 服务端通过扫描发现某条消息长期处于“半消息”时,需要主动向消息生产者询问该消息的最终状态(Commit 或是 Rollback),该过程即消息回查。

    执行流程图

    发送方向 MQ 服务端发送消息。

    MQ Server 将消息持久化成功之后,向发送方 ACK 确认消息已经发送成功,此时消息为半消息。

    发送方开始执行本地事务逻辑。

    发送方根据本地事务执行结果向 MQ Server 提交二次确认(Commit 或是 Rollback),MQ Server 收到 Commit 状态则将半消息标记为可投递,订阅方最终将收到该消息;MQ Server 收到 Rollback 状态则删除半消息,订阅方将不会接受该消息。

    在断网或者是应用重启的特殊情况下,上述步骤4提交的二次确认最终未到达 MQ Server,经过固定时间后 MQ Server 将对该消息发起消息回查。

    发送方收到消息回查后,需要检查对应消息的本地事务执行的最终结果。

    发送方根据检查得到的本地事务的最终状态再次提交二次确认,MQ Server 仍按照步骤4对半消息进行操作。

    事务消息发送对应步骤1、2、3、4,事务消息回查对应步骤5、6、7。

    RocketMQ事务消息如何使用

    1、引入 rocketmq-client

    目前还没有,等几天就会有了,或者自己下载源码进行打包到自己的私有仓库也是一样的,经过和4.2.0的引入差不多,就是换成4.3.0即可。

    <dependency>

        <groupId>org.apache.rocketmq</groupId>

        <artifactId>rocketmq-client</artifactId>

        <version>4.3.0</version>

    </dependency>

    2、编写Producer

    备注:和普通消息发送DefaultMQProducer有所不同,这里使用的是TransactionMQProducer

    关键点:自己实现TransactionListener接口,并实现executeLocalTransaction方法(执行本地事务的,一般就是操作DB相关内容)和checkLocalTransaction方法(用来提供给broker进行会查本地事务消息的,把本地事务执行的结果存储到redis或者DB中都可以,为会查做数据准备)。

    还是以:给张三的账户减去100元, 给李四的账户加上100元为例。这个时候我们发送就相当于做的是张三的账户减去100元的操作,并且操作DB成功了,下面就是另外对李四的账户加上100元的操作了,这个时候就是相当于定义刚刚发送事务消息的topic,消费端和我们普通的消费端没有什么区别。

    思考:假如这个时候我们消费端失败了怎么办呢?(这个问题后续有讨论,消费失败有2种,第一种是超时了,我们重试即可,第二种是真的处理失败了?该怎么办呢?)

    RocketMQ事务消息是怎么实现的

    第一感觉和定时消息做法非常类似,但是比定时消息要复杂。定时消息内容在:RocketMQ(九):消息发送(续)里面提到过。

    本质:定时消息先把定时消息的topic修改为SCHEDULE_TOPIC_XXXX,之后一系列处理……,我们这个事务消息也是一样,先发送Half(Prepare) Message消息的时候,其实topic内容也继续了修改(RMQ_SYS_TRANS_HALF_TOPIC),所有consumer是不可见的(如果提交就用真实topic,如果需要回滚那么就那临时的topic内容删除即可。)

    Half(Prepare) Message修改topic断点如下:

    本篇仅仅是一个入门介绍,里面具体细节后续章节会进行分析。

    思考:如果把发送普通消息和本地执行逻辑放在一个事务里面,如果执行事务成功就发送普通消息,如果失败就回滚好像也是可以,那么这种事务消息相对其有什么优势或者好处呢??? 思考下。

    为什么需要事务消息会查机制

    其实这个内容在,RocketMQ事务消息介绍里面也说明了,由于网络闪断、生产者应用重启等原因,导致某条事务消息的二次确认丢失,MQ 服务端通过扫描发现某条消息长期处于“半消息”时,需要主动向消息生产者询问该消息的最终状态(Commit 或是 Rollback),该过程即消息回查。那么RocketMQ到底怎么做的呢?下面马上就会介绍。

    RocketMQ是怎么进行事务消息会查的

    每60s会对Half(Prepare) Message的topic主题为RMQ_SYS_TRANS_HALF_TOPIC的消息进行check。

    broker会调用自己实现TransactionListener接口的checkLocalTransaction方法

    备注:就是RocketMQ已经实现这个机制,今天这篇仅仅是入门介绍,并没有涉及到很多细节,先把大概流程说明白,后续再具体细节进行开篇说明。

    RocketMQ对于分布式事务解决还有那些局限性?以及说明

    在RocketMQ事务消息如何使用的时候我们提到,如果消费失败怎么办?消费失败有2种,第一种是超时了,我们重试即可,第二种是真的处理失败了?仔细思考下,这块还是蛮复杂的,假如需要有7-8个业务模块呢,其中执行到第6个业务模块就失败呢?这种重试好几次还是失败,我们该如何处理呢???是回滚前面5个操作吗?好复杂好复杂,为什么RocketMQ不提供自动回滚呢?希望大家思考下,如果失败率比较大,那么就是系统问题需要优化代码业务……

    据零度了解,很多这块是通过人工介入以及T+1对账的以及补偿机制等。

    欢迎工作一到五年的Java工程师朋友们加入Java架构开发:点击链接加入群聊【Java架构交流群】:https://jq.qq.com/?_wv=1027&k=5DmXtYD

    本群提供免费的学习指导 架构资料 以及免费的解答

    不懂得问题都可以在本群提出来 之后还会有职业生涯规划以及面试指导

    同时大家可以多多关注一下小编 大家一起学习进步

    相关文章

      网友评论

        本文标题:rocketmq事务消息入门介绍

        本文链接:https://www.haomeiwen.com/subject/jlxjoftx.html