全部标签
写点什么
开源架构OpenHarmony算法元宇宙学习方法Web3.0高效工作数据库Python音视频前端AI大数据团队管理程序员运维深度思考低代码redisgolang微服务架构flutter 查看更多

库存防超卖（Redis Lua+ 分布式锁对比实践）

作者：王中阳Go
  • 2025-11-19
    北京

  • 本文字数：8294 字

    阅读完需：约 27 分钟

库存防超卖（Redis Lua+分布式锁对比实践）

这篇文章的内容都是基于我们GoFrame微服务电商项目的实践，感兴趣的朋友可以点击查看

1. 引言

在电商系统中，库存管理是一个至关重要的环节，特别是在高并发场景下（如秒杀、限时抢购等），如何保证库存的准确性，避免超卖现象，是系统稳定性和用户体验的关键。本文档详细介绍库存超卖问题，分析现有的解决方案，并通过实践对比 Redis Lua 脚本和分布式锁两种方案在库存扣减场景下的优缺点，提供完整的实现代码和最佳实践建议。

2. 库存超卖问题分析

2.1 什么是库存超卖

库存超卖是指系统在高并发情况下，实际销售的商品数量超过了系统中记录的库存数量。这可能导致商家无法履行订单，造成用户投诉和经济损失，严重影响平台信誉。

2.2 超卖问题产生的原因

在传统的库存管理逻辑中，通常包括以下步骤：


  1. 查询当前库存数量

  2. 判断库存是否足够

  3. 如果足够，则扣减库存


在高并发场景下，由于多个请求同时访问数据库或缓存，可能会出现以下情况：


请求A: 查询库存 -> 库存为10请求B: 查询库存 -> 库存为10请求A: 扣减库存 -> 库存变为9请求B: 扣减库存 -> 库存变为8
复制代码


如果库存只有 1 个商品，但同时有 10 个请求查询到库存为 1，那么这 10 个请求都会执行扣减操作，导致库存变为负数。这种情况在秒杀等高并发场景下尤为常见。

2.3 现有系统中的库存管理分析

在我们的电商系统中，传统的库存扣减实现虽然使用了数据库事务来保证原子性，但在高并发场景下，仍然可能出现超卖问题。主要原因是：


  1. 数据库层的锁粒度较粗，可能导致性能瓶颈

  2. 数据库连接数有限，在大量并发请求下可能成为系统瓶颈

  3. 网络延迟和多线程并发操作导致的竞态条件

3. 库存防超卖解决方案

3.1 解决方案概述

为了解决库存超卖问题，我们实现了两种常见的解决方案：


  1. 基于 Redis 分布式锁的库存扣减：使用 Redis 实现分布式锁，确保同一时间只有一个请求能够扣减特定商品的库存

  2. 基于 Redis Lua 脚本的库存扣减：利用 Redis Lua 脚本的原子性，将库存检查和扣减操作在 Redis 端原子执行

3.2 Redis 分布式锁方案

分布式锁是解决分布式系统中并发控制的一种常用机制。在库存扣减场景中，我们使用 Redis 实现分布式锁，确保同一时间只有一个请求能够扣减特定商品的库存。

3.2.1 分布式锁的实现原理

使用 Redis 的SET命令的NX选项来实现分布式锁。当多个客户端同时设置同一个键时，只有一个能成功，这样就实现了互斥锁的效果。锁的 value 使用随机字符串生成，确保只有持有锁的客户端才能释放锁。

3.2.2 分布式锁的关键问题

  1. 锁的过期时间：防止锁持有方崩溃导致锁无法释放

  2. 锁的续期：对于长时间运行的操作，需要自动续期以避免锁过期

  3. 锁的释放：确保只有持有锁的客户端能够释放锁，避免误释放

  4. 锁的重试：在获取锁失败时，合理的重试策略可以提高成功率

3.3 Redis Lua 脚本方案

Redis Lua 脚本可以在 Redis 服务器端原子执行多条命令，避免了在客户端和服务器之间多次通信带来的竞态条件。

3.3.1 Lua 脚本的优势

  1. 原子性：脚本中的所有命令要么全部执行，要么全部不执行

  2. 减少网络开销：将多条命令合并为一个脚本发送到 Redis 服务器

  3. 减少客户端逻辑：将复杂的业务逻辑放在脚本中，简化客户端代码

  4. 消除竞态条件：在 Redis 服务器端执行，避免了多客户端并发操作的竞态条件

3.3.2 库存扣减的 Lua 脚本思路

