使用事务消息
在DailyMart系统中，用户发起支付后，订单系统需要调用库存服务执行库存扣减逻辑。
由于这是跨服务调用，因此会产生分布式事务。在这里，我们使用RocketMQ的事务消息来实现分布式事务。
1、首先，在订单服务的应用服务层处理支付逻辑，并调用RocketMQ发送事务消息：
@Override
public 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
@Slf4j
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
@Slf4j
@RocketMQMessageListener(consumerGroup = "dailymart_inventory_group", topic = "TRADE-ORDER", selectorExpression = "ORDER-PAID")
public class InventoryDeductionConsumer extends EnhanceMessageHandler implements RocketMQListener {
@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事务消息的原理。结合文章开头的代码，是不是已经很清晰了呢？
