【分佈式鎖】03-使用Redisson實現RedLock原理

前言

前面已經學習了Redission可重入鎖以及公平鎖的原理，接着看看Redission是如何來實現RedLock的。

RedLock原理

RedLock是基於redis實現的分佈式鎖，它可以保證如下特性：

• 互斥性：在任什麼時候候，只能有一個客戶端可以持有鎖；避免死鎖：
• 當客戶端拿到鎖後，即便發生了網絡分區或者客戶端宕機，也不會發生死鎖；（利用key的存活時間）
• 容錯性：只要多數節點的redis實例正常運行，就可以對外提供服務，加鎖或者釋放鎖；

RedLock算法思想，意思是不能只在一個redis實例上建立鎖，應該是在多個redis實例上建立鎖，n / 2 + 1，必須在大多數redis節點上都成功建立鎖，才能算這個總體的RedLock加鎖成功，避免說僅僅在一個redis實例上加鎖而帶來的問題。

這裏附上一個前幾天對RedLock解析比較透徹的文章：
https://mp.weixin.qq.com/s/gOYWLg3xYt4OhS46woN_Lg

Redisson實現原理

Redisson中有一個MultiLock的概念，能夠將多個鎖合併爲一個大鎖，對一個大鎖進行統一的申請加鎖以及釋放鎖

而Redisson中實現RedLock就是基於MultiLock 去作的，接下來就具體看看對應的實現吧

RedLock使用案例

先看下官方的代碼使用：
(https://github.com/redisson/redisson/wiki/8.-distributed-locks-and-synchronizers#84-redlock)

RLock lock1 = redisson1.getLock("lock1"); RLock lock2 = redisson2.getLock("lock2"); RLock lock3 = redisson3.getLock("lock3"); RLock redLock = anyRedisson.getRedLock(lock1, lock2, lock3); // traditional lock method
redLock.lock(); // or acquire lock and automatically unlock it after 10 seconds
redLock.lock(10, TimeUnit.SECONDS); // or wait for lock aquisition up to 100 seconds // and automatically unlock it after 10 seconds
boolean res = redLock.tryLock(100, 10, TimeUnit.SECONDS); if (res) { try { ... } finally { redLock.unlock(); } }

這裏是分別對3個redis實例加鎖，而後獲取一個最後的加鎖結果。

RedissonRedLock實現原理

上面示例中使用redLock.lock()或者tryLock()最終都是執行RedissonRedLock中方法。

RedissonRedLock 繼承自RedissonMultiLock， 實現了其中的一些方法：

public class RedissonRedLock extends RedissonMultiLock { public RedissonRedLock(RLock... locks) { super(locks); } /** * 鎖能夠失敗的次數，鎖的數量-鎖成功客戶端最小的數量 */ @Override protected int failedLocksLimit() { return locks.size() - minLocksAmount(locks); } /** * 鎖的數量 / 2 + 1，例若有3個客戶端加鎖，那麼最少須要2個客戶端加鎖成功 */
    protected int minLocksAmount(final List<RLock> locks) { return locks.size()/2 + 1; } /** * 計算多個客戶端一塊兒加鎖的超時時間，每一個客戶端的等待時間 * remainTime默認爲4.5s */ @Override protected long calcLockWaitTime(long remainTime) { return Math.max(remainTime / locks.size(), 1); } @Override public void unlock() { unlockInner(locks); } }

看到locks.size()/2 + 1 ，例如咱們有3個客戶端實例，那麼最少2個實例加鎖成功纔算分佈式鎖加鎖成功。

接着咱們看下lock()的具體實現

RedissonMultiLock實現原理

```java public class RedissonMultiLock implements Lock { final List<RLock> locks = new ArrayList<RLock>(); 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) { // 若是等待時間設置了，那麼將等待時間 * 2
            newLeaseTime = unit.toMillis(waitTime)*2; } // time爲當前時間戳
        long time = System.currentTimeMillis(); long remainTime = -1; if (waitTime != -1) { remainTime = unit.toMillis(waitTime); } // 計算鎖的等待時間，RedLock中：若是remainTime=-1，那麼lockWaitTime爲1
        long lockWaitTime = calcLockWaitTime(remainTime); // RedLock中failedLocksLimit即爲n/2 + 1
        int failedLocksLimit = failedLocksLimit(); List<RLock> acquiredLocks = new ArrayList<RLock>(locks.size()); // 循環每一個redis客戶端，去獲取鎖
        for (ListIterator<RLock> iterator = locks.listIterator(); iterator.hasNext();) { RLock lock = iterator.next(); boolean lockAcquired; try { // 調用tryLock方法去獲取鎖，若是獲取鎖成功，則lockAcquired=true
                if (waitTime == -1 && leaseTime == -1) { lockAcquired = lock.tryLock(); } else { long awaitTime = Math.min(lockWaitTime, remainTime); lockAcquired = lock.tryLock(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS); } } catch (Exception e) { lockAcquired = false; } // 若是獲取鎖成功，將鎖加入到list集合中
            if (lockAcquired) { acquiredLocks.add(lock); } else { // 若是獲取鎖失敗，判斷失敗次數是否等於失敗的限制次數 // 好比，3個redis客戶端，最多隻能失敗1次 // 這裏locks.size = 3, 3-x=1，說明只要成功了2次就能夠直接break掉循環
                if (locks.size() - acquiredLocks.size() == failedLocksLimit()) { break; } // 若是最大失敗次數等於0
                if (failedLocksLimit == 0) { // 釋放全部的鎖，RedLock加鎖失敗
 unlockInner(acquiredLocks); if (waitTime == -1 && leaseTime == -1) { return false; } failedLocksLimit = failedLocksLimit(); acquiredLocks.clear(); // 重置迭代器 重試再次獲取鎖
                    while (iterator.hasPrevious()) { iterator.previous(); } } else { // 失敗的限制次數減一 // 好比3個redis實例，最大的限制次數是1，若是遍歷第一個redis實例，失敗了，那麼failedLocksLimit會減成0 // 若是failedLocksLimit就會走上面的if邏輯，釋放全部的鎖，而後返回false
                    failedLocksLimit--; } } 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<RFuture<Boolean>>(acquiredLocks.size()); for (RLock rLock : acquiredLocks) { RFuture<Boolean> future = rLock.expireAsync(unit.toMillis(leaseTime), TimeUnit.MILLISECONDS); futures.add(future); } for (RFuture<Boolean> rFuture : futures) { rFuture.syncUninterruptibly(); } } return true; } }

 

核心代碼都已經加了註釋，實現原理其實很簡單，基於RedLock思想，遍歷全部的Redis客戶端，而後依次加鎖，最後統計成功的次數來判斷是否加鎖成功。

申明

本文章首發自本人博客：https://www.cnblogs.com/wang-meng 和公衆號：壹枝花算不算浪漫，如若轉載請標明來源!

感興趣的小夥伴可關注我的公衆號：壹枝花算不算浪漫

原文出處：https://www.cnblogs.com/wang-meng/p/12536660.html