编写一个 Lua 脚本，在脚本中完成以下操作：


  1. 获取当前商品的库存

  2. 判断库存是否足够

  3. 如果足够，扣减库存并返回成功

  4. 如果不足，返回失败

4. 项目实现结构

shop-goframe-micro-service-refacotor/├── app/│   └── goods/│       ├── utility/│       │   └── stock/│       │       ├── stock.go           # 库存管理器接口定义│       │       ├── distributed_lock.go # 基于分布式锁的实现│       │       ├── redis_lua.go        # 基于Lua脚本的实现│       │       └── stock_test.go       # 对比测试代码└── doc/    └── 库存防超卖（Redis Lua+分布式锁对比实践）.md  # 本文档
复制代码

5. 两种方案对比分析

5.1 技术原理对比

5.2 性能对比

5.3 可靠性对比

5.4 适用场景对比

6.最佳实践建议

6.1 方案选择建议

  1. 基于业务复杂度选择


  • 简单的库存扣减：优先使用 Lua 脚本方案

  • 复杂业务逻辑（需要查询其他服务、执行多个步骤）：使用分布式锁方案

  • 基于并发量选择

  • 高并发场景（如秒杀、抢购）：强制使用 Lua 脚本

  • 低并发场景：两种方案均可，可根据维护成本选择

  • 基于团队技术栈选择

  • 如果团队熟悉 Lua：优先考虑 Lua 脚本方案

  • 如果团队对分布式锁理解更深入：可以选择分布式锁方案

  • 混合使用策略

  • 对于核心的库存扣减操作：使用 Lua 脚本确保性能和原子性

  • 对于需要协调多个资源的复杂业务：使用分布式锁

6.2 实现注意事项

分布式锁实现注意事项：


  1. 确保使用 SET 命令的 NX 选项来实现互斥性，同时设置过期时间

  2. 必须设置合理的锁超时时间，避免死锁（建议 2-5 秒）

  3. 使用 Lua 脚本释放锁，确保锁的原子性释放，避免误删

  4. 实现锁重试机制，使用指数退避算法提高锁获取成功率

  5. 考虑使用 Redis 集群提高可用性，避免单点故障

  6. 使用 UUID 或随机字符串作为锁的值，确保唯一性

  7. 实现锁自动续期机制（可选），对于长时间运行的操作


Lua 脚本实现注意事项：


  1. 保持 Lua 脚本简洁，避免在脚本中执行复杂逻辑

  2. 脚本执行超时时间设置合理（建议 2-5 秒）

  3. 错误处理要完善，包括 Redis 连接错误和库存不足情况

  4. 考虑在脚本中加入合理的日志或监控信息

  5. 优化 Lua 脚本性能，避免在脚本中使用过多的条件判断和循环

  6. 预先加载 Lua 脚本到 Redis，减少网络传输开销

6.3 性能优化建议

  1. Redis 连接优化


  • 使用连接池管理 Redis 连接

  • 合理设置连接池大小和超时时间

  • 考虑使用 Redis Sentinel 或 Redis Cluster 提高可用性

  • 缓存策略优化

  • 库存预热：系统启动时将热点商品库存加载到 Redis

  • 本地缓存：对非热点数据使用本地缓存减少 Redis 访问

  • 分级缓存：对不同热度的商品使用不同的缓存策略

  • 并发控制优化

  • 限流措施：在 API 网关层实施限流，保护库存服务

  • 熔断降级：当 Redis 服务不可用时，降级到数据库操作

  • 削峰填谷：使用消息队列处理高并发请求，如 RabbitMQ

  • 监控与告警

  • 监控 Redis 内存使用情况和性能指标

  • 监控库存操作的成功率、响应时间和错误率

  • 对异常情况（如频繁的库存不足）设置告警阈值

  • 实现操作审计日志，方便问题排查

6.4 运维建议

  1. Redis 高可用配置


  • 使用 Redis 主从复制架构

  • 配置 Redis 哨兵或集群模式

  • 定期备份 Redis 数据，确保数据可恢复性

  • 故障演练

  • 定期进行 Redis 故障切换演练

  • 模拟 Redis 连接异常，验证系统恢复能力

  • 测试库存扣减异常情况下的系统行为

  • 容量规划

  • 根据业务增长预测，提前规划 Redis 容量

  • 监控 Redis 性能指标，及时扩容

  • 考虑使用 Redis 集群实现水平扩展

  • 安全考虑

  • 配置 Redis 访问密码和 IP 白名单

  • 避免在 Redis 中存储敏感信息

  • 定期更新 Redis 版本，修复安全漏洞

