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

王中阳AI编程

发布于 2026-03-17 19:50:01

1630

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

1. 引言

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

2. 库存超卖问题分析

2.1 什么是库存超卖

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

2.2 超卖问题产生的原因

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

查询当前库存数量
判断库存是否足够
如果足够，则扣减库存

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

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

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

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

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

数据库层的锁粒度较粗，可能导致性能瓶颈
数据库连接数有限，在大量并发请求下可能成为系统瓶颈
网络延迟和多线程并发操作导致的竞态条件

3. 库存防超卖解决方案

3.1 解决方案概述

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

基于Redis分布式锁的库存扣减：使用Redis实现分布式锁，确保同一时间只有一个请求能够扣减特定商品的库存
基于Redis Lua脚本的库存扣减：利用Redis Lua脚本的原子性，将库存检查和扣减操作在Redis端原子执行

3.2 Redis分布式锁方案

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

3.2.1分布式锁的实现原理

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

3.2.2分布式锁的关键问题

锁的过期时间：防止锁持有方崩溃导致锁无法释放
锁的续期：对于长时间运行的操作，需要自动续期以避免锁过期
锁的释放：确保只有持有锁的客户端能够释放锁，避免误释放
锁的重试：在获取锁失败时，合理的重试策略可以提高成功率

3.3 Redis Lua脚本方案

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

3.3.1 Lua脚本的优势

原子性：脚本中的所有命令要么全部执行，要么全部不执行
减少网络开销：将多条命令合并为一个脚本发送到Redis服务器
减少客户端逻辑：将复杂的业务逻辑放在脚本中，简化客户端代码
消除竞态条件：在Redis服务器端执行，避免了多客户端并发操作的竞态条件

3.3.2 库存扣减的Lua脚本思路

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

获取当前商品的库存
判断库存是否足够
如果足够，扣减库存并返回成功
如果不足，返回失败

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 技术原理对比

对比项	Redis分布式锁	Redis Lua脚本
实现原理	使用SET NX命令实现锁机制，在库存操作前后加锁/解锁	利用Redis单线程执行特性，将库存检查和扣减封装在一个原子性Lua脚本中
原子性保证	操作由多个Redis命令组成，依赖分布式锁保证原子性	整个操作在Redis服务器端作为单个原子命令执行
网络开销	至少需要3次网络交互（加锁、操作、解锁）	仅需要1次网络交互（执行脚本）
复杂度	较高，需要处理锁的获取、释放、超时等逻辑	中等，主要是Lua脚本编写
代码量	较多，需要实现锁的管理逻辑	较少，主要是脚本定义和调用

5.2 性能对比

性能指标	Redis分布式锁	Redis Lua脚本
响应时间	较慢，受网络延迟影响大，需要多次交互	较快，网络交互少，一次请求完成所有操作
吞吐量	较低，高并发下锁竞争激烈，会出现线程等待	较高，避免了锁竞争，充分利用Redis性能
资源占用	Redis连接占用时间长，CPU利用率较高	Redis连接占用时间短，CPU利用率较低
扩展性	随并发增加，性能下降明显，呈非线性下降	随并发增加，性能相对稳定，接近线性扩展
高并发下的稳定性	较差，容易出现锁竞争和饥饿现象	较好，性能稳定，适合秒杀等高并发场景

5.3 可靠性对比

可靠性指标	Redis分布式锁	Redis Lua脚本
防超卖能力	强，但依赖锁的正确实现	强，由Redis单线程执行保证
死锁风险	存在，需要设置合理的超时时间	无，不使用锁机制，不存在死锁问题
异常恢复	依赖锁超时自动释放，可能有时间窗口	自动回滚，不影响其他操作，原子性更强
一致性保证	最终一致性，可能存在瞬时不一致	强一致性，操作原子性，保证数据一致性
故障隔离	单个操作失败可能影响其他操作获取锁	操作失败互不影响，故障隔离性好

5.4 适用场景对比

场景类型	Redis分布式锁	Redis Lua脚本
高并发秒杀	不推荐，性能瓶颈明显	强烈推荐，性能最优，原子性强
复杂业务逻辑	推荐，可以在锁内执行复杂操作	不推荐，Lua脚本不易处理复杂逻辑
多资源协调	推荐，可以协调多个资源的操作	不适用，难以处理跨多个键的复杂操作
库存扣减	适用，但性能不如Lua脚本	最佳选择，性能和可靠性兼顾
需要事务性操作	适合，可以在锁内执行多个步骤	适合简单事务，复杂事务难以实现
服务资源有限	不适合，锁竞争会加剧资源占用	更适合，资源利用效率更高

6.最佳实践建议

6.1 方案选择建议

基于业务复杂度选择

简单的库存扣减：优先使用Lua脚本方案
复杂业务逻辑（需要查询其他服务、执行多个步骤）：使用分布式锁方案
基于并发量选择
高并发场景（如秒杀、抢购）：强制使用Lua脚本
低并发场景：两种方案均可，可根据维护成本选择
基于团队技术栈选择
如果团队熟悉Lua：优先考虑Lua脚本方案
如果团队对分布式锁理解更深入：可以选择分布式锁方案
混合使用策略
对于核心的库存扣减操作：使用Lua脚本确保性能和原子性
对于需要协调多个资源的复杂业务：使用分布式锁

6.2 实现注意事项

分布式锁实现注意事项：

