【分布式】如何使用RocketMQ实现下单-库存-支付这个场景的分布式事务问题

news2025/3/14 14:50:27

在 下单-库存-支付 场景中，通过消息队列实现最终一致性，需保证三个微服务的操作最终一致，且在支付失败或库存不足时触发回滚补偿。以下是具体实现方案：

1. 整体流程设计

正常流程（成功场景）

订单服务 创建订单（状态为待支付），发送 订单创建成功事件。
库存服务 消费事件，扣减库存，发送 库存扣减成功事件。
支付服务 消费事件，执行支付，发送 支付成功事件。
订单服务 消费支付成功事件，更新订单状态为已完成。

异常流程（失败场景）

• 库存不足：库存服务直接发送 库存不足事件，订单服务取消订单。
• 支付失败：支付服务发送 支付失败事件，触发库存回滚和订单取消。

2. 核心组件与消息设计

消息队列（以RocketMQ为例）

事件类型	Topic	生产者	消费者	说明
订单创建成功事件	`order_created`	订单服务	库存服务	触发库存扣减
库存扣减成功事件	`stock_reduced`	库存服务	支付服务	触发支付
支付成功事件	`payment_done`	支付服务	订单服务	完成订单
库存不足事件	`stock_failed`	库存服务	订单服务	取消订单
支付失败事件	`payment_failed`	支付服务	订单/库存服务	触发库存回滚和订单取消

数据表设计（关键字段）

• 订单表（Order）

CREATE TABLE orders (
  id VARCHAR(64) PRIMARY KEY,  -- 订单ID（全局唯一）
  user_id BIGINT,
  amount DECIMAL(10,2),
  status ENUM('pending', 'completed', 'canceled')  -- 订单状态
);

• 库存表（Inventory）

CREATE TABLE inventory (
  product_id BIGINT PRIMARY KEY,
  stock INT CHECK(stock >= 0)  -- 库存不可为负
);

3. 实现步骤

(1) 订单服务：创建订单并发送事件

// 订单服务（OrderService.java）
@Transactional
public void createOrder(Order order) {
    // 1. 检查参数合法性（如用户存在性）
    // 2. 插入订单记录（状态为pending）
    orderDao.insert(order);
    
    // 3. 发送订单创建成功事件（事务消息）
    Message msg = new Message("order_created", JSON.toJSONBytes(order));
    TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction(
        "order_group", 
        msg, 
        order.getId()  // 事务ID（用订单ID标识）
    );
    
    // 4. 若消息发送失败，抛出异常触发事务回滚
    if (result.getLocalTransactionState() != LocalTransactionState.COMMIT_MESSAGE) {
        throw new RuntimeException("订单创建事件发送失败");
    }
}

(2) 库存服务：扣减库存

// 库存服务（InventoryService.java）
@RocketMQMessageListener(topic = "order_created", consumerGroup = "inventory_group")
public class OrderCreatedListener implements RocketMQListener<MessageExt> {
    @Override
    @Transactional
    public void onMessage(MessageExt message) {
        Order order = JSON.parseObject(message.getBody(), Order.class);
        
        // 1. 检查库存是否充足
        Inventory inventory = inventoryDao.selectByProductId(order.getProductId());
        if (inventory.getStock() < order.getQuantity()) {
            // 发送库存不足事件
            rocketMQTemplate.sendOneWay("stock_failed", JSON.toJSONBytes(order));
            return;
        }
        
        // 2. 扣减库存（乐观锁防止超卖）
        int updated = inventoryDao.reduceStock(
            order.getProductId(), 
            order.getQuantity(), 
            inventory.getVersion()
        );
        if (updated == 0) {
            throw new RetryException("库存扣减冲突，稍后重试"); // 触发消息重试
        }
        
        // 3. 发送库存扣减成功事件
        rocketMQTemplate.sendOneWay("stock_reduced", JSON.toJSONBytes(order));
    }
}

(3) 支付服务：执行支付

// 支付服务（PaymentService.java）
@RocketMQMessageListener(topic = "stock_reduced", consumerGroup = "payment_group")
public class StockReducedListener implements RocketMQListener<MessageExt> {
    @Override
    @Transactional
    public void onMessage(MessageExt message) {
        Order order = JSON.parseObject(message.getBody(), Order.class);
        
        // 1. 调用第三方支付接口
        boolean success = paymentClient.pay(order.getUserId(), order.getAmount());
        if (!success) {
            // 发送支付失败事件
            rocketMQTemplate.sendOneWay("payment_failed", JSON.toJSONBytes(order));
            return;
        }
        
        // 2. 发送支付成功事件
        rocketMQTemplate.sendOneWay("payment_done", JSON.toJSONBytes(order));
    }
}

(4) 订单服务：处理成功/失败事件

// 订单服务（OrderService.java）
@RocketMQMessageListener(topic = "payment_done", consumerGroup = "order_complete_group")
public class PaymentDoneListener implements RocketMQListener<MessageExt> {
    @Override
    @Transactional
    public void onMessage(MessageExt message) {
        Order order = JSON.parseObject(message.getBody(), Order.class);
        orderDao.updateStatus(order.getId(), "completed");
    }
}

@RocketMQMessageListener(topic = {"stock_failed", "payment_failed"}, consumerGroup = "order_cancel_group")
public class OrderCancelListener implements RocketMQListener<MessageExt> {
    @Override
    @Transactional
    public void onMessage(MessageExt message) {
        Order order = JSON.parseObject(message.getBody(), Order.class);
        
        // 1. 取消订单（状态置为canceled）
        orderDao.updateStatus(order.getId(), "canceled");
        
        // 2. 若事件来自支付失败，需触发库存回滚（发送库存回滚事件）
        if (message.getTopic().equals("payment_failed")) {
            rocketMQTemplate.sendOneWay("stock_rollback", JSON.toJSONBytes(order));
        }
    }
}

(5) 库存服务：处理回滚事件

// 库存服务（InventoryService.java）
@RocketMQMessageListener(topic = "stock_rollback", consumerGroup = "inventory_rollback_group")
public class StockRollbackListener implements RocketMQListener<MessageExt> {
    @Override
    @Transactional
    public void onMessage(MessageExt message) {
        Order order = JSON.parseObject(message.getBody(), Order.class);
        inventoryDao.addStock(order.getProductId(), order.getQuantity());
    }
}