使用Redis還是Zookeeper來實現分散式鎖，最終還是要基於業務來決定，可以參考以下兩種情況：（1）如果業務併發量很大，Redis分散式鎖高效的讀寫性能更能支持高併發（2）如果業務要求鎖的強一致性，那麼使用Zookeeper可能是更好的選擇 ...

1.基於Redis實現分散式鎖

Redis分散式鎖原理如上圖所示，當有多個Set命令發送到Redis時，Redis會串列處理，最終只有一個Set命令執行成功，從而只有一個線程加鎖成功

2：SetNx命令加鎖

利用_Redis的setNx命令在Redis資料庫中創建一個<Key，Value>記錄，這條命令只有當Redis中沒有這個Key的時候才執行成功，當已經有這個Key的時候會返回失敗_

利用如上的_setNx命令便可以簡單的實現加鎖功能，當多個線程去執行這個加鎖命令時，只有一個線程執行成功，然後執行業務邏輯，其他線程加鎖失敗返回或者重試_

3：死鎖問題

上面的_setNx命令實現了基本的加鎖功能，但存在一個致命的問題是，當程式在執行業務代碼崩潰時，無法再執行到下麵的解鎖指令，從而導致出現死鎖問題_

為瞭解決死鎖問題，這裡就需要_引入過期時間的概念，過期時間是給當前這個key設置一定的存活時間，當存活時間到期後，Redis就會自動刪除這個過期的Key，從而使得程式在崩潰時也能到期自動釋放鎖_

如上圖所示，使用Redis的_expire命令來為鎖設置過期時間，從而實現到期自動解鎖的功能，但這裡仍然還存在一個問題就是加鎖與給鎖設置過期時間這兩個操作命令並不是原子命令_

考慮下麵這種情況：

當程式在加鎖完成後，在設置過期時間前崩潰，這時仍然會造成鎖無法自動釋放，從而產生死鎖現象

4：使用原子命令

針對上面加鎖與設置過期時間不是原子命令的問題，Redis為我們提供了一個原子命令如下：

通過_SetNx(key,value,timeOut)這個結合加鎖與設置過期時間的原子命令_就能完整的實現基於Redis的分散式鎖的加鎖步驟

5：解鎖原理

解鎖原理就是基於Redis的_del刪除key指令_

6：錯誤刪除鎖問題

上面直接刪除key來解鎖方式會存在一個問題，考慮下麵這種情況：

（1）線程1執行業務時間過長導致自己加的鎖過期

（2）這時線程2進來加鎖成功

（3）然後線程1業務邏輯執行完畢開始執行del key命令

（4）這時就會出現錯誤刪除線程2加的鎖

（5）錯誤刪除線程2的鎖後，線程3又可以加鎖成功，導致有兩個線程執行業務代碼

7：加入鎖標識

為瞭解決這種錯誤刪除其他線程的鎖的問題，在這裡需要對加鎖命令進行改造，需要在value欄位裡加入當前線程的id，在這裡可以使用uuid來實現。線程在刪除鎖的時候，用自己的uuid與Redis中鎖的uuid進行比較，如果是自己的鎖就進行刪除，不是則不刪除

如上圖所示，加鎖時_在value欄位中存入當前線程的id，然後在解鎖時通過比較當前的鎖是否是自己的來判斷是否加鎖成功,這樣就解決了錯誤刪除別人的鎖的問題，但這裡同樣存在原子命令問題，比較並刪除_這個操作並不是原子命令，考慮下麵這種情況

（1）線程1獲取uuid並判斷鎖是自己的

（2）準備解鎖時出現GC或者其他原因導致程式卡頓無法立即執行Del命令，導致線程1的鎖過期

（3）線程2就會在這個時候加鎖成功

（4）線程1卡頓結束繼續執行解鎖指令，就會錯誤刪除線程2的鎖

這個問題出現的根本原因還是_比較並刪除這兩個操作並不是原子命令，只要兩個命令被打斷就有可能出現併發問題，如果將兩個命令變為原子命令就能解決這個問題_

8：引入lua腳本實現原子刪除操作

_lua腳本_是一個非常輕量級的腳本語言，Redis底層天生支持lua腳本的執行，一個lua腳本中可以包含多條Redis命令，Redis會將整個lua腳本當作原子操作來執行，從而實現聚合多條Redis指令的原子操作，其原理如下圖所示：

