面試官問redis分佈式鎖，如何設計才能讓他滿意？

前言

對於分佈式鎖的問題我也查過不少資料，感受不少方式實現的並不完善，或者看着雲裏霧裏的，不知因此然，因而就整理了這篇文章，但願對您有用，有寫的不對的地方，歡迎留言指正。

首先我們來聊聊什麼是分佈式鎖，到底解決了什麼問題？直接看代碼

$stock = $this->getStockFromDb();//查詢剩餘庫存
 if ($stock>0){
       $this->ReduceStockInDb(); // 在數據庫中進行減庫存操做
       echo "successful";
 }else{
    echo  "庫存不足";
 }

很簡單的一個場景，用戶下單，我們查詢商品庫存夠不夠，不夠的話直接返回庫存不足相似的錯誤信息，若是庫存夠的話直接在數據庫中庫存-1，而後返回成功，在業務邏輯上這段代碼是沒有什麼問題的。

可是，這段代碼是存在嚴重的問題的。

若是庫存只剩 1，而且在併發比較高的狀況下，好比兩個請求同時執行了這段代碼，同時查到庫存爲 1，而後順利成章的都去數據庫執行 stock-1 的操做，這樣庫存就會變成-1，而後就會引起超賣的現象，剛纔說的是兩個請求同時執行，若是同時幾千個請求打過來，可見形成的損失是很是大的。因而呢有些聰明人就想了個辦法，辦法以下。

你們都知道 redis 有個 setnx 命令，不知道的話也不要緊，我已經幫你查過了

咱們把上面的代碼優化一下

version-1

$lock_key="lock_key";
 $res = $redis->setNx($lock_key, 1);
 if (!$res){
       return "error_code";
 }

 $stock = $this->getStockFromDb();//查詢剩餘庫存
 if ($stock>0){
       $this->ReduceStockInDb(); // 在數據庫中進行減庫存操做
       echo "successful";
 }else{
    echo  "庫存不足";
 }

$redis->delete($lock_key);

• 第一次請求進來會去 setNx，固然結果是返回 true，由於 lock_key 不存在，而後下面業務邏輯正常進行，任務執行完了以後把lock_key刪除掉，這樣下一次請求進來重複上述邏輯
• 第二次請求進來一樣會去執行 setNx，結果返回 false，由於lock_key已經存在，而後直接返回錯誤信息(你雙11搶購秒殺產品的時候給你返回的系統繁忙就是這麼來的)，不執行庫存減 1 的操做

• 有的同窗可能有疑惑，我們不是說高併發的狀況下麼？要是兩個請求同時 setNx 的話獲取的結果不都是 true 了，一樣會同時去執行業務邏輯，問題不是同樣沒解決麼？可是你們要明白 redis 是單線程的，具有原子性，不一樣的請求執行 setnx 是順序執行的，因此這個是不用擔憂的。

看似問題解決了，其實並否則。

咱們這裏僞代碼寫的簡單，查詢一下庫存，而後減1操做而已，可是真實的生產環境中的狀況是很是複雜的，在一些極端狀況下，程序極可能會報錯，崩潰，若是第一次執行加鎖了以後程序報錯了，那這個鎖永遠存在，接下來的請求永遠也請求不進來了，因此我們繼續優化

version-2

try{ //新加入try catch處理，這樣程序萬一報錯會把鎖刪除掉
   $lock_key="lock_key";
   $expire_time = 5;//新加入過時時間，這樣鎖不會一直佔有
   $res = $redis->setNx($lock_key, 1, $expire_time);
   if (!$res){
         return "error_code";
   }

   $stock = $this->getStockFromDb();//查詢剩餘庫存
   if ($stock>0){
         $this->ReduceStockInDb(); // 在數據庫中進行減庫存操做
         echo "successful";
   }else{
      echo  "庫存不足";
   }
 }finally {
    $redis->delete($lock_key);
}

• 在setnx的時候給加上過時時間，這樣至少不會讓鎖一直存在成爲死鎖
• 作try catch處理，萬一程序拋出異常把鎖刪掉，也是爲了解決死鎖問題

此次是把死鎖問題解決了，可是問題仍是存在，你們能夠先想想還存在什麼問題再接着往下看。

存在的問題以下

• 咱們的過時時間是5秒鐘，萬一這個請求執行了6秒鐘怎麼辦？超出的那一秒，跟沒有加鎖有什麼區別？其實不只僅如此，還有一個更嚴重的問題存在。好比第二個請求也是執行6秒，那麼在第二個請求在超出的那1秒才進來的時候，第一個請求執行完了，固然會刪除第二個請求加的鎖，若是一直併發都很大的話，鎖跟沒有加沒什麼區別。
• 針對上述問題，最直接的辦法是加長過時時間，可是這個不是解決問題的最終辦法。把時間設置過長也會產生新的問題，好比各類緣由機器崩潰了，須要重啓，而後你把鎖設置的時間是1年，同時也沒有delete掉，難道機器重啓了再等一年？另外這樣設置固定值的解決方案在計算機當中是不容許的，曾經的「千年蟲」問題就是相似的緣由致使的
• 在加超時時間的時候必定要注意必定是一次性加上，保證其原子性，不要先setnx以後，再設置expire_time，這樣的話萬一在setnx以後那一個瞬間系統掛了，這個鎖依然會成爲一個永久的死鎖
• 其實上述問題的主要緣由在於，請求1會刪掉請求2的鎖，因此說鎖須要保證惟一性。

