目錄
- 環境
- 實現
- 基本結構代碼
- 業務代碼主體
- 庫存管理模塊
- 后續問題
- 高并發
- 臨界值與樂觀鎖問題
- 完整代碼總結
- 后話
環境
我們現在要做商品秒殺系統。功能很簡單，就是庫存刪減。用戶先下單減庫存，之后再進行扣款。
實現
基本結構代碼
那么我們先看下如何搭建好基本的代碼。
業務代碼主體
基本步驟就以下幾點
- 刪減庫存
- 填寫訂單基本信息
public class SecKillBusinessService {// 庫存 serviceprivate StockDataService stockService;// 訂單 serviceprivate OrderService orderService;public Response order( String userId , String productId ){// 獲取當前時間點LocalDateTime time = LocalDateUtils.now();// 1. 刪減庫存this.stockService.reduceStock( userId , productId , 1 );//2. 下單OrderEntity order = new OrderEntity();order.setId(xxxx);order.setUserId(userId);order.setTime( time );this.orderService.add( order );return Response.success();}
}庫存管理模塊
public class StockService {// 庫存底層數據private StockDataService dataService;public void reduceStock(String userId , String productId , Integer number ){// 獲取剩余庫存數量int surplusStock = this.dataService.getStock( productId );if( surplusStock == 0 || surplusStock-number < 0 ){throw new ResponseFailedExpection( "庫存不足");}// 自減數量 , 當庫存不足時扣減失敗，當前失敗碼暫定為-1int surplusNumber = this.dataService.decrementStock( productId , number );if( surplusNumber < 0 ){throw new ResponseFailedExpection("庫存不足");}}}
StockDataService 我們先通過查詢Mysql來實現。
public class StockDataServiceRedisImpl implement StockmentDataService {public int getStock( String productId ){// SELECT * FROM t_a_product WHERE product_id = #{productId}}@Transactionpublic int decrmentStock( String productId , Integer number ){// 簡單的樂觀鎖// UPDATE t_a_product SET stock-=#{number} WHERE product_id = #{productId} AND stock>=#{number} }
}后續問題
秒殺的主要問題復雜代碼集中在如何在高并發環境下扣減庫存，庫存不會出現庫存數據計數錯誤，且更高效。
高并發
當數據量上來的時候，我們很快就會發現問題。當流量大的時候，數據庫IO很快就會打滿。然后查詢慢，插入慢。最后Mysql掛掉，服務不可用。
主要的問題，就是數據庫難以應付高并發。那么我們如何處理？
很簡單，我們使用Redis來替代Mysql ， 我們新建一個新的StockDataService來進行替換。
為了保證計數問題，我們無非要么用樂觀鎖要么用悲觀鎖要么二者都用。 高并發情況下，我們不可能用悲觀鎖來讓程序在同一時間只允許一個請求在運行。（因為會引發大規模排隊）因此我們采用樂觀鎖
public class StocklDataServiceRedisImpl implement StockmentDataService {private RedisService redisService;private static final String GET_STOCK_KEY = "GET_STOCK";private String getStockRedisKey( String productId ){return GET_STOCK_KEY + productId;}/**redis之中的庫存數在其他模塊便填充，我們可以放在后臺配置的時候，也可以通過定時任務在商品生效一個小時之前。*/public int getStock( String productId ){return redisService.get(this.getStockRedisKey(productId) , Integer.class);}public int decrmentStock( String productId , Integer number ){String redisKey = this.getStockRedisKey(productId);int surplusNumber = this.redisSerivce.decrement(redisKey ,number);// 如果減少的數量超過庫存上限，那么歸還庫存if( surplusNumber <0 ){this.redisService.incrment(redisKey ,number);return -1;}return surplusNumber;}
我們簡單的用redis做了一個減庫存的相關功能， 并且還簡單做了一個樂觀鎖邏輯。 來處理臨界值時庫存扣減超量問題。
臨界值與樂觀鎖問題
在討論當前情況之前， 我們得先對臨界值有一個簡單的認識。 就是一個商品的臨界值時多少？
由于本人水平有限，我先簡單的做個定義。 0.8 * 當前剩余庫存數 = 當前所需的數量
簡單的說，假設當前庫存10000份，當前庫存數已經只剩下了500，當前服務器內計算到的所需要的總數達到400甚至更多時，我們就需要，那么我們就到達了臨界值狀態。
那么現在我們回到問題，
雖然我們樂觀鎖能簡單解決大部分問題，但是當庫存來到臨界值的時候，我們就會悲傷的發現。 大量的請求會失效。這些請求即無用又會給redis造成極大的壓力。
問題的本質是什么呢？是因為查詢+查庫存的這兩步驟無法原子化，庫存數量在刪減庫存的時候并不可靠。
我們就直接說Redis的解決方案。
Redis Lua 腳本
不認識的可以簡單的這樣認為，他會把不同的腳本原子化處理。也可以說Redis會自己將一連串的Lua用分布式鎖鎖住然后執行。只是用它來實現分布式事務鎖不太容易出現性能問題。
-- 方式 2：Lua 腳本實現原子扣減
local stockKey = KEYS[1]
local number = KEYS[2]
local stock = tonumber(redis.call('GET', stockKey))if stock >= number thenredis.call('DECR', stockKey)return stock - number
elsereturn -1
end我們可以直接更改 StocklDataServiceRedisImpl
public class StocklDataServiceRedisImpl implement StockmentDataService {private RedisService redisService;private static final String GET_STOCK_KEY = "GET_STOCK";private String getStockRedisKey( String productId ){return GET_STOCK_KEY + productId;}public int decrmentStock( String productId , Integer number ){// Lua 腳本String script = "xxx";// 通過 Lua 一次性扣減庫存DefaultRedisScript<Integer> redisScript = new DefaultRedisScript<>(script, Integer.class);List<String> keys = Arrays.asList(this.getStockRedisKey(productId , new StringBuilder.append(number).toString()));return this.redisService.executeLua(redisScript, keys);}并且由于Redis Lua 能保證原子性，甚至能更改 StockService 邏輯 不需要對當前庫存進行校驗。僅處理一個Redis命令即可。
自然可能由于其他因素，是否如此憑個人好惡
public class StockService {// 庫存底層數據private StockDataService dataService;public void reduceStock(String userId , String productId , Integer number ){// 自減數量 , 當庫存不足時扣減失敗，當前失敗碼暫定為-1int surplusNumber = this.dataService.decrementStock( productId , number );if( surplusNumber < 0 ){throw new ResponseFailedExpection("庫存不足");}}}
完整代碼總結
完善之后，當前代碼為
SecKillBusinessService .java
public class SecKillBusinessService {// 庫存 serviceprivate StockDataService stockService;// 訂單 serviceprivate OrderService orderService;public Response order( String userId , String productId ){// 獲取當前時間點LocalDateTime time = LocalDateUtils.now();// 1. 刪減庫存this.stockService.reduceStock( userId , productId , 1 );//2. 下單OrderEntity order = new OrderEntity();order.setId(xxxx);order.setUserId(userId);order.setTime( time );this.orderService.add( order );return Response.success();}
}StockService .java
public class StockService {// 庫存底層數據private StockDataService dataService;public void reduceStock(String userId , String productId , Integer number ){// 獲取剩余庫存數量int surplusStock = this.dataService.getStock( productId );if( surplusStock == 0 || surplusStock-number < 0 ){throw new ResponseFailedExpection( "庫存不足");}// 自減數量 , 當庫存不足時扣減失敗，當前失敗碼暫定為-1int surplusNumber = this.dataService.decrementStock( productId , number );if( surplusNumber < 0 ){throw new ResponseFailedExpection("庫存不足");}}}
StocklDataServiceRedisImpl .java
public class StocklDataServiceRedisImpl implement StockmentDataService {private RedisService redisService;private static final String GET_STOCK_KEY = "GET_STOCK";private String getStockRedisKey( String productId ){return GET_STOCK_KEY + productId;}/**redis之中的庫存數在其他模塊便填充，我們可以放在后臺配置的時候，也可以通過定時任務在商品生效一個小時之前。*/public int getStock( String productId ){return redisService.get(this.getStockRedisKey(productId) , Integer.class);}public int decrmentStock( String productId , Integer number ){// Lua 腳本String script = "xxx";// 通過 Lua 一次性扣減庫存DefaultRedisScript<Integer> redisScript = new DefaultRedisScript<>(script, Integer.class);List<String> keys = Arrays.asList(this.getStockRedisKey(productId , new StringBuilder.append(number).toString()));return this.redisService.executeLua(redisScript, keys);}后話
我們可以想象一下，如果沒有Redis Lua 功能， 我們需要做什么？
為了減少樂觀鎖出現的大面積下單失敗，我們只能依賴于悲觀鎖。
但是悲觀鎖嚴重影響性能不可取，因此我們只能折中。設置一個危險值，當庫存大于危險值時使用樂觀鎖，低于危險值時采用悲觀鎖。
危險值應該大于接口請求數上限，且為了不讓大量蜂擁而入的無用請求排隊。我們需要登記每個請求，且當請求量大于庫存數就直接拒絕服務。
這應該就是我們常說的，少即是多，以及磨刀不誤砍柴工吧。
：數據探索的“第一張圖表”)
 集群 安裝)
)
