Spring Boot 事务与 RocketMQ 消费：避免事务未提交导致消息提前消费

本文将探讨在使用 Spring Boot 和 RocketMQ 进行订单处理时，如何确保事务正确提交，避免事务未提交导致 RocketMQ 消息提前消费的问题。

问题描述

当使用 Spring Cloud Stream 和 RocketMQ 进行消息发送和消费时，如果涉及到事务操作，可能会出现事务未提交，而 RocketMQ 消息却先被消费的情况。例如，在创建订单时需要修改库存，并且将订单信息发送到 RocketMQ，如果在库存修改成功之前，RocketMQ 消费者已经消费了消息，就会导致数据不一致。

解决方法

为了解决这个问题，需要采取以下措施确保事务正确提交：

1. 添加 @Transactional 注解

在订单方法上添加 @Transactional 注解，以启用事务管理。如果方法中的某些操作需要在事务内执行，可以将这些操作封装在一个独立的方法中，并在订单方法中调用该方法。

2. 禁用自动提交消费进度

在 RocketMQ 消费者配置中设置 rocketmq.consumer.enable-auto-commit 属性为 false，以禁用自动提交消费进度。这样可以确保消息消费后不会立即提交消费进度，而是等待事务成功提交后再进行提交。

3. 手动提交消费进度

在订单方法中的事务提交之前，需要手动提交 RocketMQ 的消费进度。可以通过注入 RocketMQTransactionService 来实现，然后在事务提交前调用 transactionService.commit(transactionId) 方法来提交消费进度。

示例代码

import org.apache.rocketmq.spring.core.RocketMQTransactionService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Service
public class OrderService {

    @Autowired
    private RocketMQTransactionService transactionService;

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

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

        // 提交 RocketMQ 消费进度
        transactionService.commit(transactionId);
    }

    @Transactional
    public void updateInventory() {
        // 修改库存逻辑
    }
}

注意事项

• transactionId 是一个唯一的事务 ID，可以根据实际情况生成。
• 确保 updateInventory() 方法中的逻辑能够在事务内完成，并且能够在事务提交前获得 transactionId。

总结

通过以上步骤，可以确保订单方法和库存修改方法都在同一个事务内执行，并且在事务成功提交后再提交 RocketMQ 的消费进度，避免了事务未提交而导致 RocketMQ 先消费的问题。