在DailyMart系统中，用户发起支付后，订单系统需要调用库存服务执行库存扣减逻辑。
由于这是跨服务调用，因此会产生分布式事务。在这里，我们使用RocketMQ的事务消息来实现分布式事务。
1、首先，在订单服务的应用服务层处理支付逻辑，并调用RocketMQ发送事务消息：
@Overridepublic String payment(String orderSn) { // todo 集成支付宝支付 // 支付流水号 String outOrderNo = IdUtils.get32UUID(); TradeOrder tradeOrder = Optional.ofNullable(tradeOrderService.getByOrderSn(orderSn)).orElseThrow(() -> new BusinessException("订单编号不存在")); // 如果订单处于待支付状态 if (Objects.equals(tradeOrder.getStatus(), OrderStatusEnum.WAITING_PAYMENT.getStatus())) { OrderPaidEvent orderPaidEvent = new OrderPaidEvent(orderSn, outOrderNo); TransactionSendResult sendResult = enhanceTemplate.sendTransaction("TRADE-ORDER", "ORDER-PAID"); if (SendStatus.SEND_OK == sendResult.getSendStatus() && sendResult.getLocalTransactionState() == LocalTransactionState.COMMIT_MESSAGE) { return tradeOrder.getOrderSn(); } else { throw new BusinessException("支付失败..."); } } else { throw new BusinessException("订单已支付，请勿重复提交..."); }}2、在订单服务的基础设施层，创建一个类实现 RocketMQLocalTransactionListener 接口：
该接口有两个方法：
@Component@Slf4jpublic class OrderPaidTransactionConsumer implements RocketMQLocalTransactionListener { @Resource private TransactionTemplate transactionTemplate; @Resource private TradeOrderService tradeOrderService; /** * 执行本地事务 * 将订单状态修改成已支付 */ @Override public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) { final OrderPaidEvent orderPaidEvent = JsonUtils.byte2Obj((byte[]) message.getPayload(), OrderPaidEvent.class); try { // 放到同一个本地事务中 this.transactionTemplate.executeWithoutResult(status -> { String orderSn = orderPaidEvent.getOrderSn(); // 修改成待发货 tradeOrderService.changeOrderStatus(orderSn, OrderStatusEnum.AWAITING_SHIPMENT); }); return RocketMQLocalTransactionState.COMMIT; } catch (Exception e) { log.error("修改订单状态失败", e); // ROLLBACK 则回滚消息，rocketmq将废弃这条消息 return RocketMQLocalTransactionState.ROLLBACK; // 如果是UNKNOWN, 则触发回查 } } /** * 检查本地事务执行状态 * 消息回查时，对于正在进行中的事务不要返回Rollback或Commit结果，应继续保持Unknown的状态。 */ @Override public RocketMQLocalTransactionState checkLocalTransaction(Message message) { final OrderPaidEvent orderPaidEvent = JsonUtils.byte2Obj((byte[]) message.getPayload(), OrderPaidEvent.class); String orderSn = orderPaidEvent.getOrderSn(); TradeOrder tradeOrder = tradeOrderService.getByOrderSn(orderSn); // 如果已经修改成待发货说明本地事务执行成功，此时消费端可以直接消费 if (Objects.equals(tradeOrder.getStatus(), OrderStatusEnum.AWAITING_SHIPMENT.getStatus())) { return RocketMQLocalTransactionState.COMMIT; } else { // 这里查不到的时候返回 UNKNOWN在于，有可能事务还没有提交，回查就开始了 return RocketMQLocalTransactionState.UNKNOWN; } }}3、在库存服务的基础设施层，监听消息以执行库存扣减逻辑：
@Component@Slf4j@RocketMQMessageListener(consumerGroup = "dailymart_inventory_group", topic = "TRADE-ORDER", selectorExpression = "ORDER-PAID")public class InventoryDeductionConsumer extends EnhanceMessageHandler<OrderPaidEvent> implements RocketMQListener<OrderPaidEvent> { @Resource private InventoryDomainService inventoryDomainService; @Override public void onMessage(OrderPaidEvent orderPaidEvent) { super.dispatchMessage(orderPaidEvent); } @Override protected void handleMessage(OrderPaidEvent orderPaidEvent) throws Exception { // 执行库存扣减逻辑 String orderSn = orderPaidEvent.getOrderSn(); inventoryDomainService.deductionInventory(orderSn); }}通过以上步骤，我们完成了RocketMQ事务消息的发送，利用事务消息的特性保证分布式事务的最终一致性。与普通消息相比，事务消息在处理时需要实现 RocketMQLocalTransactionListener 接口，这是事务消息的核心。
介绍完事务消息的使用，接下来我们再来聊聊事务消息的原理。
首先，让我们思考一下，如果不使用事务消息会有什么问题。
很容易想到的一个问题就是消息丢失。当保存订单后由于网络问题导致消息丢失，如下图所示：
图片
在不使用RocketMQ的情况下，我们往往会通过 本地消息表 + 补偿重试 的机制来保证消息一定会发送出去。其原理可以参考上篇文章 [Dailymart26：微服务中躲不过的坑 - 分布式事务]。
那RocketMQ是如何解决这个问题的呢？
在基于RocketMQ的事务消息中，我们不是先执行自身的订单支付逻辑，而是先让订单系统发送一条 half消息 到MQ去。这个half消息本质上是一个订单支付成功的消息，只不过此时库存系统是看不见这个half消息的。然后，我们等待接收这个half消息写入成功的响应通知。
图片
发送half消息的本质其实是为了探测MQ是否仍然正常运行。但问题来了，如上所述，消息会发生丢失，那么half消息丢失怎么办呢？
在发送half消息时，由于网络原因或者MQ直接挂了，就会导致half消息发送失败。这个时候订单系统需要执行一系列的回滚操作。在我们的场景中，应该执行退款操作，将钱退还给用户，并告知用户交易失败。
如果成功收到half消息的正常响应，此时订单系统应该执行自己的业务逻辑。在我们这个场景中，就是修改订单数据库状态，将其修改为待发货状态。这部分逻辑就对应上述代码中的executeLocalTransaction()方法。
图片
如果订单系统执行本地事务失败，则需要发送一个rollback请求给MQ，让其删除这条half消息。
图片
如果订单系统的本地事务执行正常，此时需要发送一个commit请求给MQ，要求MQ对之前的half消息进行commit操作，这样库存系统就可以消费这条消息了。
图片
订单创建消息处于half状态时，库存系统是看不见它的。必须等到订单系统执行commit请求，消息被commit后，库存系统才能看到并获取这条消息进行后续处理。
以上就是RocketMQ事务消息的正向流程。
然而，还有一个问题：如果订单系统发送half消息成功后却没有收到half消息的响应，该如何处理呢？
在这种情况下，订单系统可能会误以为是发送half消息到MQ失败了。订单系统就会执行回滚流程，退还支付金额，关闭订单。
图片
然而，此时MQ系统中已经存在了一条half消息。这条half消息又该如何处理呢？
在RocketMQ中，有一套补偿流程。RocketMQ会定期扫描处于half状态的消息。如果一直没有对这个消息执行 commit/rollback 操作，超过了一定的时间，RocketMQ就会回调你的订单系统的一个接口，用以确认你本地事务的情况。
当订单系统收到MQ的回查请求时，就需要检索一下数据库，根据订单状态决定执行commit还是rollback。
这部分逻辑就对应上述代码中checkLocalTransaction()方法。
图片
通过上述说明，可以看到，RocketMQ是根据rollback或commit操作来决定half消息的状态的。如果业务系统执行了commit操作，则将half消息设置为可见，库存系统可以消费；如果业务系统执行了rollback操作，MQ就会删除half消息。那么问题来了：如果订单系统在执行rollback或commit操作时失败又该如何处理呢？
这时候仍然依赖于前文提到的回查机制。
由于此时MQ中的消息一直处于half状态，超过一定的超时时间后，MQ会发现这个half消息有问题，然后回调你的订单系统的接口。此时订单系统需要根据订单状态来决定执行commit请求还是rollback请求。
以上，就是RocketMQ事务消息的原理。结合文章开头的代码，是不是已经很清晰了呢？
本文链接：http://www.28at.com/showinfo-26-68319-0.html实战与原理：如何基于RocketMQ实现分布式事务？
声明：本网页内容旨在传播知识，若有侵权等问题请及时与本网联系，我们将在第一时间删除处理。邮件：2376512515@qq.com
上一篇： Python Pathlib模块：一站式解决文件路径难题
下一篇： 我爱说实话，Mica-Http 超好用！
Copyright © 2016-2023 天津谷骐科技有限公司 版权所有 sitemap.xml
违法及侵权请联系：2376512515@qq.com 津ICP备18001702号
津公网安备 12010102000574号