7. 代码实现解析

7.1 接口设计

我们设计了统一的StockManager接口，使得两种实现可以无缝切换：


// StockManager 库存管理器接口type StockManager interface {   // ReduceStock 扣减库存   // 返回值：是否成功扣减，错误信息   ReduceStock(ctx context.Context, goodsId uint32, count int) (bool, error)
   // ReturnStock 返还库存   // 返回值：是否成功返还，错误信息   ReturnStock(ctx context.Context, goodsId uint32, count int) (bool, error)
   // GetStock 获取当前库存   // 返回值：当前库存数量，错误信息   GetStock(ctx context.Context, goodsId uint32) (int, error)
   // InitStock 初始化库存   // 返回值：是否成功初始化，错误信息   InitStock(ctx context.Context, goodsId uint32, count int) (bool, error)}
复制代码

7.2 分布式锁实现详解

分布式锁实现的核心是DistributedLockStockManager结构体，它包含以下关键方法：


  1. 锁的获取：使用 SET 命令的 NX 选项，设置过期时间避免死锁

  2. 锁的释放：使用 Lua 脚本确保只有持有锁的客户端才能释放锁

  3. 库存操作：在获取锁后执行库存的扣减、返还等操作


核心实现代码如下：


// acquireLock 获取分布式锁func (m *DistributedLockStockManager) acquireLock(ctx context.Context, goodsId uint32) (string, error) {   lockKey := m.getLockKey(goodsId)   lockValue := gconv.String(gtime.TimestampNano())      // 使用SET命令的NX选项实现分布式锁，同时设置过期时间   success, err := m.redisClient.Set(ctx, lockKey, lockValue, m.lockTimeout).Result()   if err != nil {      return "", gerror.Wrapf(err, "获取分布式锁失败，商品ID:%d", goodsId)   }      if success != "OK" {      return "", gerror.Newf("获取分布式锁失败，锁已被占用，商品ID:%d", goodsId)   }      return lockValue, nil}
// releaseLock 释放分布式锁func (m *DistributedLockStockManager) releaseLock(ctx context.Context, goodsId uint32, lockValue string) error {   lockKey := m.getLockKey(goodsId)      // 使用Lua脚本确保原子性释放锁，避免误删其他客户端的锁   releaseLuaScript := `      if redis.call("get", KEYS[1]) == ARGV[1] then         return redis.call("del", KEYS[1])      else         return 0      end   `      _, err := m.redisClient.Eval(ctx, releaseLuaScript, []string{lockKey}, lockValue)   if err != nil {      return gerror.Wrapf(err, "释放分布式锁失败，商品ID:%d", goodsId)   }      return nil}
// ReduceStock 扣减库存（使用分布式锁）func (m *DistributedLockStockManager) ReduceStock(ctx context.Context, goodsId uint32, count int) (bool, error) {   // 参数校验   if count < 1 {      return false, gerror.New("扣减数量必须大于0")   }
   // 获取分布式锁   lockValue, err := m.acquireLock(ctx, goodsId)   if err != nil {      return false, err   }
   // 确保释放锁   defer func() {      err := m.releaseLock(ctx, goodsId, lockValue)      if err != nil {         g.Log().Errorf(ctx, "释放分布式锁失败: %v", err)      }   }()
   // 获取当前库存并扣减   stockKey := m.getStockKey(goodsId)   currentStockStr, err := m.redisClient.Get(ctx, stockKey)   if err != nil {      return false, gerror.Wrapf(err, "获取当前库存失败，商品ID:%d", goodsId)   }
   // 解析库存并检查是否充足   currentStock := 0   if currentStockStr.String() != "" {      currentStock = gconv.Int(currentStockStr.String())   }
   if currentStock < count {      return false, gerror.Newf("库存不足，商品ID:%d，当前库存:%d，请求数量:%d", goodsId, currentStock, count)   }
   // 扣减库存   newStock := currentStock - count   _, err = m.redisClient.Set(ctx, stockKey, newStock, 0) // 0表示永不过期   if err != nil {      return false, gerror.Wrapf(err, "更新库存失败，商品ID:%d", goodsId)   }
   return true, nil}
复制代码

7.3 Lua 脚本实现详解

Lua 脚本实现的核心是RedisLuaStockManager结构体，它包含以下关键部分：


  1. Lua 脚本定义：定义用于库存扣减、返还和初始化的 Lua 脚本

  2. 脚本执行：使用 Redis 的 Eval 命令执行 Lua 脚本

  3. 结果处理：解析脚本执行结果并返回


