前言

前面已經學習了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)

 1RLock lock1 = redisson1.getLock("lock1");
 2RLock lock2 = redisson2.getLock("lock2");
 3RLock lock3 = redisson3.getLock("lock3");
 4
 5RLock redLock = anyRedisson.getRedLock(lock1, lock2, lock3);
 6
 7// traditional lock method
 
 8redLock.lock();
 9
10// or acquire lock and automatically unlock it after 10 seconds
11redLock.lock(10, TimeUnit.SECONDS);
12
13// or wait for lock aquisition up to 100 seconds 
14// and automatically unlock it after 10 seconds
15boolean res = redLock.tryLock(100, 10, TimeUnit.SECONDS);
16if (res) {
17   try {
18     ...
19   } finally {
20       redLock.unlock();
21   }
22}

這裡是分別對3個redis例項加鎖，然後獲取一個最後的加鎖結果。

RedissonRedLock實現原理

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

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

 1public class RedissonRedLock extends RedissonMultiLock {
 2    public RedissonRedLock(RLock... locks) {
 3        super(locks);
 4    }
 5
 6    /**
 7     * 鎖可以失敗的次數，鎖的數量-鎖成功客戶端最小的數量
 8     */
 9    @Override
10    protected int failedLocksLimit() {
11        return locks.size() - minLocksAmount(locks);
12    }
13
14    /**
15     * 鎖的數量 / 2 + 1，例如有3個客戶端加鎖，那麼最少需要2個客戶端加鎖成功
16     */
17    protected int minLocksAmount(final List<RLock> locks) {
18        return locks.size()/2 + 1;
19    }
20
21    /** 
22     * 計算多個客戶端一起加鎖的超時時間，每個客戶端的等待時間
23     * remainTime預設為4.5s
24     */
25    @Override
26    protected long calcLockWaitTime(long remainTime) {
27        return Math.max(remainTime / locks.size(), 1);
28    }
29
30    @Override
31    public void unlock() {
32        unlockInner(locks);
33    }
34
35}

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

接著我們看下lock()的具體實現

RedissonMultiLock實現原理

  1public class RedissonMultiLock implements Lock {
  2
  3    final List<RLock> locks = new ArrayList<RLock>();
  4
  5    public RedissonMultiLock(RLock... locks) {
  6        if (locks.length == 0) {
  7            throw new IllegalArgumentException("Lock objects are not defined");
  8        }
  9        this.locks.addAll(Arrays.asList(locks));
 10    }
 11
 12    public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
 13        long newLeaseTime = -1;
 14        if (leaseTime != -1) {
 15            // 如果等待時間設定了，那麼將等待時間 * 2
 16            newLeaseTime = unit.toMillis(waitTime)*2;
 17        }
 18
 19        // time為當前時間戳
 20        long time = System.currentTimeMillis();
 21        long remainTime = -1;
 22        if (waitTime != -1) {
 23            remainTime = unit.toMillis(waitTime);
 24        }
 25        // 計算鎖的等待時間，RedLock中：如果remainTime=-1，那麼lockWaitTime為1
 26        long lockWaitTime = calcLockWaitTime(remainTime);
 27
 28        // RedLock中failedLocksLimit即為n/2 + 1
 29        int failedLocksLimit = failedLocksLimit();
 30        List<RLock> acquiredLocks = new ArrayList<RLock>(locks.size());
 31        // 迴圈每個redis客戶端，去獲取鎖
 32        for (ListIterator<RLock> iterator = locks.listIterator(); iterator.hasNext();) {
 33            RLock lock = iterator.next();
 34            boolean lockAcquired;
 35            try {
 36                // 呼叫tryLock方法去獲取鎖，如果獲取鎖成功，則lockAcquired=true
 37                if (waitTime == -1 && leaseTime == -1) {
 38                    lockAcquired = lock.tryLock();
 39                } else {
 40                    long awaitTime = Math.min(lockWaitTime, remainTime);
 41                    lockAcquired = lock.tryLock(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS);
 42                }
 43            } catch (Exception e) {
 44                lockAcquired = false;
 45            }
 46
 47            // 如果獲取鎖成功，將鎖加入到list集合中
 48            if (lockAcquired) {
 49                acquiredLocks.add(lock);
 50            } else {
 51                // 如果獲取鎖失敗，判斷失敗次數是否等於失敗的限制次數
 52                // 比如，3個redis客戶端，最多隻能失敗1次
 53                // 這裡locks.size = 3, 3-x=1，說明只要成功了2次就可以直接break掉迴圈
 54                if (locks.size() - acquiredLocks.size() == failedLocksLimit()) {
 55                    break;
 56                }
 57
 58                // 如果最大失敗次數等於0
 59                if (failedLocksLimit == 0) {
 60                    // 釋放所有的鎖，RedLock加鎖失敗
 61                    unlockInner(acquiredLocks);
 62                    if (waitTime == -1 && leaseTime == -1) {
 63                        return false;
 64                    }
 65                    failedLocksLimit = failedLocksLimit();
 66                    acquiredLocks.clear();
 67                    // 重置迭代器 重試再次獲取鎖
 68                    while (iterator.hasPrevious()) {
 69                        iterator.previous();
 70                    }
 71                } else {
 72                    // 失敗的限制次數減一
 73                    // 比如3個redis例項，最大的限制次數是1，如果遍歷第一個redis例項，失敗了，那麼failedLocksLimit會減成0
 74                    // 如果failedLocksLimit就會走上面的if邏輯，釋放所有的鎖，然後返回false
 75                    failedLocksLimit--;
 76                }
 77            }
 78
 79            if (remainTime != -1) {
 80                remainTime -= (System.currentTimeMillis() - time);
 81                time = System.currentTimeMillis();
 82                if (remainTime <= 0) {
 83                    unlockInner(acquiredLocks);
 84                    return false;
 85                }
 86            }
 87        }
 88
 89        if (leaseTime != -1) {
 90            List<RFuture<Boolean>> futures = new ArrayList<RFuture<Boolean>>(acquiredLocks.size());
 91            for (RLock rLock : acquiredLocks) {
 92                RFuture<Boolean> future = rLock.expireAsync(unit.toMillis(leaseTime), TimeUnit.MILLISECONDS);
 93                futures.add(future);
 94            }
 95
 96            for (RFuture<Boolean> rFuture : futures) {
 97                rFuture.syncUninterruptibly();
 98            }
 99        }
100
101        return true;
102    }
103}

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

申明

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

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

相關推薦