這裡在解鎖時，使用lua腳本將_比較並刪除_操作變為原子操作

//lua腳本如下
luaScript =  " if redis.call('get',key) == value then
                  return redis.call('del',key) 
               else 
                  return 0 
               end;"

如上面的lua腳本所示，Redis會將整個lua腳本當作一個單獨的命令執行，從而實現多個命令的原子操作，避免多線程競爭問題，最終結合lua腳本實現了一個完整的分散式的加鎖和解鎖過程，偽代碼如下：

uuid = getUUID();
//加鎖
lockResut = redisClient.setNx(key,uuid,timeOut);
if(!lockResult){
    return;
}
try{
   //執行業務邏輯
}finally{
    //解鎖
    redisClient.eval(delLuaScript,keys,values)
}
//解鎖的lua腳本
delLuaScript =  " if redis.call('get',key) == value then
                     return redis.call('del',key) 
                  else 
                     return 0 
                  end;"

到此，我們最終實現了一個加鎖和解鎖功能較為完整的redis分散式鎖了，當然作為一個鎖來說，還有一些其他的功能需要進一步完善，例如_考慮鎖失效問題，可重入問題等_

9：自動續期功能

在執行業務代碼時，由於業務執行時間長，最終可能導致在業務執行過程中，自己的鎖超時，然後鎖自動釋放了，在這種情況下第二個線程就會加鎖成功，從而導致數據不一致的情況發生，如下圖所示：

對於上述的這種情況，原因是由_於設置的過期時間太短或者業務執行時間太長導致鎖過期，但是為了避免死鎖問題又必須設置過期時間，那這就需要引入自動續期的功能，即在加鎖成功時，開啟一個定時任務，自動刷新Redis加鎖key的超時時間，_從而避免上訴情況發生，如下圖所示：

uuid = getUUID();
//加鎖
lockResut = redisClient.setNx(key,uuid,timeOut);
if(!lockResult){
    return;
}
//開啟一個定時任務
new Scheduler(key,time,uuid,scheduleTime)
try{
   //執行業務邏輯
}finally{
    //刪除鎖
    redisClient.eval(delLuaScript,keys,values)
    //取消定時任務
    cancelScheduler(uuid);
}

如上訴代碼所示，在加鎖成功後可以啟動一個定時任務來對鎖進行自動續期，定時任務的執行邏輯是：

（1）判斷Redis中的鎖是否是自己的

（2）如果存在的話就使用expire命令重新設置過期時間

這裡由於需要兩個Redis的命令，所以也需要使用lua腳本來實現原子操作，代碼如下所示：

luaScript = "if redis.call('get',key) == value) then
                return redis.call('expire',key,timeOut);
             else
                return 0;
             end；"

10：可重入鎖

對於一個功能完整的鎖來說，可重入功能是必不可少的特性，所謂的鎖可重入就是同一個線程，第一次加鎖成功後，在第二次加鎖時，無需進行排隊等待，只需要判斷是否是自己的鎖就行了，可以直接再次獲取鎖來執行業務邏輯，如下圖所示：

實現可重入機制的原理就是_在加鎖的時候記錄加鎖次數，在釋放鎖的時候減少加鎖次數，這個加鎖的次數記錄可以存在Redis中，如下圖所示：_

如上圖所示，加入可重入功能後，加鎖的步驟就變為如下步驟：

（1）判斷鎖是否存在

（2）判斷鎖是否是自己的

（3）增加加鎖的次數

由於增加次數以及減少次數是多個操作，這裡需要再次使用lua腳本來實現，同時由於這裡需要在Redis中存入加鎖的次數，所以需要使用到Redis中的Map數據結構_Map(key,uuid,lockCount)，_加鎖lua腳本如下：

//鎖不存在
if (redis.call('exists', key) == 0) then
    redis.call('hset', key, uuid, 1); 
    redis.call('expire', key, time); 
    return 1;
end;
//鎖存在，判斷是否是自己的鎖
if (redis.call('hexists', key, uuid) == 1) then
    redis.call('hincrby', key, uuid, 1); 
    redis.call('expire', key, uuid);
    return 1; 