确保使用SET命令的NX选项来实现互斥性，同时设置过期时间
必须设置合理的锁超时时间，避免死锁（建议2-5秒）
使用Lua脚本释放锁，确保锁的原子性释放，避免误删
实现锁重试机制，使用指数退避算法提高锁获取成功率
考虑使用Redis集群提高可用性，避免单点故障
使用UUID或随机字符串作为锁的值，确保唯一性
实现锁自动续期机制（可选），对于长时间运行的操作

Lua脚本实现注意事项：

保持Lua脚本简洁，避免在脚本中执行复杂逻辑
脚本执行超时时间设置合理（建议2-5秒）
错误处理要完善，包括Redis连接错误和库存不足情况
考虑在脚本中加入合理的日志或监控信息
优化Lua脚本性能，避免在脚本中使用过多的条件判断和循环
预先加载Lua脚本到Redis，减少网络传输开销

6.3 性能优化建议

Redis连接优化

使用连接池管理Redis连接
合理设置连接池大小和超时时间
考虑使用Redis Sentinel或Redis Cluster提高可用性
缓存策略优化
库存预热：系统启动时将热点商品库存加载到Redis
本地缓存：对非热点数据使用本地缓存减少Redis访问
分级缓存：对不同热度的商品使用不同的缓存策略
并发控制优化
限流措施：在API网关层实施限流，保护库存服务
熔断降级：当Redis服务不可用时，降级到数据库操作
削峰填谷：使用消息队列处理高并发请求，如RabbitMQ
监控与告警
监控Redis内存使用情况和性能指标
监控库存操作的成功率、响应时间和错误率
对异常情况（如频繁的库存不足）设置告警阈值
实现操作审计日志，方便问题排查

6.4 运维建议

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结构体，它包含以下关键方法：

锁的获取：使用SET命令的NX选项，设置过期时间避免死锁
锁的释放：使用Lua脚本确保只有持有锁的客户端才能释放锁
库存操作：在获取锁后执行库存的扣减、返还等操作

核心实现代码如下：

// 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)
   }
   
   returnnil
}

// ReduceStock 扣减库存（使用分布式锁）
func (m *DistributedLockStockManager) ReduceStock(ctx context.Context, goodsId uint32, count int) (bool, error) {
   // 参数校验
   if count < 1 {
      returnfalse, gerror.New("扣减数量必须大于0")
   }

   // 获取分布式锁
   lockValue, err := m.acquireLock(ctx, goodsId)
   if err != nil {
      returnfalse, err
   }

   // 确保释放锁
   deferfunc() {
      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 {
      returnfalse, gerror.Wrapf(err, "获取当前库存失败，商品ID:%d", goodsId)
   }

   // 解析库存并检查是否充足
   currentStock := 0
   if currentStockStr.String() != "" {
      currentStock = gconv.Int(currentStockStr.String())
   }

   if currentStock < count {
      returnfalse, gerror.Newf("库存不足，商品ID:%d，当前库存:%d，请求数量:%d", goodsId, currentStock, count)
   }

   // 扣减库存
   newStock := currentStock - count
   _, err = m.redisClient.Set(ctx, stockKey, newStock, 0) // 0表示永不过期
   if err != nil {
      returnfalse, gerror.Wrapf(err, "更新库存失败，商品ID:%d", goodsId)
   }

   returntrue, nil
}

7.3 Lua脚本实现详解

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

Lua脚本定义：定义用于库存扣减、返还和初始化的Lua脚本
脚本执行：使用Redis的Eval命令执行Lua脚本
结果处理：解析脚本执行结果并返回

核心实现代码如下：

// 定义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 {
      returnfalse, gerror.New("扣减数量必须大于0")
   }

   // 获取库存键
   stockKey := m.getStockKey(goodsId)

   // 执行Lua脚本
   result, err := m.redisClient.Eval(ctx, reduceStockLuaScript, []string{stockKey}, count)
   if err != nil {
      returnfalse, gerror.Wrapf(err, "执行库存扣减Lua脚本失败，商品ID:%d，请求数量:%d", goodsId, count)
   }

   // 解析结果
   resultInt, ok := result.(int64)
   if !ok {
      returnfalse, gerror.Newf("无法解析Lua脚本结果，商品ID:%d，请求数量:%d", goodsId, count)
   }

   if resultInt == 0 {
      // 获取当前库存，用于错误消息
      currentStock, _ := m.GetStock(ctx, goodsId)
      returnfalse, gerror.Newf("库存不足，商品ID:%d，当前库存:%d，请求数量:%d", goodsId, currentStock, count)
   }

   return resultInt == 1, nil
}

7.4 测试验证实现

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

并发性能测试：模拟多线程同时扣减库存，验证性能和结果正确性
边界情况测试：测试库存不足、负数库存等边界情况
异常恢复测试：测试在异常情况下系统的恢复能力

核心测试代码如下：

// 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(chanbool, concurrentRequests)
      errChan := make(chan error, concurrentRequests)
      var wg sync.WaitGroup

      // 记录开始时间
      startTime := time.Now()

      // 启动并发请求
      for i := 0; i < concurrentRequests; i++ {
         wg.Add(1)
         gofunc() {
            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(chanbool, concurrentRequests)
      errChan := make(chan error, concurrentRequests)
      var wg sync.WaitGroup

      // 记录开始时间
      startTime := time.Now()

      // 启动并发请求
      for i := 0; i < concurrentRequests; i++ {
         wg.Add(1)
         gofunc() {
            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)
   })
}