实战与原理：如何基于RocketMQ实现分布式事务？

来源：JAVA日知录

今天，实战Q实事务我们将深入讨论在DailyMart中如何使用RocketMQ的原理于事务消息以及事务消息的底层原理。

使用事务消息

在DailyMart系统中，何基用户发起支付后，现分订单系统需要调用库存服务执行库存扣减逻辑。布式

由于这是跨服务调用，因此会产生分布式事务。原理于在这里，何基我们使用RocketMQ的现分事务消息来实现分布式事务。

1、布式首先，实战Q实事务在订单服务的原理于应用服务层处理支付逻辑，并调用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 接口：

该接口有两个方法：

executeLocalTransaction：用于执行本地事务。checkLocalTransaction：在RocketMQ执行消息回查时检查本地事务执行结果，用于确定消息提交还是回滚。@Component@Slf

4j

public 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@Slf

4j

@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是如何解决这个问题的呢？

1. 发送half消息，探测MQ是否正常

在基于RocketMQ的事务消息中，我们不是先执行自身的订单支付逻辑，而是先让订单系统发送一条 half消息到MQ去。这个half消息本质上是一个订单支付成功的消息，只不过此时库存系统是看不见这个half消息的。然后，我们等待接收这个half消息写入成功的响应通知。

发送half消息的本质其实是为了探测MQ是否仍然正常运行。但问题来了，如上所述，消息会发生丢失，那么half消息丢失怎么办呢？

2. half消息发送失败

在发送half消息时，由于网络原因或者MQ直接挂了，就会导致half消息发送失败。这个时候订单系统需要执行一系列的回滚操作。云服务器在我们的场景中，应该执行退款操作，将钱退还给用户，并告知用户交易失败。

3. half消息成功，订单系统执行自己的业务逻辑

如果成功收到half消息的正常响应，此时订单系统应该执行自己的业务逻辑。在我们这个场景中，就是修改订单数据库状态，将其修改为待发货状态。这部分逻辑就对应上述代码中的executeLocalTransaction()方法。

4. 订单本地事务执行失败

如果订单系统执行本地事务失败，则需要发送一个rollback请求给MQ，让其删除这条half消息。

5. 订单本地事务执行成功

如果订单系统的本地事务执行正常，此时需要发送一个commit请求给MQ，要求MQ对之前的half消息进行commit操作，这样库存系统就可以消费这条消息了。

订单创建消息处于half状态时，库存系统是看不见它的。必须等到订单系统执行commit请求，消息被commit后，库存系统才能看到并获取这条消息进行后续处理。

6. half消息发送成功，但是没收到half的响应

以上就是RocketMQ事务消息的正向流程。

然而，还有一个问题：如果订单系统发送half消息成功后却没有收到half消息的响应，该如何处理呢？

在这种情况下，订单系统可能会误以为是发送half消息到MQ失败了。订单系统就会执行回滚流程，退还支付金额，关闭订单。

然而，此时MQ系统中已经存在了一条half消息。这条half消息又该如何处理呢？

在RocketMQ中，有一套补偿流程。RocketMQ会定期扫描处于half状态的消息。如果一直没有对这个消息执行 commit/rollback 操作，超过了一定的时间，RocketMQ就会回调你的订单系统的一个接口，用以确认你本地事务的情况。

当订单系统收到MQ的回查请求时，就需要检索一下数据库，根据订单状态决定执行commit还是rollback。

这部分逻辑就对应上述代码中checkLocalTransaction()方法。

7. rollback 或者 commit 失败怎么办？

通过上述说明，可以看到，RocketMQ是根据rollback或commit操作来决定half消息的状态的。如果业务系统执行了commit操作，则将half消息设置为可见，库存系统可以消费；如果业务系统执行了rollback操作，MQ就会删除half消息。那么问题来了：如果订单系统在执行rollback或commit操作时失败又该如何处理呢？

这时候仍然依赖于前文提到的回查机制。

由于此时MQ中的消息一直处于half状态，超过一定的超时时间后，MQ会发现这个half消息有问题，然后回调你的订单系统的接口。此时订单系统需要根据订单状态来决定执行commit请求还是rollback请求。

以上，就是RocketMQ事务消息的原理。结合文章开头的代码，是不是已经很清晰了呢？

- End-