核心实现代码如下：


// 定义Lua脚本const (   // reduceStockLuaScript 库存扣减Lua脚本   reduceStockLuaScript = `      -- 获取当前库存      local currentStock = redis.call('get', KEYS[1])      if currentStock == false then         currentStock = 0      else         currentStock = tonumber(currentStock)      end            -- 检查库存是否足够      if currentStock >= tonumber(ARGV[1]) then         -- 扣减库存         redis.call('set', KEYS[1], currentStock - tonumber(ARGV[1]))         return 1  -- 扣减成功      else         return 0  -- 库存不足      end   `      // returnStockLuaScript 库存返还Lua脚本   returnStockLuaScript = `      -- 获取当前库存      local currentStock = redis.call('get', KEYS[1])      if currentStock == false then         currentStock = 0      else         currentStock = tonumber(currentStock)      end            -- 返还库存      redis.call('set', KEYS[1], currentStock + tonumber(ARGV[1]))      return 1  -- 返还成功   `      // initStockLuaScript 库存初始化Lua脚本   initStockLuaScript = `      -- 设置初始库存      redis.call('set', KEYS[1], tonumber(ARGV[1]))      return 1  -- 初始化成功   `)
// ReduceStock 扣减库存（使用Lua脚本）func (m *RedisLuaStockManager) ReduceStock(ctx context.Context, goodsId uint32, count int) (bool, error) {   // 参数校验   if count < 1 {      return false, gerror.New("扣减数量必须大于0")   }
   // 获取库存键   stockKey := m.getStockKey(goodsId)
   // 执行Lua脚本   result, err := m.redisClient.Eval(ctx, reduceStockLuaScript, []string{stockKey}, count)   if err != nil {      return false, gerror.Wrapf(err, "执行库存扣减Lua脚本失败，商品ID:%d，请求数量:%d", goodsId, count)   }
   // 解析结果   resultInt, ok := result.(int64)   if !ok {      return false, gerror.Newf("无法解析Lua脚本结果，商品ID:%d，请求数量:%d", goodsId, count)   }
   if resultInt == 0 {      // 获取当前库存，用于错误消息      currentStock, _ := m.GetStock(ctx, goodsId)      return false, gerror.Newf("库存不足，商品ID:%d，当前库存:%d，请求数量:%d", goodsId, currentStock, count)   }
   return resultInt == 1, nil}
复制代码

7.4 测试验证实现

我们实现了全面的测试用例，模拟高并发场景下两种方案的表现：


  1. 并发性能测试：模拟多线程同时扣减库存，验证性能和结果正确性

  2. 边界情况测试：测试库存不足、负数库存等边界情况

  3. 异常恢复测试：测试在异常情况下系统的恢复能力


核心测试代码如下：


