@Transactional(rollbackFor = Exception.class) 和 RocketMQ 消息消费库存更新问题解决方案

@Transactional(rollbackFor = Exception.class) 和 RocketMQ 消息消费库存更新问题解决方案

问题描述： 在使用 @Transactional 注解管理事务，并使用 RocketMQ 消息机制进行库存更新时，如何确保消息消费时查询到的库存是扣减后的最新库存？

代码示例：

@Transactional(rollbackFor = Exception.class)
    TackAnOrderVo tackAnOrder(@RequestBody @Validated PlaceOrderVo vo){
   
//提交订单

//提交库存

/提交其它

   // 发送消息到 MQ
        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter(){
            @Override
            public void afterCommit() {
                //发送mq更新es库存标志
                for (Long drugId : drugIdList) {
                    DrugChangeMessage drugChangeMessage = new DrugChangeMessage();
                    drugChangeMessage.setDrugId(drugId);
                    drugChangeMessage.setStoreId(order.getStoreId());
                    drugChangeMessage.setChangeType(5);
                    drugChangeMessage.setMerchantId(Long.parseLong(merchantId));
                    drugChangeMessage.setEnterpriseId(Long.parseLong(enterpriseId));
                    drugChangeMessage.setPlatformId(Long.parseLong(platformId));
                    drugProducer.sendDrugChangeMsg(drugChangeMessage);
                }

            }
        });

}

解决方案： 为了确保消费的消息反映更新后的库存数量，您可以在消费者的服务中修改代码，以便在库存扣减完成后获取更新后的库存。实现此目的的一种方法是在查询库存之前添加延迟。以下是一个示例：

@Service
public class InventoryConsumerService {

    @Autowired
    private InventoryService inventoryService;

    @RocketMQMessageListener(
        topic = 'inventory_topic',
        consumerGroup = 'inventory_consumer_group'
    )
    public class InventoryMessageListener implements RocketMQListener<InventoryChangeMessage> {

        @Override
        public void onMessage(InventoryChangeMessage message) {
            // 等待延迟，然后再查询库存
            try {
                Thread.sleep(5000); // 延迟 5 秒（您可以根据需要调整延迟）
            } catch (InterruptedException e) {
                // 处理异常
            }

            // 获取更新后的库存
            Inventory inventory = inventoryService.getInventoryById(message.getInventoryId());

            // 使用更新后的库存处理消息
            // ...
        }
    }

}

在此示例中，消费者服务侦听来自 'inventory_topic' 主题和 'inventory_consumer_group' 消费者组的消息。在收到消息后，它会等待一段延迟（在本例中为 5 秒），然后再使用 inventoryService.getInventoryById() 方法获取更新后的库存。您可以根据需要调整延迟时间。

通过引入这种延迟，您可以确保在消费者服务中查询库存之前，库存已更新。