Redis 分佈式鎖的 10 個坑

前言

日常開發中，經常會碰到秒殺搶購等業務。爲了避免併發請求造成的庫存超賣等問題，我們一般會用到Redis分佈式鎖。但是使用Redis分佈式鎖，很容易踩坑哦~ 本文田螺哥將給大家分析闡述，Redis分佈式鎖的10個坑~

  1. 非原子操作（setnx + expire）

一說到實現Redis的分佈式鎖，很多小夥伴馬上就會想到setnx+ expire命令。也就是說，先用setnx來搶鎖，如果搶到之後，再用expire給鎖設置一個過期時間。

僞代碼如下：

if（jedis.setnx(lock_key,lock_value) == 1）{ //加鎖
    jedis.expire(lock_key，timeout）; //設置過期時間
    doBusiness //業務邏輯處理
}

這塊代碼是有坑的，因爲setnx和expire兩個命令是分開寫的，並不是原子操作！如果剛要執行完setnx加鎖，正要執行expire設置過期時間時，進程crash或者要重啓維護了，那麼這個鎖就 “長生不老” 了，別的線程永遠獲取不到鎖啦。

  1. 被別的客戶端請求覆蓋（ setnx + value 爲過期時間）

爲了解決：發生異常時，鎖得不到釋放的問題。有小夥伴提出，可以把過期時間放到setnx的value裏面。如果加鎖失敗，再拿出value值和當前系統時間校驗一下是否過期即可。僞代碼實現如下：

long expireTime = System.currentTimeMillis() + timeout; //系統時間+設置的超時時間
String expireTimeStr = String.valueOf(expireTime); //轉化爲String字符串

// 如果當前鎖不存在，返回加鎖成功
if (jedis.setnx(lock_key, expireTimeStr) == 1) {
        return true;
} 

// 如果鎖已經存在，獲取鎖的過期時間
String oldExpireTimreStr = jedis.get(lock_key);

// 如果獲取到的老的預期過期時間，小於系統當前時間，表示已經過期了
if (oldExpireTimreStr != null && Long.parseLong(oldExpireTimreStr) < System.currentTimeMillis()) {

     //鎖已過期，獲取上一個鎖的過期時間，並設置現在鎖的過期時間（不瞭解redis的getSet命令的小夥伴，可以去官網看下哈）
    String oldValueStr = jedis.getSet(lock_key, expireTimeStr);
    
    if (oldValueStr != null && oldValueStr.equals(oldExpireTimreStr)) {
      //考慮多線程併發的情況，只有一個線程的設置值和當前值相同，它纔可以加鎖
      return true;
    }
}
        
//其他情況，均返回加鎖失敗
return false;
}

這種實現的方案，也是有坑的：如果鎖過期的時候，併發多個客戶端同時請求過來，都執行jedis.getSet()，最終只能有一個客戶端加鎖成功，但是該客戶端鎖的過期時間，可能被別的客戶端覆蓋。

  1. 忘記設置過期時間

之前review代碼的時候，看到這樣實現的分佈式鎖，僞代碼：

try{
  if(jedis.setnx(lock_key,lock_value) == 1){//加鎖
     doBusiness //業務邏輯處理
     return true; //加鎖成功，處理完業務邏輯返回
  }
  return false; //加鎖失敗
} finally {
    unlock(lockKey);- //釋放鎖
}

這塊有什麼問題呢？是的，忘記設置過期時間了。如果程序在運行期間，機器突然掛了，代碼層面沒有走到finally代碼塊，即在宕機前，鎖並沒有被刪除掉，這樣的話，就沒辦法保證解鎖，所以這裏需要給lockKey加一個過期時間。注意哈，使用分佈式鎖，一定要設置過期時間哈。

  1. 業務處理完，忘記釋放鎖

很多小夥伴，會使用Redis的set指令擴展參數來實現分佈式鎖。

set指令擴展參數：SET key value[EX seconds][PX milliseconds][NX|XX]

- NX :表示key不存在的時候，才能set成功，也即保證只有第一個客戶端請求才能獲得鎖，
  而其他客戶端請求只能等其釋放鎖，才能獲取。
- EX seconds :設定key的過期時間，時間單位是秒。
- PX milliseconds: 設定key的過期時間，單位爲毫秒
- XX: 僅當key存在時設置值

小夥伴會寫出如下僞代碼：

