08、Dubbo实战:分布式事务之MQ方案

自从小王玩起了微服务,发现微服务果然很强大,好处真是太多,心中暗喜,然而,却也遇到了分布式中最棘手的问题:分布式事务。小王遍访各路神仙,也无个完美开源解决方案,当然,也有些实际可行的手法,虽不算完美,但也可拿来研究一番,那今天我们也来说说分布式事务。 分布式事务的起源,即因各服务是独立的,各自使用独立的DB,那本地事务可以保证事务式执行,但其他服务上关联的事务呢?之前 Dubbo学习系列之六(微服务架构实战)项目中铺垫的最大bug在于:如果订单付款中异常,本地订单数据将会自动回滚,然而库存服务和物流服务,收到通信消息后,就执行了,并没有同步回滚!这就是典型的分布式事务。
分布式事务的解决思路有二:其一,强一致性方案,2PC(两阶段提交)和TCC(try-confirm-cancel)都是对所有相关的事务做同步协调处理,先尝试准备资源,锁住数据,如果都可执行,才都开始执行事务,执行后再确认,如果有一个事务失败,将全部回滚,这个方案缺点太多,执行效率低,回滚代价高,锁住资源多,对各业务侵入严重,必须人工进行各种事件编码,除了如金融类强调一致性的场景,并不推荐使用。其二,弱一致性(最终一致性)方案,基于可靠性MQ的异步,根据BASE理论以及CAP理论,即形成最终一致性。本地事务先执行,然后发送消息至其他服务,并确保消费者服务执行相关事务。举例:只要订单付款成功,那么就一定会减掉用户的账户金额。下面就动手实现这个例子吧!准备:Idea201902/JDK11/ZK3.5.5/Gradle5.4.1/RabbitMQ3.7.13/Mysql8.0.11/Lombok0.26/Erlang21.2/postman7.5.0/Redis3.2/Rocket4.5.2

难度:新手--战士--老兵--大师

目标:1.模拟商城系统,订单付款后,通过RocketMQ消息机制实现分布式事务方法 2.使用RabbitMQ延迟队列实现订单过期取消 步骤:1.项目架构及代码基础设施,见往期文章,整体不变,只做增量,如有重大修改,会加以说明。2.Rocket的安装见文章最后部分后记第2点。
3、 RocketMQ事务实现逻辑:“两段提交,回查确认”,首先msg发给Rocket,但msg并没立刻发送给下游消费者,而是等待发起者的事务执行结果,如果执行无误,则Rocket才将消息发送到下游,如果事务执行失败,则消息返回,下游无感知;
4、 先实现第二个目标--使用RabbitMQ延迟队列实现订单过期取消,先看一个整体逻辑图:;

*

exchangeA-->queueC是“订单到物流线”, exchange A--> q ueueA-->exchangeB-->queueB是“过期订单取消线”。 5.订单到物流线参考往期文章-- Dubbo学习系列之六(微服务架构实战) ,有微调,过期订单取消线,看几个核心点:com.biao.mall.business.impl.DubboOrderServiceImpl中saveOrder(orderBO)方法中的最后添加消息的逻辑,即将订单放入过期订单取消线,注意所有订单都会放入,然后在取消处理逻辑中判断是否取消,因为不能选择性从queue取出特定消息:

*

再定义消息消费者,处理订单消息:
com.biao.mall.business.service.DLXMsgConsumer

*

再定义一个订单取消逻辑:

com.biao.mall.business.impl.DubboOrderServiceImpl中的cancelOrder(String orderId)方法:

在这里判断订单是否需要取消,已经付款的直接返回。需要取消的然后更新订单状态,更新库存:

*

6、 测试流程:启动Redis-->ZK-->Stock-->Business-->Logistics;

模拟一个订单:

*

数据库会生成一个订单:

*

队列中会生成一个消息:

*

一段时间过期后,自动进入DLX处理,DB中可以看到最后订单被取消,置为expired,这个过程中同时也可观察到dubbo_stock表,锁定库存数变化又回退的过程。

*

7、 再来看第一个逻辑实现过程,这个稍微复杂一点:现改变下分析顺序,我们按实际逻辑处理顺序来:;

*

controller前端传入后,进入com.biao.mall.business.impl.DubboOrderServiceImpl这个核心逻辑来处理:

*

这里使用了线程池,充分利用硬件,保证效率。注意这里必须使用事务型消息生产者:TransactionMQProducer,其绑定了一个事务监听器:TransactionListenerImpl,这个是我认为是事务MQ的核心,监听器可以根据处理事务结果决定下一步处理,可以看到msg/orderId/this作为实参传入,为啥这样设计?有点怪异的感觉。看看传入后的地方或许有答案:com.biao.mall.business.impl.DubboOrderServiceImpl

*

