在分布式系統中，保證資源的互斥訪問是一個關鍵的點，而 Redis 作為高性能的鍵值存儲系統，在分布式鎖這塊也被廣泛的應用。然而，在使用 Redis 實現分布式鎖時需要考慮很多的因素，以確保系統正確的使用還有程序的性能。

下面我們將探討一下使用Redis實現分布式鎖時需要注意的關鍵點。

首先還是大家都知道，使用 Redis 實現分布式鎖，是兩步操作，設置一個key，增加一個過期時間，所以我們首先需要保證的就是這兩個操作是一個原子操作。

1.原子性

在獲取鎖和釋放鎖的過程中，要保證這個操作的原子性，確保加鎖操作與設置過期時間操作是原子的。Redis 提供了原子操作的命令，如SETNX（SET if Not eXists）或者 SET 命令的帶有NX（Not eXists）選項，可以用來確保鎖的獲取和釋放是原子的。

String result = jedis.set(lockKey, requestId, "NX", "PX", expireTime);
if ("OK".equals(result)) {
    returntrue;
}
returnfalse;

2.鎖的過期時間

為了保證鎖的釋放，防止死鎖的發生，獲取到的鎖需要設置一個過期時間，也就是說當鎖的持有者因為出現異常情況未能正確的釋放鎖時，鎖也會到達這個時間之后自動釋放，避免對系統造成影響。

try{
  String result = jedis.set(lockKey, requestId, "NX", "PX", expireTime);
  if ("OK".equals(result)) {
      returntrue;
  }
  returnfalse;
} finally {
    unlock(lockKey);
}

此時有些朋友可能就會說，如果釋放鎖的過程中，發生系統異常或者網絡斷線問題，不也會造成鎖的釋放失敗嗎？

是的，這個極小概率的問題確實是存在的。所以我們設置鎖的過期時間就是必須的。當發生異常無法主動釋放鎖的時候，就需要靠過期時間自動釋放鎖了。

不管操作成功與否，都要釋放鎖，不能忘了釋放鎖，可以說鎖的過期時間就是對忘了釋放鎖的一個兜底。

3.鎖的唯一標識

在上面對鎖都加鎖正常的情況下，在鎖釋放時，能正確的釋放自己的鎖嗎，所以每個客戶端應該提供一個唯一的標識符，確保在釋放鎖時能正確的釋放自己的鎖，而不是釋放成為其他的鎖。一般可以使用客戶端的ID作為標識符，在釋放鎖時進行比較，確保只有當持有鎖的客戶端才能釋放自己的鎖。

如果我們加的鎖沒有加入唯一標識，在多線程環境下，可能就會出現釋放了其他線程的鎖的情況發生。

有些朋友可能就會說了，在多線程環境中，線程A加鎖成功之后，線程B在線程A沒有釋放鎖的前提下怎么可以再次獲取到鎖呢？所以也就沒有釋放其他線程的鎖這個說法。

下面我們看這么一個場景，如果線程A執行任務需要10s，鎖的時間是5s，也就是當鎖的過期時間設置的過短，在任務還沒執行成功的時候就釋放了鎖，此時，線程B就會加鎖成功，等線程A執行任務執行完成之后，執行釋放鎖的操作，此時，就把線程B的鎖給釋放了，這不就出問題了嗎。

所以，為了解決這個問題就是在鎖上加入線程的ID或者唯一標識請求ID。對于鎖的過期時間短這個只能根據業務處理時間大概的計算一個時間，還有就是看門狗，進行鎖的續期。

偽代碼如下

if (jedis.get(lockKey).equals(requestId)) {
    jedis.del(lockKey);
    returntrue;
}
returnfalse;

4.鎖非阻塞獲取

非阻塞獲取意味著獲取鎖的操作不會阻塞當前線程或進程的執行。通常，在嘗試獲取鎖時，如果鎖已經被其他客戶端持有，常見的做法是讓當前線程或進程等待直到鎖被釋放。這種方式稱為阻塞獲取鎖。

相比之下，非阻塞獲取鎖不會讓當前線程或進程等待鎖的釋放，而是立即返回獲取鎖的結果。如果鎖已經被其他客戶端持有，那么獲取鎖的操作會失敗，返回一個失敗的結果或者一個空值，而不會阻塞當前線程或進程的執行。