if(jedis.set(lockKey, requestId, "NX", "PX", expireTime)==1){ //加鎖
   doBusiness //業務邏輯處理
   return true; //加鎖成功，處理完業務邏輯返回
}
return false; //加鎖失敗

這塊僞代碼，初看覺得沒啥問題，但是細想，不太對呀。因爲忘記釋放鎖了！如果每次加鎖成功，都要等到超時時間才釋放鎖，是會有問題的。這樣程序不高效，應當每次處理完業務邏輯，都要釋放鎖。

正例如下：

try{
  if(jedis.set(lockKey, requestId, "NX", "PX", expireTime)==1){//加鎖
     doBusiness //業務邏輯處理
     return true; //加鎖成功，處理完業務邏輯返回
  }
  return false; //加鎖失敗
} finally {
    unlock(lockKey);- //釋放鎖
}
  1. B 的鎖被 A 給釋放了

我們來看下這塊僞代碼：

try{
  if(jedis.set(lockKey, requestId, "NX", "PX",expireTime)==1){//加鎖
     doBusiness //業務邏輯處理
     return true; //加鎖成功，處理完業務邏輯返回
  }
  return false; //加鎖失敗
} finally {
    unlock(lockKey); //釋放鎖
}

大家覺得會有哪些坑呢？

假設在這樣的併發場景下：A、B兩個線程來嘗試給 Redis 的 keylockKey加鎖，A線程先拿到鎖（假如鎖超時時間是3秒後過期）。如果線程A執行的業務邏輯很耗時，超過了3秒還是沒有執行完。這時候，Redis會自動釋放lockKey鎖。剛好這時，線程B過來了，它就能搶到鎖了，開始執行它的業務邏輯，恰好這時，線程A執行完邏輯，去釋放鎖的時候，它就把B的鎖給釋放掉了。

正確的方式應該是，在用set擴展參數加鎖時，放多一個這個線程請求的唯一標記，比如requestId，然後釋放鎖的時候，判斷一下是不是剛剛的請求。

try{
  if(jedis.set(lockKey, requestId, "NX", "PX",expireTime)==1){//加鎖
     doBusiness //業務邏輯處理
     return true; //加鎖成功，處理完業務邏輯返回
  }
  return false; //加鎖失敗
} finally {
    if (requestId.equals(jedis.get(lockKey))) { //判斷一下是不是自己的requestId
      unlock(lockKey);//釋放鎖
    }   
}
  1. 釋放鎖時，不是原子性

以上的這塊代碼，還是有坑：

   if (requestId.equals(jedis.get(lockKey))) { //判斷一下是不是自己的requestId
      unlock(lockKey);//釋放鎖
    }

因爲判斷是不是當前線程加的鎖和釋放鎖不是一個原子操作。如果調用unlock(lockKey)釋放鎖的時候，鎖已經過期，所以這把鎖已經可能已經不屬於當前客戶端，會解除他人加的鎖。

因此，這個坑就是：判斷和刪除是兩個操作，不是原子的，有一致性問題。釋放鎖必須保證原子性，可以使用Redis+Lua腳本來完成，類似Lua腳本如下：

if redis.call('get',KEYS[1]) == ARGV[1] then 
   return redis.call('del',KEYS[1]) 
else
   return 0
end;
  1. 鎖過期釋放，業務沒執行完

加鎖後，如果超時了，Redis會自動釋放清除鎖，這樣有可能業務還沒處理完，鎖就提前釋放了。怎麼辦呢？

有些小夥伴認爲，稍微把鎖過期時間設置長一些就可以啦。其實我們設想一下，是否可以給獲得鎖的線程，開啓一個定時守護線程，每隔一段時間檢查鎖是否還存在，存在則對鎖的過期時間延長，防止鎖過期提前釋放。

當前開源框架 Redisson 解決了這個問題。我們一起來看下Redisson底層原理圖吧：

只要線程一加鎖成功，就會啓動一個watch dog看門狗，它是一個後臺線程，會每隔10秒檢查一下，如果線程一還持有鎖，那麼就會不斷的延長鎖key的生存時間。因此，Redisson就是使用Redisson解決了鎖過期釋放，業務沒執行完問題。

  1. Redis 分佈式鎖和 @transactional 一起使用失效

大家看下這塊僞代碼:

@Transactional
public void updateDB(int lockKey) {
  boolean lockFlag = redisLock.lock(lockKey);
  if (!lockFlag) {
    throw new RuntimeException(“請稍後再試”);
  }
   doBusiness //業務邏輯處理
   redisLock.unlock(lockKey);
}

在事務中, 使用了Redis分佈式鎖. 這個方法一旦執行, 事務生效，接着就Redis分佈式鎖生效，代碼執行完後, 先釋放Redis分佈式鎖, 然後再提交事務數據，最後事務結束。在這個過程中, 事務沒有提交之前, 分佈式鎖已經被釋放，導致分佈式鎖失效

這是因爲:

spring的Aop，會在updateDB方法之前開啓事務，之後再加鎖，當鎖住的代碼執行完成後，再提交事務，因此鎖住的代碼塊執行是在事務之內執行的，可以推斷在代碼塊執行完時，事務還未提交，鎖已經被釋放，此時其他線程拿到鎖之後進行鎖住的代碼塊，讀取的庫存數據不是最新的。

正確的實現方法, 可以在updateDB方法之前就上鎖，即還沒有開事務之前就加鎖，那麼就可以保證線程的安全性.

  1. 鎖可重入

前面討論的Redis分佈式鎖，都是不可重入的。

所謂的不可重入，就是當前線程執行某個方法已經獲取了該鎖，那麼在方法中嘗試再次獲取鎖時，會阻塞，不可以再次獲得鎖。同一個人拿一個鎖 ，只能拿一次不能同時拿2次。

不可重入的分佈式鎖的話，是可以滿足絕大多數的業務場景。但是有時候一些業務場景，我們還是需要可重入的分佈式鎖，大家實現分佈式鎖的過程中，需要注意一下，你當前的業務場景是否需要可重入的分佈式鎖。

Redis只要解決這兩個問題，就能實現重入鎖了：

實現一個可重入的分佈式鎖，我們可以參考JDK的ReentrantLock的設計思想。實際上，可以直接使用Redisson框架，它是支持可重入鎖的。

10.Redis 主從複製導致的坑

實現Redis分佈式鎖的話，要注意Redis主從複製的坑。因爲Redis一般都是集羣部署的：

如果線程一在Redis的master節點上拿到了鎖，但是加鎖的key還沒同步到slave節點。恰好這時，master節點發生故障，一個slave節點就會升級爲master節點。線程二就可以獲取同個key的鎖啦，但線程一也已經拿到鎖了，鎖的安全性就沒了。

爲了解決這個問題，Redis 作者 antirez 提出一種高級的分佈式鎖算法：Redlock。Redlock核心思想是這樣的：

搞多個 Redis master 部署，以保證它們不會同時宕掉。並且這些 master 節點是完全相互獨立的，相互之間不存在數據同步。同時，需要確保在這多個 master 實例上，是與在 Redis 單實例，使用相同方法來獲取和釋放鎖。

我們假設當前有5個Redis master節點，在5臺服務器上面運行這些Redis實例。

RedLock 的實現步驟如下:

  1. 獲取當前時間，以毫秒爲單位。

  2. 按順序向5個master節點請求加鎖。客戶端設置網絡連接和響應超時時間，並且超時時間要小於鎖的失效時間。（假設鎖自動失效時間爲10秒，則超時時間一般在5-50毫秒之間, 我們就假設超時時間是50ms吧）。如果超時，跳過該master節點，儘快去嘗試下一個master節點。

  3. 客戶端使用當前時間減去開始獲取鎖時間（即步驟1記錄的時間），得到獲取鎖使用的時間。當且僅當超過一半（N/2+1，這裏是5/2+1=3個節點）的Redis master節點都獲得鎖，並且使用的時間小於鎖失效時間時，鎖纔算獲取成功。（如上圖，10s> 30ms+40ms+50ms+4m0s+50ms）

  4. 如果取到了鎖，key的真正有效時間就變啦，需要減去獲取鎖所使用的時間。

  5. 如果獲取鎖失敗（沒有在至少N/2+1個master實例取到鎖，有或者獲取鎖時間已經超過了有效時間），客戶端要在所有的master節點上解鎖（即便有些master節點根本就沒有加鎖成功，也需要解鎖，以防止有些漏網之魚）。

簡化下步驟就是：

參考與感謝

作者：撿田螺的小男孩

來源：撿田螺的小男孩

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/89c_BYs5bJ0bm-L8H2bJxg

猜你喜歡