当前位置：首页 > news >正文

订单支付后库存不扣减，如何用RabbitMQ来优化？

news 2025/11/5 8:06:00

上周在Review学员代码的时候，我们发现了一个很基础但很重要的问题：支付回调流程中缺少了库存扣减环节。这类问题虽然基础，但如果直接进入生产环境，可能导致库存的数据和实际销售的情况不一致，出现超卖的情况。能够及时发现这种问题，这就是Review代码的重要性。

先看这段有问题的代码：

// 原来的支付回调逻辑（问题代码）
func PaymentCallback(ctx context.Context, orderID uint32) error {// 只更新订单状态为已支付_, err := dao.OrderInfo.Ctx(ctx).Where("id=?", orderID).Data(g.Map{"status": consts.OrderStatusPaid}).Update()if err != nil {return err}// 缺少库存扣减逻辑！商品库存还是原样return nil
}

这个问题的核心在于流程设计的不完整，用户支付成功后只是更新了订单状态，却没有同步调整商品库存，可能导致其他用户购买时看到的库存数据不正确。

想要解决这个问题，需要补充缺失的逻辑，更要考虑分布式系统下的流程合理性，这里我们选择引入RabbitMQ实现事件驱动架构，既能解决当前问题，也能方便后续的业务扩展。

问题分析

业务逻辑理解不正确

原逻辑对订单流程的理解是"创建订单→支付成功→完成交易"，但正确的流程应该要包含库存相关的环节：

创建订单→预扣库存→支付成功→确认交易→后续处理

不同服务之间的协作

在微服务架构中：

订单服务负责订单状态流转
商品服务负责库存数据维护

两个服务需要通过规范的协作机制保证数据一致性，而不是简单的同步调用。

解决方案

我们重新设计了包含库存管理的订单流程，通过RabbitMQ实现服务间的解耦通信：

创建订单时预扣库存

将库存扣减提前到订单创建的阶段，通过数据库事务保证操作的原子性：

// app/goods/internal/logic/goods_info/goods_info.go
func ReduceStock(ctx context.Context, req *rabbitmq.OrderCreatedEvent) error {// 使用数据库事务确保原子性err := g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {for _, goods := range req.GoodsInfo {// 1. 查询当前库存var goodsInfo entity.GoodsInfoif err := dao.GoodsInfo.Ctx(ctx).TX(tx).Where("id = ?", goods.GoodsId).Fields("stock").Scan(&goodsInfo); err != nil {return gerror.Wrapf(err, "查询商品{%d}库存失败", goods.GoodsId)}// 2. 判断库存是否足够if goodsInfo.Stock < goods.Count {return gerror.Newf("商品{%d}库存不足(当前:%d, 需要:%d)", goods.GoodsId, goodsInfo.Stock, goods.Count)}// 3. 扣减库存newStock := goodsInfo.Stock - goods.Countg.Log().Infof(ctx, "商品{%d}新库存:%d", goods.GoodsId, newStock)if _, err := dao.GoodsInfo.Ctx(ctx).TX(tx).Where("id = ?", goods.GoodsId).Data(g.Map{"stock": newStock}).Update(); err != nil {return gerror.Wrapf(err, "更新商品{%d}库存失败", goods.GoodsId)}}return nil})return err
}

设计思路：

提前锁定库存，避免支付过程中商品被重复购买
事务保证库存检查与扣减的原子性，防止并发问题
库存不足时直接阻断订单创建，提升用户体验

支付成功后的确认处理

支付完成后，通过事件通知触发后续清理工作：

// 支付回调逻辑
func PaymentCallback(ctx context.Context, orderID uint32) error {// 1. 更新订单状态_, err := dao.OrderInfo.Ctx(ctx).Where("id=?", orderID).Data(g.Map{"status": consts.OrderStatusPaid}).Update()if err != nil {return err}// 2. 获取订单详情（包含商品信息）orderDetail, err := GetOrderDetail(ctx, orderID)if err != nil {return err}// 3. 发布库存确认事件（这里库存已在创建订单时预扣）// 主要是清理缓存等后续操作go func() {// 异步清理商品缓存if err := goodsRedis.DeleteKeys(context.Background(), orderDetail.GoodsIDs); err != nil {g.Log().Errorf(ctx, "清理商品缓存失败: %v", err)}}()return nil
}

订单超时的库存返还机制

