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 = https://www.cnblogs.com/jingdongkeji/archive/2023/05/31/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 = https://www.cnblogs.com/jingdongkeji/archive/2023/05/31/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可能是更好的選擇

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

轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/554016.html

標籤：其他

上一篇：Oracle 12c/19c PDB資料庫配置自動啟動

下一篇：返回列表

圖解Redis和Zookeeper分布式鎖