這篇文章的內容都是基於我們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。