为避免用户下单后未支付导致库存长时间锁定，设计超时返还逻辑：

// app/order/utility/consumer/order_timeout_consumer.go
func (c *OrderTimeoutConsumer) HandleMessage(ctx context.Context, msg amqp.Delivery) error {// 解析订单超时事件var event rabbitmq.OrderTimeoutEventerr := rabbitmq.UnmarshalEvent(msg.Body, &event)if err != nil {return err}// 判断是否真正超时（30分钟未支付）eventTime, _ := time.Parse(time.RFC3339, event.TimeStamp)if time.Now().After(eventTime.Add(30 * time.Minute)) {// 处理订单超时err = order_info.HandleOrderTimeoutResult(ctx, event.OrderId)if err != nil {return err}// 发布库存返还事件eventReq, err := order_info.GetOrderDetail(ctx, event.OrderId)if err == nil {go rabbitmq.PublishReturnStockEvent(event.OrderId, eventReq)}}return nil
}

库存返还的具体实现

通过并发处理提升库存返还效率：

// app/goods/internal/logic/goods_info/goods_info.go
func ReturnStock(ctx context.Context, req *rabbitmq.OrderStockReturnEvent) ([]*rabbitmq.OrderGoodsInfo, error) {// 使用goroutine并发处理每个商品resultChan := make(chan *rabbitmq.OrderGoodsInfo, len(req.GoodsInfo))var wg sync.WaitGroupwg.Add(len(req.GoodsInfo))for _, stockInfo := range req.GoodsInfo {go func(stockInfo *rabbitmq.OrderGoodsInfo) {defer wg.Done()defer func() {if r := recover(); r != nil {g.Log().Errorf(ctx, "库存返还panic: %v", r)}}()// 查询当前库存var goodsInfo entity.GoodsInfoerr := dao.GoodsInfo.Ctx(ctx).Where("id=?", stockInfo.GoodsId).Fields("stock").Scan(&goodsInfo)if err != nil {resultChan <- &rabbitmq.OrderGoodsInfo{GoodsId: stockInfo.GoodsId,Count:   stockInfo.Count,}return}// 返还库存newStock := goodsInfo.Stock + stockInfo.Count_, err = dao.GoodsInfo.Ctx(ctx).Where("id=?", stockInfo.GoodsId).Data(g.Map{"stock": newStock}).Update()if err != nil {resultChan <- &rabbitmq.OrderGoodsInfo{GoodsId: stockInfo.GoodsId,Count:   stockInfo.Count,}return}g.Log().Infof(ctx, "商品{%d}库存返还成功，新库存:%d", stockInfo.GoodsId, newStock)}(stockInfo)}wg.Wait()close(resultChan)// 收集处理结果var resultArr []*rabbitmq.OrderGoodsInfofor res := range resultChan {resultArr = append(resultArr, res)}return resultArr, nil
}

消息队列的事件驱动架构

定义核心事件实现服务解耦：

// 事件定义
type OrderCreatedEvent struct {OrderId   uint32             `json:"order_id"`GoodsInfo []*OrderGoodsInfo `json:"goods_info"`
}type OrderStockReturnEvent struct {OrderId   uint32             `json:"order_id"`GoodsInfo []*OrderGoodsInfo `json:"goods_info"`
}

事件流设计：

用户下单→OrderCreated事件→商品服务扣减库存↓
支付超时→OrderTimeout事件→商品服务返还库存↓
支付成功→订单状态更新+缓存清理

技术难点与解决方案

难点1：分布式系统的数据一致性

问题：订单与库存数据分属不同服务，如何保证操作协同？

解决方案：

采用最终一致性模型，通过事件重试确保数据对齐
每个事件处理都设计幂等性，避免重复执行导致错误

难点2：高并发下的库存准确性

问题：多用户同时购买时如何防止库存数据混乱？

解决方案：

// 数据库事务+行级锁保证并发安全
err := g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {// 事务内查询自动加行锁，阻止并发修改var goodsInfo entity.GoodsInfodao.GoodsInfo.Ctx(ctx).TX(tx).Where("id=?", goodsId).Scan(&goodsInfo)// 检查并更新库存if goodsInfo.Stock >= count {dao.GoodsInfo.Ctx(ctx).TX(tx).Data(g.Map{"stock": goodsInfo.Stock - count}).Update()}return nil
})