非阻塞獲取鎖通常適用于一些對實時性要求較高、不希望阻塞的場景，比如輪詢等待鎖的釋放。當獲取鎖失敗時，可以立即執行一些其他操作或者進行重試，而不需要等待鎖的釋放。

在 Redis 中，可以使用 SETNX 命令嘗試獲取鎖，如果返回成功（即返回1），表示獲取鎖成功；如果返回失敗（即返回0），表示獲取鎖失敗。通過這種方式，可以實現非阻塞獲取鎖的操作。

try {
  Long start = System.currentTimeMillis();
  while(true) {
     String result = jedis.set(lockKey, requestId, "NX", "PX", expireTime);
     if ("OK".equals(result)) {
        if(!exists(path)) {
           mkdir(path);
        }
        returntrue;
     }
     
     long time = System.currentTimeMillis() - start;
      if (time>=timeout) {
          returnfalse;
      }
      try {
          Thread.sleep(50);
      } catch (InterruptedException e) {
          e.printStackTrace();
      }
  }
} finally{
    unlock(lockKey,requestId);
}  
returnfalse;

在規定的時間范圍內，假如說500ms，自旋不斷獲取鎖，不斷嘗試加鎖。

如果成功，則返回。如果失敗，則休息50ms然后在開始重試獲取鎖。如果到了超時時間，也就是500ms時，則直接返回失敗。

說到了多次嘗試加鎖，在 Redis，分布式鎖是互斥的，假如我們對某個 key 進行了加鎖，如果 該key 對應的鎖還沒有釋放的話，在使用相同的key去加鎖，大概率是會失敗的。

下面有這樣一個場景，需要獲取滿足條件的菜單樹，后臺程序在代碼中遞歸的去獲取，知道獲取到所有的滿足條件的數據。我們要知道，菜單是可能隨時都會變的，所以這個地方是可以加入分布式鎖進行互斥的。

后臺程序在遞歸獲取菜單樹的時候，第一層加鎖成功，第二層、第n層 加鎖不久加鎖失敗了嗎？

遞歸中的加鎖偽代碼如下：

privateint expireTime = 1000;

public void fun(int level,String lockKey,String requestId){
  try{
     String result = jedis.set(lockKey, requestId, "NX", "PX", expireTime);
     if ("OK".equals(result)) {
        if(level<=10){
           this.fun(++level,lockKey,requestId);
        } else {
           return;
        }
     }
     return;
  } finally {
     unlock(lockKey,requestId);
  }
}

如果我們直接使用的話，看起來問題不大，但是真正執行程序之后，就會發現報錯啦。

因為從根節點開始，第一層遞歸加鎖成功之后，還沒有釋放這個鎖，就直接進入到了第二層的遞歸之中。因為鎖名為lockKey，并且值為requestId的鎖已經存在，所以第二層遞歸大概率會加鎖失敗，最后就是返回結果，只有底層遞歸的結果返回了。

所以，我們還需要一個可重入的特性。

5.可重入

redisson 框架中已經實現了可重入鎖的功能，所以我們可以直接使用：

privateint expireTime = 1000;

public void run(String lockKey) {
  RLock lock = redisson.getLock(lockKey);
  this.fun(lock,1);
}

public void fun(RLock lock,int level){
  try{
      lock.lock(5, TimeUnit.SECONDS);
      if(level<=10){
         this.fun(lock,++level);
      } else {
         return;
      }
  } finally {
     lock.unlock();
  }
}

上述的代碼僅供參考，這也只是提供一個思路。

下面我們還是聊一下 redisson 可重入鎖的原理。

加鎖主要通過以下代碼實現的。

if (redis.call('exists', KEYS[1]) == 0) 
then  
   redis.call('hset', KEYS[1], ARGV[2], 1);        
   redis.call('pexpire', KEYS[1], ARGV[1]); 
   return nil; 
end;
if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) 
then  
  redis.call('hincrby', KEYS[1], ARGV[2], 1); 
  redis.call('pexpire', KEYS[1], ARGV[1]); 
  return nil; 