我們接着優化

version-3

try{ //新加入try catch處理，這樣程序萬一報錯會把鎖刪除掉
   $lock_key="lock_key";
   $expire_time = 5;//新加入過時時間，這樣鎖不會一直佔有
   
   $client_id = session_create_id();  //對每一個請求生成惟一性的id
   $res = $redis->setNx($lock_key, $client_id, $expire_time);
   if (!$res){
         return "error_code";
   }

   $stock = $this->getStockFromDb();//查詢剩餘庫存
   if ($stock>0){
         $this->ReduceStockInDb(); // 在數據庫中進行減庫存操做
         echo "successful";
   }else{
      echo  "庫存不足";
   }
 }finally {
   if ($redis->get($lock_key) == $client_id){  //在這裏加一個判斷，保證每次刪除的鎖是當次請求加的鎖，這樣避免誤刪了別的請求加的鎖
      $redis->delete($lock_key);
   }
   
}

• 咱們在每一個請求生成了惟一client_id，而且把該值寫入了lock_key中
• 在最後刪除鎖的時候會先判斷這個lock_key是不是該請求生成的，若是不是的話則不會刪除

可是上面方案還有問題，咱們看最後 redis是先進行了get操做判斷，而後再刪除，是兩步操做，並無保證其原子性，redis的多步操做能夠用lua腳原本保證原子性，其實看到lua也不須要感受太陌生，他就是一種語言而已，在這裏的做用是把多個redis操做打包成一個命令去執行，保證了原子性而已

version-4

try{ //新加入try catch處理，這樣程序萬一報錯會把鎖刪除掉
   $lock_key="lock_key";
   $expire_time = 5;//新加入過時時間，這樣鎖不會一直佔有
   
   $client_id = session_create_id();  //對每一個請求生成惟一性的id
   $res = $redis->setNx($lock_key, $client_id, $expire_time);
   if (!$res){
         return "error_code";
   }

   $stock = $this->getStockFromDb();//查詢剩餘庫存
   if ($stock>0){
         $this->ReduceStockInDb(); // 在數據庫中進行減庫存操做
         echo "successful";
   }else{
      echo  "庫存不足";
   }
 }finally {
    $script = '  //此處用lua腳本執行是爲了get對比以後再delete的兩步操做的原子性
            if redis.call("GET", KEYS[1]) == ARGV[1] then
                return redis.call("DEL", KEYS[1])
            else
                return 0
            end
        ';
        return $instance->eval($script, [$lock_key, $client_id], 1);
   
}

這樣封裝以後，分佈式鎖應該就比較完善了。固然咱們還能夠進一步的優化一下用戶體驗

• 如今好比一個請求進來以後，若是請求被鎖住，會當即返回給用戶請求失敗，請從新嘗試，咱們能夠適當的延長一點這個時間，不要當即返回給用戶請求失敗，這樣體驗會更好
• 具體方式爲用戶請求進來若是遇到了鎖，能夠適當的等待一些時間以後重試，重試的時候若是鎖釋放了，則此次請求就能夠成功

version-5

$retry_times = 3; //重試次數
$usleep_times = 5000;//重試間隔時間

  try{ //新加入try catch處理，這樣程序萬一報錯會把鎖刪除掉
   
   
    $lock_key="lock_key";
    $expire_time = 5;//新加入過時時間，這樣鎖不會一直佔有
    while($retry_times > 0){
      $client_id = session_create_id();  //對每一個請求生成惟一性的id
      $res = $redis->setNx($lock_key, $client_id, $expire_time);
      if ($res){
           break;
      }
      echo "嘗試從新獲取鎖";
      $retry_times--;
      usleep($usleep_times);
   }
   if (!$res){  //重試三次以後都沒有獲取到鎖則給用戶返回錯誤信息
         return "error_code";
   }
   $stock = $this->getStockFromDb();//查詢剩餘庫存
   if ($stock>0){
         $this->ReduceStockInDb(); // 在數據庫中進行減庫存操做
         echo "successful";
   }else{
      echo  "庫存不足";
   }
 }finally {
    $script = '  //此處用lua腳本執行是爲了get對比以後再delete的兩步操做的原子性
            if redis.call("GET", KEYS[1]) == ARGV[1] then
                return redis.call("DEL", KEYS[1])
            else
                return 0
            end
        ';
        return $instance->eval($script, [$lock_key, $client_id], 1);
   
}

固然上面的分佈式鎖仍是不夠完善的，好比redis主從同步延遲，就會產生問題，像java中redission實現的思想是很是好的，你們感興趣能夠看看源碼，今天就聊到這裏，感興趣的朋友能夠留言你們一塊兒討論