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

下面我們將探討一下使用Redis實現(xiàn)分布式鎖時需要注意的關(guān)鍵點。

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

1.原子性

在獲取鎖和釋放鎖的過程中，要保證這個操作的原子性，確保加鎖操作與設(shè)置過期時間操作是原子的。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.鎖的過期時間

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

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

此時有些朋友可能就會說，如果釋放鎖的過程中，發(fā)生系統(tǒng)異?；蛘呔W(wǎng)絡(luò)斷線問題，不也會造成鎖的釋放失敗嗎？

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

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

3.鎖的唯一標(biāo)識

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

如果我們加的鎖沒有加入唯一標(biāo)識，在多線程環(huán)境下，可能就會出現(xiàn)釋放了其他線程的鎖的情況發(fā)生。

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

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

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

偽代碼如下

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

4.鎖非阻塞獲取

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

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

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

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

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;

在規(guī)定的時間范圍內(nèi)，假如說500ms，自旋不斷獲取鎖，不斷嘗試加鎖。

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

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

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

后臺程序在遞歸獲取菜單樹的時候，第一層加鎖成功，第二層、第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);
  }
}

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

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

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

5.可重入

redisson 框架中已經(jīng)實現(xiàn)了可重入鎖的功能，所以我們可以直接使用：

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 可重入鎖的原理。

加鎖主要通過以下代碼實現(xiàn)的。

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，可認(rèn)為是requestId

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

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

(3) 如果當(dāng)前 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，說明還有引用，則重試設(shè)置過期時間。
• 如果減1后，重入鎖的 value 值還等于0，則可以刪除鎖，然后發(fā)消息通知等待線程搶鎖。

6.鎖競爭

對于大量寫入的業(yè)務(wù)場景，使用普通的分布式鎖就可以實現(xiàn)我們的需求。但是對于寫入操作少的，有大量讀取操作的業(yè)務(wù)場景，直接使用普通的redis鎖就會浪費性能了。所以對于鎖的優(yōu)化來說，我們就可以從業(yè)務(wù)場景，讀寫鎖來區(qū)分鎖的顆粒度，盡可能將鎖的粒度變細(xì)，提升我們系統(tǒng)的性能。

(1) 讀寫鎖

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

讀鎖：

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

寫鎖：

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

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

讀寫鎖特點：

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

(2) 分段鎖

上面我們通過業(yè)務(wù)層面的讀寫鎖進(jìn)行了鎖粒度的減小，下面我們在通過鎖的分段減少鎖粒度實現(xiàn)鎖性能的提升。

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

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

具體的實現(xiàn)就是，在秒殺的過程中，對用戶進(jìn)行取模操作，算出來當(dāng)前用戶應(yīng)該對哪一份商品進(jìn)行秒殺。

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

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

7.鎖超時

在上面我們也說過了，因為業(yè)務(wù)執(zhí)行時間太長，導(dǎo)致鎖自動釋放了，也就是說業(yè)務(wù)的執(zhí)行時間遠(yuǎn)遠(yuǎn)大于鎖的過期時間，這個時候 Redis 會自動釋放該鎖。

針對這種情況，我們可以使用鎖的續(xù)期，增加一個定時任務(wù)，如果到了超時時間，業(yè)務(wù)還沒有執(zhí)行完成，就需要對鎖進(jìn)行一個續(xù)期。

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

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

對于自動續(xù)期操作，我們還是推薦使用 lua 腳本來實現(xiàn)：

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

需要注意的一點就是，鎖的續(xù)期不是一直續(xù)期的，業(yè)務(wù)如果一直執(zhí)行不完，到了一個總的超時時間，或者執(zhí)行續(xù)期的次數(shù)超過幾次，我們就不再進(jìn)行續(xù)期操作了。

上面我們講了這么幾個點，下面我們來說一下 Redis 集群中的問題，如果發(fā)生網(wǎng)絡(luò)分區(qū)，主從切換問題，那么該怎么解決呢？

8.網(wǎng)絡(luò)分區(qū)

假設(shè) Redis 初始還是主從，一主三從模式。

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

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

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

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

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

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

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

RedissonRedLock 加鎖過程如下：

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

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

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

由此可見、在實際的高并發(fā)業(yè)務(wù)中，RedissonRedLock 的使用并不多。

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

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

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

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

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