Golang并发请求下商品库存扣减异常问题求助

问题描述 #

我的项目支持用户购买商品，要求商品库存需大于等于购买请求的数量；若满足条件，先保存购买记录，再更新对应商品的库存。但当通过Golang应用发送20个并发请求（每个请求购买20件库存为100的A商品）时，出现异常：生成了20条购买记录，库存被更新为60，而正常情况下应在5次请求后因库存不足报错。以下是处理请求的方法（仅调用GORM的保存和更新操作），无法调整方法实现预期效果，该如何解决？

var wg sync.WaitGroup
wg.Add(1)

errs := make(chan error, 1)
go func(ctx context.Context, request *dto.CreateTicketPurchaseRequest) {
    defer wg.Done()
    productStockCount, err := t.stockRepo.GetTicketStockCount(productId) // first check available stock count
    if err != nil {
        errs <- err
        return
    }

    if productStockCount.StockCount < payload.Quantity {
        err = fmt.Errorf("no enough products")
        errs <- err
        return
    }

    u := new(entity.TicketPurchases)
    u.Quantity = payload.Quantity
    u.UserId = payload.UserId
    u.StockId = productId
    err = t.purchaseRepo.PurchaseTicket(u) // if stock count is ok, then purchase product
    if err != nil {
        errs <- err
        return
    }

    productStockCount.StockCount = productStockCount.StockCount - u.Quantity
    err = t.stockRepo.RemoveStockCount(productStockCount) // update stock count for product

    if err != nil {
        errs <- err
        return
    }
    close(errs)

}(ctx, payload)
wg.Wait()
if err = <-errs; err != nil {
    return c.JSON(http.StatusBadRequest, nil)
}

解决方案 #

1. 数据库事务+行级悲观锁 #

核心思路是在查询库存时就给数据行加锁，阻止其他事务读取或修改，确保整个购买流程（库存检查→创建记录→扣减库存）的原子性。

• 修改库存查询方法，添加行级锁：

// stockRepo.GetTicketStockCount 实现示例
func (r *stockRepo) GetTicketStockCount(productId uint) (*entity.ProductStock, error) {
    var stock entity.ProductStock
    // 使用FOR UPDATE加锁，其他事务必须等待当前事务提交后才能操作该行
    err := r.db.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id = ?", productId).First(&stock).Error
    return &stock, err
}

• 将整个业务逻辑包裹在数据库事务中：

tx := t.db.Begin()
defer func() {
    if r := recover(); r != nil {
        tx.Rollback()
    }
}()

if tx.Error != nil {
    errs <- tx.Error
    return
}

// 加锁查询库存，避免并发篡改
productStockCount, err := t.stockRepo.WithTx(tx).GetTicketStockCount(productId)
if err != nil {
    tx.Rollback()
    errs <- err
    return
}

if productStockCount.StockCount < payload.Quantity {
    tx.Rollback()
    errs <- fmt.Errorf("no enough products")
    return
}

// 创建购买记录
u := new(entity.TicketPurchases)
u.Quantity = payload.Quantity
u.UserId = payload.UserId
u.StockId = productId
if err := t.purchaseRepo.WithTx(tx).PurchaseTicket(u); err != nil {
    tx.Rollback()
    errs <- err
    return
}

// 扣减库存
productStockCount.StockCount -= u.Quantity
if err := t.stockRepo.WithTx(tx).RemoveStockCount(productStockCount); err != nil {
    tx.Rollback()
    errs <- err
    return
}

if err := tx.Commit().Error; err != nil {
    errs <- err
    return
}
close(errs)

注：需要让Repo支持传入事务对象（实现WithTx方法），GORM可通过tx.Scopes()或直接传递tx实例来复用事务上下文。

2. 乐观锁（版本号机制） #

通过给库存表添加版本号字段，每次更新时校验版本号，确保只有当前版本的记录被修改，避免并发覆盖。适合并发量中等的场景。

• 库存实体添加版本号字段：

type ProductStock struct {
    ID         uint
    StockCount int
    Version    int // 乐观锁版本号，每次更新自增1
}

• 修改库存扣减方法，加入版本号校验：

// stockRepo.RemoveStockCount 实现示例
func (r *stockRepo) RemoveStockCount(stock *entity.ProductStock) error {
    // GORM会自动处理版本号校验，若版本不匹配则返回ErrRecordNotFound
    return r.db.Model(stock).Where("version = ?", stock.Version).Updates(map[string]interface{}{
        "stock_count": stock.StockCount,
        "version":     stock.Version + 1,
    }).Error
}

• 业务逻辑中增加重试机制处理并发冲突：

maxRetries := 3
var err error

for i := 0; i < maxRetries; i++ {
    productStockCount, err := t.stockRepo.GetTicketStockCount(productId)
    if err != nil {
        errs <- err
        return
    }

    if productStockCount.StockCount < payload.Quantity {
        errs <- fmt.Errorf("no enough products")
        return
    }

    // 创建购买记录
    u := new(entity.TicketPurchases)
    u.Quantity = payload.Quantity
    u.UserId = payload.UserId
    u.StockId = productId
    if err := t.purchaseRepo.PurchaseTicket(u); err != nil {
        errs <- err
        return
    }

    productStockCount.StockCount -= u.Quantity
    err = t.stockRepo.RemoveStockCount(productStockCount)
    
    // 更新成功则结束流程
    if err == nil {
        close(errs)
        return
    }

    // 乐观锁冲突（版本不匹配）则重试
    if errors.Is(err, gorm.ErrRecordNotFound) {
        continue
    }

    errs <- err
    return
}

errs <- fmt.Errorf("too many concurrent requests, please try again later")

3. 原子更新语句（数据库层面直接校验+扣减） #

跳过先查询库存的步骤，用一条SQL完成库存检查和扣减，直接利用数据库的原子性保证操作安全，是最高效的方案。

• 新增库存扣减方法：

// stockRepo.TryDeductStock 尝试扣减库存，返回是否成功
func (r *stockRepo) TryDeductStock(productId uint, quantity int) (bool, error) {
    result := r.db.Model(&entity.ProductStock{}).
        Where("id = ? AND stock_count >= ?", productId, quantity).
        UpdateColumn("stock_count", gorm.Expr("stock_count - ?", quantity))
    
    if result.Error != nil {
        return false, result.Error
    }
    // RowsAffected>0说明扣减成功，否则库存不足
    return result.RowsAffected > 0, nil
}

• 调整业务逻辑：

// 先尝试原子扣减库存
success, err := t.stockRepo.TryDeductStock(productId, payload.Quantity)
if err != nil {
    errs <- err
    return
}
if !success {
    errs <- fmt.Errorf("no enough products")
    return
}

// 库存扣减成功后创建购买记录
u := new(entity.TicketPurchases)
u.Quantity = payload.Quantity
u.UserId = payload.UserId
u.StockId = productId
err = t.purchaseRepo.PurchaseTicket(u)
if err != nil {
    // 这里需要添加补偿机制，比如回滚库存或记录异常待人工处理
    t.stockRepo.AddStock(productId, payload.Quantity)
    errs <- err
    return
}

close(errs)

内容的提问来源于stack exchange，提问作者oakkose