end; 
//鎖不是自己的，返回加鎖失敗
return 0；

加入可重入功能後的解鎖邏輯就變為：

（1）判斷鎖是否是自己的

（2）如果是自己的則減少加鎖次數，否則返回解鎖失敗

//判斷鎖是否是自己的,不是自己的直接返回錯誤
if (redis.call('hexists', key，uuid) == 0) then
    return 0;
end;
//鎖是自己的，則對加鎖次數-1
local counter = redis.call('hincrby', key, uuid, -1);
if (counter > 0) then 
    //剩餘加鎖次數大於0，則不能釋放鎖，重新設置過期時間
    redis.call('expire', key, uuid); 
    return 1;
else
//等於0，代表可以釋放鎖了
    redis.call('del', key); 
    return 1; 
end;

到此，我們在實現基本的_加鎖與解鎖的邏輯上，又加入了可重入和自動續期的功能_，自此一個完整的Redis分散式鎖的雛形就實現了，偽代碼如下：

uuid = getUUID();
//加鎖
lockResut = redisClient.eval(addLockLuaScript,keys,values);
if(!lockResult){
    return;
}
//開啟一個定時任務
new Scheduler(key,time,uuid,scheduleTime)
try{
   //執行業務邏輯
}finally{
    //刪除鎖
    redisClient.eval(delLuaScript,keys,values)
    //取消定時任務
    cancelScheduler(uuid);
}

11：Zookeeper實現分散式鎖

Zookeeper是一個分散式協調服務，分散式協調主要是來解決分散式系統中多個應用之間的數據一致性,Zookeeper內部的數據存儲方式類似於文件目錄形式的存儲結構,它的記憶體結果如下圖所示：

12：Zookeeper加鎖原理

在Zookeeper中的指定路徑下創建節點，然後客戶端根據當前路徑下的節點狀態來判斷是否加鎖成功，如下圖一種情況為例，線程1創建節點成功後，線程2再去創建節點就會創建失敗

13：Zookeeper節點類型

持久節點：在Zookeeper中創建後會進行持久儲存，直到客戶端主動刪除

臨時節點：以客戶端會話Session維度創建節點，一旦客戶端會話斷開，節點就會自動刪除

臨時/持久順序節點：在同一個路徑下創建的節點會對每個節點按創建先後順序編號

zookeeper.exists("/watchpath",new Watcher() {
    @Override
    public void process(WatchedEvent event) {
	System.out.println("進入監聽器");
	System.out.println("監聽路徑Path："+event.getPath());
	System.out.println("監聽事件類型EventType："+event.getType());				
    }			
});

14：利用臨時順序節點和監聽機制來實現分散式鎖

實現分散式鎖的方式有多種，我們可以使用臨時節點和順序節點這種方案來實現分散式鎖：

1：使用臨時節點可以在客戶端程式崩潰時自動釋放鎖，避免死鎖問題

2：使用順序節點的好處是，可以利用鎖釋放的事件監聽機制，來實現_阻塞監聽式的分散式鎖_

下麵將基於這兩個特性來實現分散式鎖

15：加鎖原理

1：首先在Zookeeper上創建臨時順序節點Node01、Node02等

2：第二步客戶端拿到加鎖路徑下所有創建的節點

3：判斷自己的序號是否最小，如果最小的話，代表加鎖成功，如果不是最小的話，就對前一個節點創建監聽器

4：如果前一個節點刪除，監聽器就會通知客戶端來準備重新獲取鎖

加鎖原理和代碼入下圖所示：

//加鎖路徑
String lockPath;
//用來阻塞線程
CountDownLatch cc = new CountDownLatch(1);
//創建鎖節點的路徑
Sting LOCK_ROOT_PATH = "/locks"

//先創建鎖
public void createLock(){
    //lockPath = /locks/lock_01 
    lockPath = zkClient.create(LOCK_ROOT_PATH+"/lock_", CreateMode.EPHEMERAL_SEQUENTIAL);
}

