自从小王玩起了微服务,发现微服务果然很强大,好处真是太多,心中暗喜,然而,却也遇到了分布式中最棘手的问题:分布式事务。小王遍访各路神仙,也无个完美开源解决方案,当然,也有些实际可行的手法,虽不算完美,但也可拿来研究一番,那今天我们也来说说分布式事务。
分布式事务的起源,即因各服务是独立的,各自使用独立的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是“订单到物流线”,exchangeA-->queueA-->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事务中各业务逻辑的位置是有约定的,就要做转换设计,业务逻辑如何切割整合,各方的调用等。
往期文章导航:
Dubbo学习系列之四(MybaitsPlus+Druid使用)
欢迎关注我的 公众号,各种技术分析使用实战: