SpringBoot 事务与 RocketMQ 消息消费顺序一致性问题解决方法

在 Spring Boot 中使用事务的订单方法调用 Spring Cloud Stream RocketMQ 修改库存时，可能会遇到事务未提交的问题，并且 RocketMQ 先消费了消息的情况。为了解决这个问题，可以使用 TransactionEventListener 来监听事务的提交状态并执行相应的操作。

首先，需要在 Spring Boot 的配置文件中配置 RocketMQ 相关的属性，如下所示：

spring:
  cloud:
    stream:
      rocketmq:
        binder:
          namesrv-addr: localhost:9876

然后，在订单方法上添加事务注解 @Transactional，确保订单方法中的操作在一个事务中执行。在订单方法中，调用修改库存的方法。

@Transactional
public void createOrder(Order order) {
    // 创建订单逻辑

    // 调用修改库存的方法
    updateInventory(order);

    // 其他操作
}

在修改库存的方法中，通过使用 RocketMQ 的 Template 发送消息，将修改库存的操作放入一个 RocketMQ 事务中。需要注意的是，这里需要使用 RocketMQ 的事务消息发送器 RocketMQLocalTransactionListener。

@Autowired
private RocketMQTemplate rocketMQTemplate;

public void updateInventory(Order order) {
    // 修改库存逻辑

    // 发送事务消息
    rocketMQTemplate.sendMessageInTransaction('inventory-topic', MessageBuilder.withPayload(order).build(), null);
}

接下来，需要实现 RocketMQLocalTransactionListener 接口，处理 RocketMQ 的事务消息。

@Component
public class InventoryTransactionListener implements RocketMQLocalTransactionListener {

    @Autowired
    private InventoryService inventoryService;

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        // 执行本地事务，修改库存

        // 根据事务执行结果返回相应的状态
        if (inventoryService.updateInventory((Order) message.getPayload())) {
            return RocketMQLocalTransactionState.COMMIT;
        } else {
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(MessageExt messageExt) {
        // 检查本地事务状态

        // 根据本地事务状态返回相应的状态
        if (inventoryService.checkInventory((Order) messageExt.getPayload())) {
            return RocketMQLocalTransactionState.COMMIT;
        } else {
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }
}

在上述代码中，executeLocalTransaction 方法用于执行本地事务，即修改库存的操作。通过调用 InventoryService 中的方法来实现修改库存的逻辑。根据事务执行结果，返回相应的事务状态。checkLocalTransaction 方法用于检查本地事务状态，即确认库存是否成功修改。根据本地事务状态，返回相应的事务状态。

最后，需要在配置类中添加注解 @EnableTransactionManagement，启用事务管理。

@Configuration
@EnableTransactionManagement
public class TransactionConfig {

}

通过以上步骤，可以实现带有事务的订单方法调用 Spring Cloud Stream RocketMQ 修改库存，并确保事务提交和 RocketMQ 的消费顺序一致。