end;
return redis.call('pttl', KEYS[1]);

• KEYS[1]：鎖名
• ARGV[1]：過期時間
• ARGV[2]：uuid + ":" + threadId，可認為是requestId

(1) 先判斷如果加鎖的key不存在，則加鎖。

(2) 接下來判斷如果key和requestId值都存在，則使用hincrby命令給該key和requestId值計數，每次都加1。注意一下，這里就是重入鎖的關鍵，鎖重入一次值就加1。

(3) 如果當前 key 存在，但值不是 requestId ，則返回過期時間。

釋放鎖的腳本如下：

if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) 
then 
  return nil
end
local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1);
if (counter > 0) 
then 
    redis.call('pexpire', KEYS[1], ARGV[2]); 
    return0; 
 else
   redis.call('del', KEYS[1]); 
   redis.call('publish', KEYS[2], ARGV[1]); 
   return1; 
end; 
return nil

• 先判斷如果 鎖名key 和 requestId 值不存在，則直接返回。
• 如果 鎖名key 和 requestId 值存在，則重入鎖減1。
• 如果減1后，重入鎖的 value 值還大于0，說明還有引用，則重試設置過期時間。
• 如果減1后，重入鎖的 value 值還等于0，則可以刪除鎖，然后發消息通知等待線程搶鎖。

6.鎖競爭

對于大量寫入的業務場景，使用普通的分布式鎖就可以實現我們的需求。但是對于寫入操作少的，有大量讀取操作的業務場景，直接使用普通的redis鎖就會浪費性能了。所以對于鎖的優化來說，我們就可以從業務場景，讀寫鎖來區分鎖的顆粒度，盡可能將鎖的粒度變細，提升我們系統的性能。

(1) 讀寫鎖

對于降低鎖的粒度，上面我們知道了讀寫鎖也算事在業務層面進行降低鎖粒度的一種方式，所以下面我們以 redisson 框架為例，看看實現讀寫鎖是如何實現的。

讀鎖：

RReadWriteLock readWriteLock = redisson.getReadWriteLock("readWriteLock");
RLock rLock = readWriteLock.readLock();
try {
    rLock.lock();
    //業務操作
} catch (Exception e) {
    log.error(e);
} finally {
    rLock.unlock();
}

寫鎖：

RReadWriteLock readWriteLock = redisson.getReadWriteLock("readWriteLock");
RLock rLock = readWriteLock.writeLock();
try {
    rLock.lock();
    //業務操作
} catch (InterruptedException e) {
   log.error(e);
} finally {
    rLock.unlock();
}

通過講鎖分為讀鎖與寫鎖，最大的提升之后就在與大大的提高系統的讀性能，因為讀鎖與讀鎖之間是沒有沖突的，不存在互斥，然后又因為業務系統中的讀操作是遠遠多與寫操作的，所以我們在提升了讀鎖的性能的同時，系統整體鎖的性能都得到了提升。

讀寫鎖特點：

• 讀鎖與讀鎖不互斥，可共享
• 讀鎖與寫鎖互斥
• 寫鎖與寫鎖互斥

(2) 分段鎖

上面我們通過業務層面的讀寫鎖進行了鎖粒度的減小，下面我們在通過鎖的分段減少鎖粒度實現鎖性能的提升。

如果你對 concurrentHashMap 的源碼了解的話你就會知道分段鎖的原理了。是的就是你想的那樣，把一個大的鎖劃分為多個小的鎖。

舉個例子，假如我們在秒殺100個商品，那么常規做法就是一個鎖，鎖 100個商品，那么分段的意思就是，將100個商品分成10份，相當于有 10 個鎖，每個鎖鎖定10個商品，這也就提升鎖的性能提升了10倍。

具體的實現就是，在秒殺的過程中，對用戶進行取模操作，算出來當前用戶應該對哪一份商品進行秒殺。

通過上述將大鎖拆分為小鎖的過程，以前多個線程只能爭搶一個鎖，現在可以爭搶10個鎖，大大降低了沖突，提升系統吞吐量。

不過需要注意的就是，使用分段鎖確實可以提升系統性能，但是相對應的就是編碼難度的提升，并且還需要引入取模等算法，所以我們在實際業務中，也要綜合考慮。

