SpringBoot 事务消息：如何保证ES库存和数据库一致性

在使用 Spring Boot 开发电商系统时，经常会遇到订单方法修改数据库库存，然后调用 Spring Cloud Stream 发送 RocketMQ 消息来修改 ES 库存，导致事务未提交，RocketMQ 先消费了，ES 的库存和数据库不一致的情况。如何保证事务提交后，MQ 才消费内容？

要保证事务提交后，MQ 才消费，可以使用 MQ 的事务消息功能来实现。以下是一种可能的解决方案：

在 Spring Boot 的订单方法上添加事务注解，确保订单操作和库存修改是在同一个事务中进行。

@Transactional
public void updateOrderAndStock(Order order) {
    // 修改订单状态
    orderDao.updateOrder(order);
    // 修改库存
    stockDao.updateStock(order.getProductId(), order.getAmount());
}

配置 Spring Cloud Stream 与 RocketMQ 的集成，确保消息发送和消费的可靠性。使用 RocketMQ 的事务消息功能，配置事务生产者和事务消费者。
在订单方法中，通过 RocketMQ 的事务消息发送器发送消息，消息内容包含需要修改的 ES 库存信息。

@Autowired
private RocketMQTemplate rocketMQTemplate;

public void updateOrderAndStock(Order order) {
    // 修改订单状态
    orderDao.updateOrder(order);
    // 修改库存
    stockDao.updateStock(order.getProductId(), order.getAmount());

    // 发送RocketMQ事务消息
    rocketMQTemplate.sendMessageInTransaction('topic', 'tag', MessageBuilder.withPayload(esStockInfo).build(), null);
}

编写 RocketMQ 的事务监听器，监听并处理事务消息的提交与回滚。

@Component
public class RocketMQTransactionListener implements RocketMQLocalTransactionListener {

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        // 执行ES库存修改操作
        try {
            // 修改ES库存
            esStockDao.updateStock(esStockInfo);
            // 提交事务
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            // 回滚事务
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        // 根据消息状态判断事务是否提交或回滚
        // 如果ES库存修改成功，则事务提交，否则事务回滚
        if (esStockDao.checkStockUpdateStatus(message.getPayload())) {
            return RocketMQLocalTransactionState.COMMIT;
        } else {
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }
}

通过以上步骤，可以实现在订单方法事务提交后再消费 MQ 消息，从而确保 ES 库存和数据库的一致性。