这里我特意构造了一个用于传参的构造函数,对应上一图的实参,答案就是第一个方法executeLocalTransaction是用于处理本地事务的,即业务逻辑要放这里!而且特别注意,我使用的是orderService中构造一个独立的payOrderTrans方法,其实是上一版本中payOrder方法的切割,然后在这里引入使用,且这里的arg不是对应形参Object arg!!执行正常即返回对应的状态常量。

*

第二个方法checkLocalTransaction是用于检查本地事务执行的结果,即对第一个方法的跟踪,并提供回调,最终结果返回到DubboOrderServiceImpl中。补充说明下三个状态,事务执行结果就是这三个之一:

*

UNKNOW 时,程序会自动不断check状态,默认15次(可指定)之后报异常,COMMIT表示执行完成,向下游发送消息,ROLLBACK即回滚,下游无感知;我注释了三种状态的check次数。

*

7、 再回到DubboOrderServiceImpl,其中的transactionMQProducer通过com.biao.mall.common.conf.RocketTransMqProducerConf注入,;

注意这里必须使用事务型Producer:

*

8、 同理,消息consumer注入在com.biao.mall.common.conf.RocketMqConsumerConf,注意选择的是Push型消费者,然后加入了一个监听器(里面包含真正的消费业务处理逻辑),最后是设置订阅的主题和标签;

*

9、 再进一步,看这个的消费者监听器实现com.biao.mall.common.component.RocketConsumeMsgListener,这里有个核心方法consumeMessage,处理消息,业务逻辑我暂时做个消息展示,后续再实现,要特别注意这个的幂等处理,因消息可能会重复消费我亲测,如果开另外一个客户端做消费,就会收到重复消息,分布式多实例下,非广播消息是竞争消费,广播型消息全消费,所以幂等必须要处理!;

*

10、 回到DubboOrderServiceImpl中,rocketMQService对象实现,两个方法,一个发消息,一个收消息:;

*

11、 为了简化并完备逻辑,我直接在DubboOrderServiceImpl中做消息消费处理即//扣款注释下;

12、 测试:启动ZK-->Redis-->Rocket-->stock-->business-->logistics;

postman先生成一个未付款订单:

*

模拟付款:

*

控制台的输出:rocketMQ接收,并模拟扣款处理:

*

最后订单数据,paid为1,完成!其他表数据变化就不展示了。

*

13、 代码结构集体照:;

*

14、 项目代码地址:其中的day11;

https://github.com/xiexiaobiao/dubbo-project.git

项目复盘记:

1、 MQ在微服务中相当之重要,所以出场比较多,这里我简要的做了个几款主流MQ的比较,之所以我这里同时使用了Rabbit和Rocket,折腾一把,是为了实际的体会与学习使用之目的,后续文章肯定会加入kafka;

*

2、 Rocket的安装要点:A.下载bin-release.zip版本,B.要配置ROCKETMQ_HOME环境变量C.如果Windows+JDK11环境,要修改runserver.cmd/runbroker.cmd文件,因JDK兼容性问题,如下命令;

*

会提示xxx命令无法识别,修改时直接去掉-XX xxx即可,详细可参考网络资源 D.Rocket的网页管理工具,需自己打包生成jar,项目是https://github.com/apache/rocketmq-externals.git中的console子项目,具体打包步骤参考官网介绍,很详细,但JDK11时要额外添加依赖包,

*

3、 RocketMQ4.5.2+JDK11在window下目前兼容性很差,官方提供的启动脚本不能正常启动,因启动脚本是根据JDK1.8开发,修改启动脚本,能启动RocketMQ后,console管理工具启动也无法连接RocketMQ,重新安装JDK1.8,一切正常!整了一天未果,但我比较钻尖,就是要使用JDK11,最后只能曲线救国,在linux上安装RocketMQ和console,成功!生成console的jar时,以下警告,可以忽略,我尝试添加netty最新的依赖,可以消除该警告;

*

开启producer,遇到这个错误:*

关闭linux防火墙,或者开相应的端口!linux下安装RocketMQ方法在此不表,稍微复杂,但值得操作。

4、 Windows下启动Rocket遇到的错误示例(JDK11):;

*

Windows下启动Rocket正常的状态(JDK1.8):

*

linux下正常启动(JDK11),也要修改启动脚本,参考我的另一篇:

linux下JDK11和RocketMQ安装,启动命令及成功结果如下:

*

*

5、 linux下安装RocketMQ,详细见我的另一篇:linux下JDK11和RocketMQ安装;

6、 编程就是“思考十分钟,编写一分钟”,很多设计思维很重要,比如此项目中的“过期订单取消线”,如果看官说,没付款且过期的消息才进入DLX,这个回答就是错误的!因为queue无法跳跃式pop消息再比如,MQ事务中各业务逻辑的位置是有约定的,就要做转换设计,业务逻辑如何切割整合,各方的调用等;

7、 这个文章是做了几个晚上的成果,实属不易,点个右下角的"在看"吧![*][nbsp32];