事故現場

     

      ????public?SeckillActivityRequestVO?seckillHandle(SeckillActivityRequestVO?request)?{
????SeckillActivityRequestVO?response;
????????String?key?=?"key:"?+?request.getSeckillId;
????????try?{
????????????Boolean?lockFlag?=?redisTemplate.opsForValue().setIfAbsent(key,?"val",?10,?TimeUnit.SECONDS);
????????????if?(lockFlag)?{
????????????????//?HTTP請求用戶服務進行用戶相關的校驗
????????????????//?用戶活動校驗

????????????????//?庫存校驗
????????????????Object?stock?=?redisTemplate.opsForHash().get(key+":info",?"stock");
????????????????assert?stock?!=?null;
????????????????if?(Integer.parseInt(stock.toString())?<=?0)?{
????????????????????//?業(yè)務異常
????????????????}?else?{
????????????????????redisTemplate.opsForHash().increment(key+":info",?"stock",?-1);
????????????????????//?生成訂單
????????????????????//?發(fā)布訂單創(chuàng)建成功事件
????????????????????//?構建響應VO
????????????????}
????????????}
????????}?finally?{
????????????//?釋放鎖
????????????stringRedisTemplate.delete("key");
????????????//?構建響應VO
????????}
????????return?response;
????}

     

事故原因

事故分析

• 沒有其他系統(tǒng)風險容錯處理 ?
  由于用戶服務吃緊，網關響應延遲，但沒有任何應對方式，這是超賣的 導火索 。
• 看似安全的分布式鎖其實一點都不安全 ?
  雖然采用了 set key value [EX seconds] [PX milliseconds][NX|XX] 的方式，但是如果線程A執(zhí)行的時間較長沒有來得及釋放，鎖就過期了，此時線程B是可以獲取到鎖的。當線程A執(zhí)行完成之后，釋放鎖，實際上就把線程B的鎖釋放掉了。這個時候，線程C又是可以獲取到鎖的，而此時如果線程B執(zhí)行完釋放鎖實際上就是釋放的線程C設置的鎖。這是超賣的

• 非原子性的庫存校驗 ?
  非原子性的庫存校驗導致在并發(fā)場景下，庫存校驗的結果不準確。這是超賣的 根本原因 。通過以上分析，問題的根本原因在于庫存校驗嚴重依賴了分布式鎖。因為在分布式鎖正常set、del的情況下，庫存校驗是沒有問題的。但是，當分布式鎖不安全可靠的時候，庫存校驗就沒有用了。

解決方案

實現相對安全的分布式鎖

     

      ????public?void?safedUnLock(String?key,?String?val)?{
????????String?luaScript?=?"local?in?=?ARGV[1]?local?curr=redis.call('get',?KEYS[1])?if?in==curr?then?redis.call('del',?KEYS[1])?end?return?'OK'"";
????????RedisScript?redisScript?=?RedisScript.of(luaScript);
????????redisTemplate.execute(redisScript,?Collections.singletonList(key),?Collections.singleton(val));
????}

     

實現安全的庫存校驗

     

      ????//?redis會返回操作之后的結果，這個過程是原子性的
????Long?currStock?=?redisTemplate.opsForHash().increment("key",?"stock",?-1);

     

改進之后的代碼

     

      ????public?SeckillActivityRequestVO?seckillHandle(SeckillActivityRequestVO?request)?{
????SeckillActivityRequestVO?response;
????????String?key?=?"key:"?+?request.getSeckillId();
????????String?val?=?UUID.randomUUID().toString();
????????try?{
????????????Boolean?lockFlag?=?distributedLocker.lock(key,?val,?10,?TimeUnit.SECONDS);
????????????if?(!lockFlag)?{
????????????????//?業(yè)務異常
????????????}

????????????//?用戶活動校驗
????????????//?庫存校驗，基于redis本身的原子性來保證
????????????Long?currStock?=?stringRedisTemplate.opsForHash().increment(key?+?":info",?"stock",?-1);
????????????if?(currStock?0)?{?//?說明庫存已經扣減完了。
????????????????//?業(yè)務異常。
????????????????log.error("[搶購下單]?無庫存");
????????????}?else?{
????????????????//?生成訂單
????????????????//?發(fā)布訂單創(chuàng)建成功事件
????????????????//?構建響應
????????????}
????????}?finally?{
????????????distributedLocker.safedUnLock(key,?val);
????????????//?構建響應
????????}
????????return?response;
????}
????復制代碼

     

深度思考

分布式鎖有必要么

分布式鎖的選型

再次思考分布式鎖有必要么

     

      ????//?通過消息提前初始化好，借助ConcurrentHashMap實現高效線程安全
????private?static?ConcurrentHashMap?SECKILL_FLAG_MAP?=?new?ConcurrentHashMap<>();
????//?通過消息提前設置好。由于AtomicInteger本身具備原子性，因此這里可以直接使用HashMap
????private?static?Map?SECKILL_STOCK_MAP?=?new?HashMap<>();

????...

????public?SeckillActivityRequestVO?seckillHandle(SeckillActivityRequestVO?request)?{
????SeckillActivityRequestVO?response;

????????Long?seckillId?=?request.getSeckillId();
????????if(!SECKILL_FLAG_MAP.get(requestseckillId))?{
????????????//?業(yè)務異常
????????}
?????????//?用戶活動校驗
?????????//?庫存校驗
????????if(SECKILL_STOCK_MAP.get(seckillId).decrementAndGet()?0)?{
????????????SECKILL_FLAG_MAP.put(seckillId,?false);
????????????//?業(yè)務異常
????????}
????????//?生成訂單
????????//?發(fā)布訂單創(chuàng)建成功事件
????????//?構建響應
????????return?response;
????}

     

總結

特別推薦一個分享架構+算法的優(yōu)質內容，還沒關注的小伙伴，可以長按關注一下：

長按訂閱更多精彩▼
如有收獲，點個在看，誠摯感謝