// TestStockManagerComparison 测试两种库存管理方案的对比func TestStockManagerComparison(t *testing.T) {   // 测试分布式锁方案   t.Run("分布式锁方案", func(t *testing.T) {      // 初始化测试环境      ctx := context.Background()      goodsId := uint32(1)      initialStock := 100      concurrentRequests := 200      requestCount := 1
      // 初始化库存      _, err := distributedLockManager.InitStock(ctx, goodsId, initialStock)      require.NoError(t, err)
      // 创建结果通道和等待组      successChan := make(chan bool, concurrentRequests)      errChan := make(chan error, concurrentRequests)      var wg sync.WaitGroup
      // 记录开始时间      startTime := time.Now()
      // 启动并发请求      for i := 0; i < concurrentRequests; i++ {         wg.Add(1)         go func() {            defer wg.Done()            success, err := distributedLockManager.ReduceStock(ctx, goodsId, requestCount)            successChan <- success            if err != nil {               errChan <- err            } else {               errChan <- nil            }         }()      }
      // 等待所有请求完成      wg.Wait()      close(successChan)      close(errChan)
      // 计算执行时间      executionTime := time.Since(startTime)
      // 统计结果      successCount := 0      errorCount := 0      for success := range successChan {         if success {            successCount++         }      }      for err := range errChan {         if err != nil {            errorCount++         }      }
      // 获取最终库存      finalStock, err := distributedLockManager.GetStock(ctx, goodsId)      require.NoError(t, err)
      // 验证结果      expectedSuccessCount := initialStock / requestCount      expectedFinalStock := initialStock - expectedSuccessCount*requestCount
      t.Logf("执行时间: %v", executionTime)      t.Logf("成功次数: %d", successCount)      t.Logf("失败次数: %d", errorCount)      t.Logf("最终库存: %d", finalStock)      t.Logf("期望成功次数: %d", expectedSuccessCount)      t.Logf("期望最终库存: %d", expectedFinalStock)
      // 验证库存是否正确      require.Equal(t, expectedFinalStock, finalStock)      require.Equal(t, expectedSuccessCount, successCount)   })
   // 清理测试数据   ctx := context.Background()   goodsId := uint32(1)   // 清理Redis中的测试数据   _, _ = redisClient.Del(ctx, fmt.Sprintf("stock:%d", goodsId))
   // 测试Redis Lua脚本方案   t.Run("Redis Lua脚本方案", func(t *testing.T) {      // 初始化测试环境      ctx := context.Background()      goodsId := uint32(1)      initialStock := 100      concurrentRequests := 200      requestCount := 1
      // 初始化库存      _, err := redisLuaManager.InitStock(ctx, goodsId, initialStock)      require.NoError(t, err)
      // 创建结果通道和等待组      successChan := make(chan bool, concurrentRequests)      errChan := make(chan error, concurrentRequests)      var wg sync.WaitGroup
      // 记录开始时间      startTime := time.Now()
      // 启动并发请求      for i := 0; i < concurrentRequests; i++ {         wg.Add(1)         go func() {            defer wg.Done()            success, err := redisLuaManager.ReduceStock(ctx, goodsId, requestCount)            successChan <- success            if err != nil {               errChan <- err            } else {               errChan <- nil            }         }()      }
      // 等待所有请求完成      wg.Wait()      close(successChan)      close(errChan)
      // 计算执行时间      executionTime := time.Since(startTime)
      // 统计结果      successCount := 0      errorCount := 0      for success := range successChan {         if success {            successCount++         }      }      for err := range errChan {         if err != nil {            errorCount++         }      }
      // 获取最终库存      finalStock, err := redisLuaManager.GetStock(ctx, goodsId)      require.NoError(t, err)
      // 验证结果      expectedSuccessCount := initialStock / requestCount      expectedFinalStock := initialStock - expectedSuccessCount*requestCount
      t.Logf("执行时间: %v", executionTime)      t.Logf("成功次数: %d", successCount)      t.Logf("失败次数: %d", errorCount)      t.Logf("最终库存: %d", finalStock)      t.Logf("期望成功次数: %d", expectedSuccessCount)      t.Logf("期望最终库存: %d", expectedFinalStock)
      // 验证库存是否正确      require.Equal(t, expectedFinalStock, finalStock)      require.Equal(t, expectedSuccessCount, successCount)   })}
复制代码

8. 总结

通过对 Redis 分布式锁和 Redis Lua 脚本两种方案的详细实现和对比，我们可以得出以下结论：


  1. 性能方面：Redis Lua 脚本方案在高并发场景下性能明显优于分布式锁方案，主要是因为它减少了网络交互次数，避免了锁竞争带来的性能损耗。

  2. 可靠性方面：两种方案都能有效防止库存超卖，但 Redis Lua 脚本方案由于其原子性执行的特性，在某些方面更加可靠，不存在死锁风险。

  3. 适用场景：


  • 对于高并发的简单库存扣减操作（如秒杀、抢购），优先选择 Redis Lua 脚本方案

  • 对于需要执行复杂业务逻辑的场景，可以考虑使用分布式锁方案

  • 最佳实践：在实际项目中，可以根据不同的业务场景和并发需求，灵活选择合适的方案，甚至可以考虑两种方案的混合使用。


如果你对这种技术问题有疑问，或者对这个微服务项目感兴趣，都可以直接私信我：wangzhongyang1993。

划线
评论
复制
发布于: 1 小时前阅读数: 9

版权声明: 本文为 InfoQ 作者【王中阳Go】的原创文章。

原文链接:【http://xie.infoq.cn/article/42451ead2ffc848031b3afced】。文章转载请联系作者。

Go后端电商
用户头像

王中阳Go

关注

靠敲代码在北京买房的程序员 2022-10-09 加入

【微信】wangzhongyang1993【公众号】程序员升职加薪之旅【成就】InfoQ专家博主👍掘金签约作者👍B站&掘金&CSDN&思否等全平台账号：王中阳Go

评论

发布
暂无评论
Copyright © 2025, Geekbang Technology Ltd. All rights reserved. 极客邦控股（北京）有限公司 | 京 ICP 备 16027448 号 - 5京公网安备京公网安备 11010502039052号 | 产品资质
库存防超卖（Redis Lua+分布式锁对比实践）_Go_王中阳Go_InfoQ写作社区