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

在 Spring Boot 中使用事务管理的订单方法调用 MQ 修改库存时，如果事务未提交，RocketMQ 可能会先消费消息。为了解决这个问题，可以使用 TransactionEventListener 来监听事务事件，并将 RocketMQ 的参数传递给该监听器。

首先，在 Spring Boot 的配置文件中配置 RocketMQ 的相关参数，例如：

rocketmq.name-server=127.0.0.1:9876
rocketmq.producer.group=myGroup
rocketmq.consumer.group=myGroup

然后，在订单方法上添加事务注解，例如：

@Service
@Transactional
public class OrderService {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Autowired
    private OrderRepository orderRepository;

    @Autowired
    private InventoryService inventoryService;

    public void createOrder(Order order) {
        // 创建订单

        // 修改库存
        inventoryService.updateInventory(order.getItemId(), order.getQuantity());

        // 发送 MQ 消息
        rocketMQTemplate.sendMessageInTransaction('myTopic', 'myTag', MessageBuilder.withPayload(order).build(), order);
    }
}

接下来，创建 TransactionEventListener 监听器，处理 RocketMQ 的事务事件：

@Component
public class RocketMQTransactionListener implements TransactionListener {

    @Autowired
    private OrderRepository orderRepository;

    @Override
    public LocalTransactionState executeLocalTransaction(Message message, Object arg) {
        // 在本地执行事务

        // 更新订单状态
        Order order = (Order) arg;
        order.setStatus('COMMITTED');
        orderRepository.save(order);

        // 返回事务状态
        return LocalTransactionState.COMMIT_MESSAGE;
    }

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

        // 从数据库中查询订单状态
        String orderId = messageExt.getTransactionId();
        Order order = orderRepository.findById(orderId).orElse(null);

        // 根据订单状态返回事务状态
        if (order != null && 'COMMITTED'.equals(order.getStatus())) {
            return LocalTransactionState.COMMIT_MESSAGE;
        } else {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }
}

最后，在 Spring Boot 的配置类中配置 RocketMQTemplate，并设置 TransactionListener：

@Configuration
public class RocketMQConfig {

    @Autowired
    private RocketMQTransactionListener transactionListener;

    @Bean
    public RocketMQTemplate rocketMQTemplate(DefaultMQProducer defaultMQProducer) {
        RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
        rocketMQTemplate.setProducer(defaultMQProducer);
        rocketMQTemplate.setTransactionListener(transactionListener);
        return rocketMQTemplate;
    }

    @Bean
    public DefaultMQProducer defaultMQProducer(@Value('${rocketmq.name-server}') String nameServer,
                                               @Value('${rocketmq.producer.group}') String producerGroup) throws MQClientException {
        DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
        producer.setNamesrvAddr(nameServer);
        producer.start();
        return producer;
    }

    @Bean
    public TransactionMQProducer transactionMQProducer(@Value('${rocketmq.name-server}') String nameServer,
                                                       @Value('${rocketmq.producer.group}') String producerGroup) throws MQClientException {
        TransactionMQProducer producer = new TransactionMQProducer(producerGroup);
        producer.setNamesrvAddr(nameServer);
        producer.setTransactionListener(transactionListener);
        producer.start();
        return producer;
    }
}

通过以上配置，当订单方法调用 MQ 修改库存时，如果事务未提交，RocketMQ 会先消费消息，并将消息传递给 TransactionEventListener 处理事务事件。在 TransactionEventListener 中，可以根据订单状态来决定是提交消息还是回滚消息。