//獲取鎖
public boolean acquireLock(){
    //獲取當前加鎖路徑下所有的節點
    allLocks = zkClient.getChildren("/locks");
    //按節點順序大小排序
    Collections.sort(allLocks);
    //判斷自己是否是第一個節點
    int index = allLocks.indexOf(lockPath.substring(LOCK_ROOT_PATH.length() + 1));
    //如果是第一個節點，則加鎖成功
    if (index == 0) {
        System.out.println(Thread.currentThread().getName() + "獲得鎖成功, lockPath: " + lockPath);
        return true;
    } else {
        //不是序號最小的節點，則監聽前一個節點
        String preLock = allLocks.get(index - 1);
        //創建監聽器
        Stat status = zkClient.exists(LOCK_ROOT_PATH + "/" + preLockPath, watcher);
        // 前一個節點不存在了，則重新獲取鎖
        if (status == null) {
            return acquireLock();
        } else { 
            //阻塞當前進程，直到前一個節點釋放鎖
            System.out.println(" 等待前一個節點鎖釋放，prelocakPath："+preLockPath);
            //喚醒當前線程，繼續嘗試獲取鎖
            cc.await();
            return acquireLock();
        }
    }
}

private Watcher watcher = new Watcher() {
    @Override
    public void process(WatchedEvent event) {
         //監聽到前一個節點釋放鎖，喚醒當前線程
         cc.countDown();
    }
}

16：可重入鎖實現

Zookeeper實現可重入分散式鎖的機制是_在本地維護一個Map記錄，因為如果在Zookeeper節點維護數據的話，Zookeeper的寫操作是很慢，集群內部需要進行投票同步數據，_所以在本地維護一個Map記錄來記錄當前加鎖的次數和加鎖狀態，在釋放鎖的時候減少加鎖的次數，原理如下圖所示：

//利用Map記錄線程持有的鎖
ConcurrentMap<Thread, LockData> lockMap = Maps.newConcurrentMap();
public Boolean lock(){
    Thread currentThread = Thread.currentThread();
    LockData lockData = lockMap.get(currentThread);
    //LockData不為空則說明已經有鎖
    if (lockData != null)    
    {
       //加鎖次數加一
       lockData.lockCount.increment();
       return true;
    }
    //沒有鎖則嘗試獲取鎖
    Boolean lockResult = acquireLock();
    //獲取到鎖
    if (lockResult)
    {
        LockData newLockData = new LockData(currentThread,1);
        lockMap.put(currentThread, newLockData);
        return true;
    }
    //獲取鎖失敗
    return false;
}

17：解鎖原理

解鎖的步驟如下：

（1）判斷鎖是不是自己的

（2）如果是則減少加鎖次數

（3）如果加鎖次數等於0，則釋放鎖，刪除掉創建的臨時節點，下一個監聽這個節點的客戶端會感知到節點刪除事件，從而重新去獲取鎖

public Boolean releaseLock(){
    LockData lockData = lockMap.get(currentThread);
    //沒有鎖
    if(lockData == null){
       return false; 
    }
    //有鎖則加鎖次數減一
    lockCount = lockData.lockCount.decrement();
    if(lockCount > 0){
        return true;
    } 
    //加鎖次數為0
    try{
        //刪除節點
        zkClient.delete(lockPath);
        //斷開連接
        zkClient.close();
    finally{
        //刪除加鎖記錄
        lockMap.remove(currentThread);
    }
    return true;
}

18：Redis和Zookeeper鎖對比

|
|

Redis

Zookeeper

|
|

讀性能

基於記憶體

|
|

加鎖性能

直接寫記憶體加鎖

Master節點創建好後與其他Follower節點進行同步,半數成功後才能返回寫入成功

|
|

數據一致性

AP架構Redis集群之間的數據同步是存在一定的延遲的，當主節點宕機後，數據如果還沒有同步到從節點上，就會導致分散式鎖失效，會造成數據的不一致

CP架構當Leader節點宕機後，會進行集群重新選舉，如果此時只有一部分節點收到了數據的話，會在集群內進行數據同步，保證集群數據的一致性

19：總結

使用Redis還是Zookeeper來實現分散式鎖，最終還是要基於業務來決定，可以參考以下兩種情況：

（1）如果業務併發量很大，Redis分散式鎖高效的讀寫性能更能支持高併發

（2）如果業務要求鎖的強一致性，那麼使用Zookeeper可能是更好的選擇

作者：京東物流鐘磊
來源：京東雲開發者社區

圖解Redis和Zookeeper分散式鎖