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

為什麼使用Redis的紅鎖

主從結構分散式鎖定的問題

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

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

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

• #客戶端A從master取得到鎖定

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

• slave節點被晉升為master節點

• #客戶端B從新的master取得到鎖定

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

程式有時候就是這麼巧，比如說當一個節點掛掉時，多個客戶端剛好同時獲得了鎖定。只要你能容忍這種低機率的錯誤，那麼採用這個基於複製的解決方案就毫無疑問了。否則的話，我們建議你實現下面描述的解決方案。 解決方案：使用紅鎖

簡介

Redis中針對此種情況，引入了紅鎖的概念。在紅鎖系統中，取得或釋放鎖的成功標誌是在超過一半的節點上操作成功。

原理

Assuming there are N Redis masters in the distributed environment of Redis.。這些節點完全互相獨立，不存在主從複製或其他群集協調機制。我們之前已經闡述瞭如何在Redis的單一實例下安全地取得和釋放鎖。我們確保將在每（N）個實例上使用此方法來取得和釋放鎖。在這個範例中，我們假設有5個Redis master節點，這是一個比較合理的設置，所以我們需要在5台機器上面或5台虛擬機器上面運行這些實例，這樣保證他們不會同時都宕掉。

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

• 取得目前Unix時間，以毫秒為單位。

• 依序嘗試從N個實例，使用相同的key和隨機值來取得鎖定。

• 向Redis設定鎖定時，客戶端應該設定一個網路連線和回應逾時時間，而這個逾時時間應該小於鎖定的失效時間。

• 如果您的鎖定自動失效時間為10秒，那麼逾時時間應該設定在5-50毫秒之間。防止客戶端在Redis伺服器已宕機的情況下仍在無休止地等待回應結果。在規定時間內未收到伺服器端回應時，客戶端應盡快嘗試連線其他Redis實例。

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

• 成功取得鎖的條件是必須從大多數 Redis 節點（3個節點）取得鎖，且使用時間不能超過鎖的失效時間。

• 如果取到了鎖，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紅鎖使用原理是什麼的詳細內容。更多資訊請關注PHP中文網其他相關文章！