Redis中Redisson紅鎖使用原理是什么

這篇“Redis中Redisson紅鎖使用原理是什么”文章的知識點大部分人都不太理解，所以小編給大家總結了以下內容，內容詳細，步驟清晰，具有一定的借鑒價值，希望大家閱讀完這篇文章能有所收獲，下面我們一起來看看這篇“Redis中Redisson紅鎖使用原理是什么”文章吧。

為什么使用Redis的紅鎖

主從結構分布式鎖的問題

實現Redis分布式鎖的最簡單的方法就是在Redis中創建一個key，這個key有一個失效時間（TTL)，以保證鎖最終會被自動釋放掉。當客戶端釋放資源(解鎖）的時候，會刪除掉這個key。

從表面上看似乎效果不錯，但有一個嚴重的單點失敗問題：如果Redis掛了怎么辦？你可能會說，可以通過增加一個slave節點解決這個問題。但這通常是行不通的。這樣做，我們不能實現資源的獨享，因為Redis的主從同步通常是異步的。

在這種場景（主從結構）中存在明顯的競態：

• 客戶端A從master獲取到鎖

• 在master將鎖同步到slave之前，master宕掉了。

• slave節點被晉級為master節點

• 客戶端B從新的master獲取到鎖

• 這個鎖對應的資源之前已經被客戶端A已經獲取到了。安全失效！

有時候程序就是這么巧，比如說正好一個節點掛掉的時候，多個客戶端同時取到了鎖。如果你可以接受這種小概率錯誤，那用這個基于復制的方案就完全沒有問題。否則的話，我們建議你實現下面描述的解決方案。

解決方案：使用紅鎖

簡介

Redis中針對此種情況，引入了紅鎖的概念。紅鎖采用主節點過半機制，即獲取鎖或者釋放鎖成功的標志為：在過半的節點上操作成功。

原理

在Redis的分布式環境中，我們假設有N個Redis master。這些節點完全互相獨立，不存在主從復制或者其他集群協調機制。之前我們已經描述了在Redis單實例下怎么安全地獲取和釋放鎖。我們確保將在每（N）個實例上使用此方法獲取和釋放鎖。在這個樣例中，我們假設有5個Redis master節點，這是一個比較合理的設置，所以我們需要在5臺機器上面或者5臺虛擬機上面運行這些實例，這樣保證他們不會同時都宕掉。

為了取到鎖，客戶端應該執行以下操作：

• 獲取當前Unix時間，以毫秒為單位。

• 依次嘗試從N個實例，使用相同的key和隨機值獲取鎖。

• 向Redis設置鎖時，客戶端應該設置一個網絡連接和響應超時時間，這個超時時間應該小于鎖的失效時間。

• 例如你的鎖自動失效時間為10秒，則超時時間應該在5-50毫秒之間。這樣可以避免服務器端Redis已經掛掉的情況下，客戶端還在死死地等待響應結果。如果服務器端沒有在規定時間內響應，客戶端應該盡快嘗試另外一個Redis實例。

• 客戶端使用當前時間減去開始獲取鎖時間（步驟1記錄的時間）得到獲取鎖使用的時間。

• 僅當從大多數（這里是3個節點）的Redis節點都取到鎖，且使用的時間小于鎖失效時間時，鎖才算獲取成功。

• 如果取到了鎖，key的真正有效時間等于有效時間減去獲取鎖所使用的時間（步驟3計算的結果）。

• 如果因為某些原因，獲取鎖失敗（沒有在至少N/2+1個Redis實例取到鎖或者取鎖時間已經超過了有效時間），客戶端應該在所有的Redis實例上進行解鎖（即便某些Redis實例根本就沒有加鎖成功）。

Redisson紅鎖實例

官網

 官方github：8. 分布式鎖和同步器 · redisson/redisson Wik

基于Redis的Redisson紅鎖RedissonRedLock對象實現了Redlock介紹的加鎖算法。該對象也可以用來將多個RLock對象關聯為一個紅鎖，每個RLock對象實例可以來自于不同的Redisson實例。

RLock lock1 = redissonInstance1.getLock("lock1");
RLock lock2 = redissonInstance2.getLock("lock2");
RLock lock3 = redissonInstance3.getLock("lock3");
 
RedissonRedLock lock = new RedissonRedLock(lock1, lock2, lock3);
// 同時加鎖：lock1 lock2 lock3
// 紅鎖在大部分節點上加鎖成功就算成功。
lock.lock();
...
lock.unlock();

大家都知道，如果負責儲存某些分布式鎖的某些Redis節點宕機以后，而且這些鎖正好處于鎖住的狀態時，這些鎖會出現鎖死的狀態。為了避免這種情況的發生，Redisson內部提供了一個監控鎖的看門狗，它的作用是在Redisson實例被關閉前，不斷的延長鎖的有效期。默認情況下，看門狗的檢查鎖的超時時間是30秒鐘，也可以通過修改Config.lockWatchdogTimeout來另行指定。

另外Redisson還通過加鎖的方法提供了leaseTime的參數來指定加鎖的時間。超過這個時間后鎖便自動解開了。

RedissonRedLock lock = new RedissonRedLock(lock1, lock2, lock3);
// 給lock1，lock2，lock3加鎖，如果沒有手動解開的話，10秒鐘后將會自動解開
lock.lock(10, TimeUnit.SECONDS);
 
// 為加鎖等待100秒時間，并在加鎖成功10秒鐘后自動解開
boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
...
lock.unlock();

Redisson紅鎖原理

RedissonRedLock extends RedissonMultiLock，所以實際上，redLock.tryLock實際調用：org.redisson.RedissonMultiLock.java#tryLock()，進而調用到其同類的tryLock(long waitTime, long leaseTime, TimeUnit unit) ，入參為：tryLock(-1, -1, null)

org.redisson.RedissonMultiLock.java#tryLock(long waitTime, long leaseTime, TimeUnit unit)源碼如下：

final List<RLock> locks = new ArrayList<>();
 
/**
 * Creates instance with multiple {@link RLock} objects.
 * Each RLock object could be created by own Redisson instance.
 *
 * @param locks - array of locks
 */
public RedissonMultiLock(RLock... locks) {
    if (locks.length == 0) {
        throw new IllegalArgumentException("Lock objects are not defined");
    }
    this.locks.addAll(Arrays.asList(locks));
}
 
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
    long newLeaseTime = -1;
    if (leaseTime != -1) {
        newLeaseTime = unit.toMillis(waitTime)*2;
    }
    
    long time = System.currentTimeMillis();
    long remainTime = -1;
    if (waitTime != -1) {
        remainTime = unit.toMillis(waitTime);
    }
    long lockWaitTime = calcLockWaitTime(remainTime);
    /**
     * 1. 允許加鎖失敗節點個數限制（N-(N/2+1)）
     */
    int failedLocksLimit = failedLocksLimit();
    /**
     * 2. 遍歷所有節點通過EVAL命令執行lua加鎖
     */
    List<RLock> acquiredLocks = new ArrayList<>(locks.size());
    for (ListIterator<RLock> iterator = locks.listIterator(); iterator.hasNext();) {
        RLock lock = iterator.next();
        boolean lockAcquired;
        /**
         *  3.對節點嘗試加鎖
         */
        try {
            if (waitTime == -1 && leaseTime == -1) {
                lockAcquired = lock.tryLock();
            } else {
                long awaitTime = Math.min(lockWaitTime, remainTime);
                lockAcquired = lock.tryLock(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS);
            }
        } catch (RedisResponseTimeoutException e) {
            // 如果拋出這類異常，為了防止加鎖成功，但是響應失敗，需要解鎖所有節點
            unlockInner(Arrays.asList(lock));
            lockAcquired = false;
        } catch (Exception e) {
            // 拋出異常表示獲取鎖失敗
            lockAcquired = false;
        }
        
        if (lockAcquired) {
            /**
             *4. 如果獲取到鎖則添加到已獲取鎖集合中
             */
            acquiredLocks.add(lock);
        } else {
            /**
             * 5. 計算已經申請鎖失敗的節點是否已經到達 允許加鎖失敗節點個數限制 （N-(N/2+1)）
             * 如果已經到達， 就認定最終申請鎖失敗，則沒有必要繼續從后面的節點申請了
             * 因為 Redlock 算法要求至少N/2+1 個節點都加鎖成功，才算最終的鎖申請成功
             */
            if (locks.size() - acquiredLocks.size() == failedLocksLimit()) {
                break;
            }
 
            if (failedLocksLimit == 0) {
                unlockInner(acquiredLocks);
                if (waitTime == -1 && leaseTime == -1) {
                    return false;
                }
                failedLocksLimit = failedLocksLimit();
                acquiredLocks.clear();
                // reset iterator
                while (iterator.hasPrevious()) {
                    iterator.previous();
                }
            } else {
                failedLocksLimit--;
            }
        }
 
        /**
         * 6.計算 目前從各個節點獲取鎖已經消耗的總時間，如果已經等于最大等待時間，則認定最終申請鎖失敗，返回false
         */
        if (remainTime != -1) {
            remainTime -= System.currentTimeMillis() - time;
            time = System.currentTimeMillis();
            if (remainTime <= 0) {
                unlockInner(acquiredLocks);
                return false;
            }
        }
    }
 
    if (leaseTime != -1) {
        List<RFuture<Boolean>> futures = new ArrayList<>(acquiredLocks.size());
        for (RLock rLock : acquiredLocks) {
            RFuture<Boolean> future = ((RedissonLock) rLock).expireAsync(unit.toMillis(leaseTime), TimeUnit.MILLISECONDS);
            futures.add(future);
        }
        
        for (RFuture<Boolean> rFuture : futures) {
            rFuture.syncUninterruptibly();
        }
    }
 
    /**
     * 7.如果邏輯正常執行完則認為最終申請鎖成功，返回true
     */
    return true;
}

以上就是關于“Redis中Redisson紅鎖使用原理是什么”這篇文章的內容，相信大家都有了一定的了解，希望小編分享的內容對大家有幫助，若想了解更多相關的知識內容，請關注億速云行業資訊頻道。