上周在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.gofunc 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.GoodsInfo if 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.Count g.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.gofunc (c *OrderTimeoutConsumer) HandleMessage(ctx context.Context, msg amqp.Delivery) error { // 解析订单超时事件 var event rabbitmq.OrderTimeoutEvent err := 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.gofunc ReturnStock(ctx context.Context, req *rabbitmq.OrderStockReturnEvent) ([]*rabbitmq.OrderGoodsInfo, error) { // 使用goroutine并发处理每个商品 resultChan := make(chan *rabbitmq.OrderGoodsInfo, len(req.GoodsInfo)) var wg sync.WaitGroup wg.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.GoodsInfo err := 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.OrderGoodsInfo for 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.GoodsInfo dao.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})
复制代码
难点 3：系统性能与用户体验平衡
问题：库存操作频繁，如何避免影响响应速度？
解决方案：
核心流程同步处理，确保用户体验
非核心操作（如缓存清理）异步化，不阻塞主流程
批量操作使用并发处理提升效率
结语
很多时候一些严重的错误往往出现在一些小细节上面。通过这次库存相关的优化案例可以发现：看似简单的业务流程，在分布式架构下需要考虑服务协作、并发控制、异常处理等等多个方面的因素。
通过引入RabbitMQ，不仅解决了已经存在的库存同步问题，更让整个系统具备了更好的扩展性，比如未来要新增物流通知、积分等功能的时候，只需新增事件的消费者就 ok 了，不需要再去修改现有的核心代码。
本文基于真实的 GoFrame 微服务电商项目，所有代码都经过生产环境验证，这里是项目的介绍：(https://mp.weixin.qq.com/s/ACzEHtvGh2YsU_4fxo83fQ)。如果你也遇到类似问题，欢迎交流讨论！
评论