7.鎖超時

在上面我們也說過了，因為業務執行時間太長，導致鎖自動釋放了，也就是說業務的執行時間遠遠大于鎖的過期時間，這個時候 Redis 會自動釋放該鎖。

針對這種情況，我們可以使用鎖的續期，增加一個定時任務，如果到了超時時間，業務還沒有執行完成，就需要對鎖進行一個續期。

Timer timer = new Timer(); 
timer.schedule(new TimerTask() {
    @Override
    public void run(Timeout timeout) throws Exception {
      //自動續期邏輯
    }
}, 10000, TimeUnit.MILLISECONDS);

獲取到鎖之后，自動的開啟一個定時任務，每隔 10s 中自動刷新一次過期時間。這種機制就是上面我們提到過的看門狗。

對于自動續期操作，我們還是推薦使用 lua 腳本來實現：

if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then 
   redis.call('pexpire', KEYS[1], ARGV[1]);
  return1; 
end;
return0;

需要注意的一點就是，鎖的續期不是一直續期的，業務如果一直執行不完，到了一個總的超時時間，或者執行續期的次數超過幾次，我們就不再進行續期操作了。

上面我們講了這么幾個點，下面我們來說一下 Redis 集群中的問題，如果發生網絡分區，主從切換問題，那么該怎么解決呢？

8.網絡分區

假設 Redis 初始還是主從，一主三從模式。

Redis 的加鎖操作都是在 master 上操作，成功之后異步不同到 slave上。

當 master 宕機之后，我們就需要在三個slave中選舉一個出來當作 master ，假如說我們選了slave1。

現在有一個鎖A進行加鎖，正好加鎖到 master上，然后 master 還沒有同步到 slave 上，master 就宕機了，此時，后面在來新的線程獲取鎖A，也是可以加鎖成功的，所以分布式鎖也就失效了。

Redisson 框架為了解決這個問題，提供了一個專門的類，就是 RedissonRedLock，使用 RedLock 算法。

RedissonRedLock 解決問題的思路就是多搭建幾個獨立的 Redisson 集群，采用分布式投票算法，少數服從多數這種。假如有5個 Redisson 集群，只要當加鎖成功的集群有5/2+1個節點加鎖成功，意味著這次加鎖就是成功的。

• 搭建幾套相互獨立的 Redis 環境，我們這里搭建5套。
• 每套環境都有一個 redisson node 節點。
• 多個 redisson node 節點組成 RedissonRedLock。
• 環境包括單機、主從、哨兵、集群，可以一種或者多種混合都可以。

我們這個例子以主從為例來說

RedissonRedLock 加鎖過程如下：

• 向當前5個 Redisson node 節點加鎖。
• 如果有3個節點加鎖成功，那么整個 RedissonRedLock 就是加鎖成功的。
• 如果小于3個節點加鎖成功，那么整個加鎖操作就是失敗的。
• 如果中途各個節點加鎖的總耗時，大于等于設置的最大等待時間，直接返回加鎖失敗。

通過上面這個示例可以發現，使用 RedissonRedLock 可以解決多個示例導致的鎖失效的問題。但是帶來的也是整個 Redis 集群的管理問題：

• 管理多套 Redis 環境
• 增加加鎖的成本。有多少個 Redisson node就需要加鎖多少次。

由此可見、在實際的高并發業務中，RedissonRedLock 的使用并不多。

在分布式系統中，CAP 理論應該都是知道的，所以我們在選擇分布式鎖的時候也可以參考這個。

• C(Consistency) 一致性
• A(Acailability) 可用性
• P(Partition tolerance)分區容錯性

所以如果我們的業務場景，更需要數據的一致性，我們可以使用 CP 的分布式鎖，例子 zookeeper。

如果我們更需要的是保證數據的可用性，那么我們可以使用 AP 的分布式鎖，例如 Redis。

其實在我們絕大多數的業務場景中，使用Redis已經可以滿足，因為數據的不一致，我們還可以使用 BASE 理論的最終一致性方案解決。因為如果系統不可用了，對用戶來說體驗肯定不是那么好的。