分布式事务-下单，支付，扣减库存

RocketMQ 事务消息实现请详细介绍一下，另外，如果支付失败怎么办

RocketMQ 事务消息 是一种基于消息队列的分布式事务解决方案，通过消息的可靠传递来保证分布式事务的最终一致性。下面详细介绍 RocketMQ 事务消息的实现原理、流程以及如何处理支付失败的情况。

1. RocketMQ 事务消息的原理

RocketMQ 事务消息的核心思想是将分布式事务拆分为两个阶段：

第一阶段：发送预备消息
- 生产者发送一条预备消息到 RocketMQ，此时消息对消费者不可见。
第二阶段：执行本地事务并提交或回滚消息
- 生产者执行本地事务（如创建订单），并根据事务执行结果提交或回滚消息。
- 如果本地事务成功，RocketMQ 将消息标记为可消费状态，消费者可以消费该消息。
- 如果本地事务失败，RocketMQ 会删除该消息。

此外，RocketMQ 提供了 事务状态回查机制，用于解决生产者本地事务执行结果未及时通知 RocketMQ 的问题。

2. RocketMQ 事务消息的流程

以下是 RocketMQ 事务消息的详细流程：

2.1 生产者发送预备消息

生产者发送一条预备消息到 RocketMQ。
RocketMQ 收到预备消息后，将其存储到消息存储系统中，但该消息对消费者不可见。

2.2 生产者执行本地事务

生产者执行本地事务（如创建订单、冻结库存等）。
根据本地事务的执行结果，生产者向 RocketMQ 发送提交或回滚请求。

2.3 RocketMQ 处理提交或回滚请求

如果生产者发送提交请求，RocketMQ 将消息标记为可消费状态，消费者可以消费该消息。
如果生产者发送回滚请求，RocketMQ 会删除该消息。

2.4 事务状态回查

如果生产者未及时发送提交或回滚请求，RocketMQ 会定期回查生产者的事务状态。
生产者根据本地事务的执行结果，返回提交或回滚状态。

2.5 消费者消费消息

消费者从 RocketMQ 中消费已提交的消息，并执行相应的业务逻辑（如扣减库存、支付等）。

3. RocketMQ 事务消息的实现

以下是使用 RocketMQ 事务消息实现下单、支付和库存的示例：

3.1 生产者代码

public class TransactionProducer {public static void main(String[] args) throws Exception {// 创建事务生产者TransactionMQProducer producer = new TransactionMQProducer("transaction_producer_group");producer.setNamesrvAddr("localhost:9876"); // 设置NameServer地址producer.setTransactionListener(new TransactionListenerImpl()); // 设置事务监听器producer.start();// 创建消息Message msg = new Message("order_topic", "下单".getBytes(StandardCharsets.UTF_8));// 发送事务消息TransactionSendResult result = producer.sendMessageInTransaction(msg, null);System.out.println("发送结果: " + result);// 关闭生产者producer.shutdown();}
}// 事务监听器
class TransactionListenerImpl implements TransactionListener {@Overridepublic LocalTransactionState executeLocalTransaction(Message msg, Object arg) {try {// 执行本地事务（如创建订单）boolean success = createOrder();if (success) {return LocalTransactionState.COMMIT_MESSAGE; // 本地事务成功，提交消息} else {return LocalTransactionState.ROLLBACK_MESSAGE; // 本地事务失败，回滚消息}} catch (Exception e) {return LocalTransactionState.ROLLBACK_MESSAGE; // 本地事务异常，回滚消息}}@Overridepublic LocalTransactionState checkLocalTransaction(MessageExt msg) {// 事务状态回查逻辑boolean success = checkOrderStatus();if (success) {return LocalTransactionState.COMMIT_MESSAGE; // 本地事务成功，提交消息} else {return LocalTransactionState.ROLLBACK_MESSAGE; // 本地事务失败，回滚消息}}private boolean createOrder() {// 创建订单逻辑System.out.println("创建订单成功");return true;}private boolean checkOrderStatus() {// 检查订单状态逻辑System.out.println("检查订单状态成功");return true;}
}

3.2 消费者代码

public class TransactionConsumer {public static void main(String[] args) throws Exception {// 创建消费者DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("transaction_consumer_group");consumer.setNamesrvAddr("localhost:9876"); // 设置NameServer地址consumer.subscribe("order_topic", "*"); // 订阅主题// 注册消息监听器consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {for (MessageExt msg : msgs) {System.out.println("消费消息: " + new String(msg.getBody()));// 执行支付逻辑boolean success = paymentService.pay();if (!success) {return ConsumeConcurrentlyStatus.RECONSUME_LATER; // 支付失败，稍后重试}}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; // 消费成功}});// 启动消费者consumer.start();System.out.println("消费者启动成功");}
}