寫在前面

在瞭解分散式鎖具體實現方案之前，我們應該先思考一下使用分散式鎖必須要考慮的一些問題。​

• 互斥性：在任意時刻，只能有一個程序持有鎖。

• 防死鎖：即使有一個程序在持有鎖的期間崩潰而未能主動釋放鎖，要有其他方式去釋放鎖從而保證其他程序能獲取到鎖。

• 加鎖和解鎖的必須是同一個程序。

• 鎖的續期問題。

常見的分散式鎖實現方案

• 基於 Redis 實現分散式鎖

• 基於 Zookeeper 實現分散式鎖

本文采用第一種方案，也就是基於 Redis 的分散式鎖實現方案。

Redis 實現分散式鎖主要步驟

1. 指定一個 key 作為鎖標記，存入 Redis 中，指定一個 唯一的使用者標識 作為 value。
2. 當 key 不存在時才能設定值，確保同一時間只有一個客戶端程序獲得鎖，滿足 互斥性 特性。
3. 設定一個過期時間，防止因系統異常導致沒能刪除這個 key，滿足 防死鎖 特性。
4. 當處理完業務之後需要清除這個 key 來釋放鎖，清除 key 時需要校驗 value 值，需要滿足 只有加鎖的人才能釋放鎖 。

特別注意：以上實現步驟考慮到了使用分散式鎖需要考慮的互斥性、防死鎖、加鎖和解鎖必須為同一個程序等問題，但是鎖的續期無法實現。所以，博主採用 Redisson 實現 Redis 的分散式鎖，藉助 Redisson 的 WatchDog 機制 能夠很好的解決鎖續期的問題，同樣 Redisson 也是 Redis 官方推薦分散式鎖實現方案，實現起來較為簡單。

Redisson 實現分散式鎖

具體實現程式碼已經上傳到博主的倉庫，需要的朋友可以在公眾號內回覆 【分散式鎖程式碼】 獲取碼雲或 GitHub 專案下載地址。

下面從加鎖機制、鎖互斥機制、Watch dog 機制、可重入加鎖機制、鎖釋放機制、等五個方面對 Redisson 實現分散式鎖的底層原理進行分析。

加鎖原理

加鎖其實是通過一段 lua 指令碼實現的，如下：

我們把這一段 lua 指令碼抽出來看：

if (redis.call('exists', KEYS[1]) == 0) then " +
   "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
   "redis.call('pexpire', KEYS[1], ARGV[1]); " +
   "return nil; " +
   "end; " +
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
    "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
    "redis.call('pexpire', KEYS[1], ARGV[1]); " +
    "return nil; " +
    "end; " +
"return redis.call('pttl', KEYS[1]);"
 

這裡 KEYS[1] 代表的是你加鎖的 key，比如你自己設定了加鎖的那個鎖 key 就是 “myLock”。

// create a lock
RLock lock = redisson.getLock("myLock");

這裡 ARGV[1] 代表的是鎖 key 的預設生存時間，預設 30 秒。ARGV[2] 代表的是加鎖的客戶端的 ID，類似於下面這樣：285475da-9152-4c83-822a-67ee2f116a79:52。至於最後面的一個 1 是為了後面可重入做的計數統計，後面會有講解到。

我們來看一下在 Redis 中的儲存結構：

127.0.0.1:6379> HGETALL myLock
1) "285475da-9152-4c83-822a-67ee2f116a79:52"
2) "1"

上面這一段加鎖的 lua 指令碼的作用是：第一段 if 判斷語句，就是用 exists myLock 命令判斷一下，如果你要加鎖的那個鎖 key 不存在的話，你就進行加鎖。如何加鎖呢？使用 hincrby 命令設定一個 hash 結構，類似於在 Redis 中使用下面的操作：

127.0.0.1:6379> HINCRBY myLock 285475da-9152-4c83-822a-67ee2f116a79:52 1
(integer) 1

接著會執行 pexpire myLock 30000 命令，設定 myLock 這個鎖 key 的生存時間是 30 秒。到此為止，加鎖完成。

有的小夥伴可能此時就有疑問了，如果此時有第二個客戶端請求加鎖呢？ 這就是下面要說的鎖互斥機制。

鎖互斥機制

此時，如果客戶端 2 來嘗試加鎖，會如何呢？首先，第一個 if 判斷會執行 exists myLock，發現 myLock 這個鎖 key 已經存在了。接著第二個 if 判斷，判斷一下，myLock 鎖 key 的 hash 資料結構中，是否包含客戶端 2 的 ID，這裡明顯不是，因為那裡包含的是客戶端 1 的 ID。所以，客戶端 2 會執行：

return redis.call('pttl', KEYS[1]);

返回的一個數字，這個數字代表了 myLock 這個鎖 key 的剩餘生存時間。

接下來我們看一下 Redissson tryLock 的主流程：

@Override
    public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
        long time = unit.toMillis(waitTime);
        long current = System.currentTimeMillis();
        long threadId = Thread.currentThread().getId();
        // 1.嘗試獲取鎖
        Long ttl = tryAcquire(leaseTime, unit, threadId);
        // lock acquired
        if (ttl == null) {
            return true;
        }

        // 申請鎖的耗時如果大於等於最大等待時間，則申請鎖失敗.
        time -= System.currentTimeMillis() - current;
        if (time <= 0) {
            acquireFailed(threadId);
            return false;
        }

        current = System.currentTimeMillis();

        /**
         * 2.訂閱鎖釋放事件，並通過 await 方法阻塞等待鎖釋放，有效的解決了無效的鎖申請浪費資源的問題：
         * 基於資訊量，當鎖被其它資源佔用時，當前執行緒通過 Redis 的 channel 訂閱鎖的釋放事件，一旦鎖釋放會發訊息通知待等待的執行緒進行競爭.
         *
         * 當 this.await 返回 false，說明等待時間已經超出獲取鎖最大等待時間，取消訂閱並返回獲取鎖失敗.
         * 當 this.await 返回 true，進入迴圈嘗試獲取鎖.
         */
        RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
        // await 方法內部是用 CountDownLatch 來實現阻塞，獲取 subscribe 非同步執行的結果（應用了 Netty 的 Future）
        if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {
            if (!subscribeFuture.cancel(false)) {
                subscribeFuture.onComplete((res, e) -> {
                    if (e == null) {
                        unsubscribe(subscribeFuture, threadId);
                    }
                });
            }
            acquireFailed(threadId);
            return false;
        }

        try {
            // 計算獲取鎖的總耗時，如果大於等於最大等待時間，則獲取鎖失敗.
            time -= System.currentTimeMillis() - current;
            if (time <= 0) {
                acquireFailed(threadId);
                return false;

              }

            /**
             * 3.收到鎖釋放的訊號後，在最大等待時間之內，迴圈一次接著一次的嘗試獲取鎖
             * 獲取鎖成功，則立馬返回 true，
             * 若在最大等待時間之內還沒獲取到鎖，則認為獲取鎖失敗，返回 false 結束迴圈
             */
            while (true) {
                long currentTime = System.currentTimeMillis();

                // 再次嘗試獲取鎖
                ttl = tryAcquire(leaseTime, unit, threadId);
                // lock acquired
                if (ttl == null) {
                    return true;
                }
                // 超過最大等待時間則返回 false 結束迴圈，獲取鎖失敗
                time -= System.currentTimeMillis() - currentTime;
                if (time <= 0) {
                    acquireFailed(threadId);
                    return false;
                }

                /**
                 * 6.阻塞等待鎖（通過訊號量(共享鎖)阻塞,等待解鎖訊息）：
                 */
                currentTime = System.currentTimeMillis();
                if (ttl >= 0 && ttl < time) {
                    //如果剩餘時間(ttl)小於wait time ,就在 ttl 時間內，從Entry的訊號量獲取一個許可(除非被中斷或者一直沒有可用的許可)。
                    getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } else {
                    //則就在wait time 時間範圍內等待可以通過訊號量
                    getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
                }

                // 更新剩餘的等待時間(最大等待時間-已經消耗的阻塞時間)
                time -= System.currentTimeMillis() - currentTime;
                if (time <= 0) {
                    acquireFailed(threadId);
                    return false;
                }
            }
        } finally {
            // 7.無論是否獲得鎖,都要取消訂閱解鎖訊息
            unsubscribe(subscribeFuture, threadId);
        }
//        return get(tryLockAsync(waitTime, leaseTime, unit));
    }

流程分析：

1. 嘗試獲取鎖，返回 null 則說明加鎖成功，返回一個數值，則說明已經存在該鎖，ttl 為鎖的剩餘存活時間。

2. 如果此時客戶端 2 程序獲取鎖失敗，那麼使用客戶端 2 的執行緒 id（其實本質上就是程序 id）通過 Redis 的 channel 訂閱鎖釋放的事件。如果等待的過程中一直未等到鎖的釋放事件通知，當超過最大等待時間則獲取鎖失敗，返回 false，也就是第 39 行程式碼。如果等到了鎖的釋放事件的通知，則開始進入一個不斷重試獲取鎖的迴圈。

3. 迴圈中每次都先試著獲取鎖，並得到已存在的鎖的剩餘存活時間。如果在重試中拿到了鎖，則直接返回。如果鎖當前還是被佔用的，那麼等待釋放鎖的訊息，具體實現使用了 JDK 的訊號量 Semaphore 來阻塞執行緒，當鎖釋放併發布釋放鎖的訊息後，訊號量的 release() 方法會被呼叫，此時被訊號量阻塞的等待佇列中的一個執行緒就可以繼續嘗試獲取鎖了。

特別注意：以上過程存在一個細節，這裡有必要說明一下，也是分散式鎖的一個關鍵點：當鎖正在被佔用時，等待獲取鎖的程序並不是通過一個 while(true) 死迴圈去獲取鎖，而是利用了 Redis 的釋出訂閱機制,通過 await 方法阻塞等待鎖的程序，有效的解決了無效的鎖申請浪費資源的問題。

鎖的續期機制

客戶端 1 加鎖的鎖 key 預設生存時間才 30 秒，如果超過了 30 秒，客戶端 1 還想一直持有這把鎖，怎麼辦呢？

Redisson 提供了一個續期機制， 只要客戶端 1 一旦加鎖成功，就會啟動一個 Watch Dog。

private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, long threadId) {
    if (leaseTime != -1) {
        return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
    }
    RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
    ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
        if (e != null) {
            return;
        }

        // lock acquired
        if (ttlRemaining == null) {
            scheduleExpirationRenewal(threadId);
        }
    });
    return ttlRemainingFuture;
}

注意：從以上原始碼我們看到 leaseTime 必須是 -1 才會開啟 Watch Dog 機制，也就是如果你想開啟 Watch Dog 機制必須使用預設的加鎖時間為 30s。如果你自己自定義時間，超過這個時間，鎖就會自定釋放，並不會延長。

private void scheduleExpirationRenewal(long threadId) {
    ExpirationEntry entry = new ExpirationEntry();
    ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);
    if (oldEntry != null) {
        oldEntry.addThreadId(threadId);
    } else {
        entry.addThreadId(threadId);
        renewExpiration();
    }
}

protected RFuture<Boolean> renewExpirationAsync(long threadId) {
    return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
            "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                "return 1; " +
            "end; " +
            "return 0;",
        Collections.<Object>singletonList(getName()),
        internalLockLeaseTime, getLockName(threadId));
}

Watch Dog 機制其實就是一個後臺定時任務執行緒，獲取鎖成功之後，會將持有鎖的執行緒放入到一個 RedissonLock.EXPIRATION_RENEWAL_MAP裡面，然後每隔 10 秒 （internalLockLeaseTime / 3） 檢查一下，如果客戶端 1 還持有鎖 key（判斷客戶端是否還持有 key，其實就是遍歷 EXPIRATION_RENEWAL_MAP 裡面執行緒 id 然後根據執行緒 id 去 Redis 中查，如果存在就會延長 key 的時間），那麼就會不斷的延長鎖 key 的生存時間。

注意：這裡有一個細節問題，如果服務宕機了，Watch Dog 機制執行緒也就沒有了，此時就不會延長 key 的過期時間，到了 30s 之後就會自動過期了，其他執行緒就可以獲取到鎖。

可重入加鎖機制

Redisson 也是支援可重入鎖的，比如下面這種程式碼：

@Override
public void lock() {
    RLock lock = redissonSingle.getLock("myLock");
    try {
        lock.lock();

        // 執行業務
        doBusiness();

        lock.lock();

    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        // 釋放鎖
        lock.unlock();
        lock.unlock();
        logger.info("任務執行完畢, 釋放鎖!");
    }
}

我們再分析一下加鎖那段 lua 程式碼：

if (redis.call('exists', KEYS[1]) == 0) then " +
   "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
   "redis.call('pexpire', KEYS[1], ARGV[1]); " +
   "return nil; " +
   "end; " +
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
    "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
    "redis.call('pexpire', KEYS[1], ARGV[1]); " +
    "return nil; " +
    "end; " +
"return redis.call('pttl', KEYS[1]);"

第一個 if 判斷肯定不成立，exists myLock 會顯示鎖 key 已經存在。第二個 if 判斷會成立，因為 myLock 的 hash 資料結構中包含的那個 ID 即客戶端 1 的 ID，此時就會執行可重入加鎖的邏輯，使用：hincrby myLock 285475da-9152-4c83-822a-67ee2f116a79:52 1 對客戶端 1 的加鎖次數加 1。此時 myLock 資料結構變為下面這樣：

127.0.0.1:6379> HGETALL myLock
1) "285475da-9152-4c83-822a-67ee2f116a79:52"
2) "2"

到這裡，小夥伴本就都明白了 hash 結構的 key 是鎖的名稱，field 是客戶端 ID，value 是該客戶端加鎖的次數。

這裡有一個細節，如果加鎖支援可重入鎖，那麼解鎖呢？

釋放鎖機制

執行

lock.unlock()

就可以釋放分散式鎖。我們來看一下釋放鎖的流程程式碼：

@Override
public RFuture<Void> unlockAsync(long threadId) {
    RPromise<Void> result = new RedissonPromise<Void>();
    // 1. 非同步釋放鎖
    RFuture<Boolean> future = unlockInnerAsync(threadId);
    // 取消 Watch Dog 機制
    future.onComplete((opStatus, e) -> {
        cancelExpirationRenewal(threadId);

        if (e != null) {
            result.tryFailure(e);
            return;
        }

        if (opStatus == null) {
            IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
                    + id + " thread-id: " + threadId);
            result.tryFailure(cause);
            return;
        }

        result.trySuccess(null);
    });

    return result;
}

protected RFuture<Boolean> unlockInnerAsync(long threadId) {
    return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
            // 判斷鎖 key 是否存在
            "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
                "return nil;" +
            "end; " +
            // 將該客戶端對應的鎖的 hash 結構的 value 值遞減為 0 後再進行刪除
            // 然後再向通道名為 redisson_lock__channel publish 一條 UNLOCK_MESSAGE 資訊
            "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
            "if (counter > 0) then " +
                "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                "return 0; " +
            "else " +
                "redis.call('del', KEYS[1]); " +
                "redis.call('publish', KEYS[2], ARGV[1]); " +
                "return 1; "+
            "end; " +
            "return nil;",
            Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));
}

從以上程式碼來看，釋放鎖的步驟主要分三步：

1. 刪除鎖（這裡注意可重入鎖，在上面的指令碼中有詳細分析）。

2. 廣播釋放鎖的訊息，通知阻塞等待的程序（向通道名為 redisson_lock__channel publish 一條 UNLOCK_MESSAGE 資訊）。

3. 取消 Watch Dog 機制，即將 RedissonLock.EXPIRATION_RENEWAL_MAP 裡面的執行緒 id 刪除，並且 cancel 掉 Netty 的那個定時任務執行緒。

方案優點

1. Redisson 通過 Watch Dog 機制很好的解決了鎖的續期問題。

2. 和 Zookeeper 相比較，Redisson 基於 Redis 效能更高，適合對效能要求高的場景。

3. 通過 Redisson 實現分散式可重入鎖，比原生的 SET mylock userId NX PX milliseconds + lua 實現的效果更好些，雖然基本原理都一樣，但是它幫我們遮蔽了內部的執行細節。

4. 在等待申請鎖資源的程序等待申請鎖的實現上也做了一些優化，減少了無效的鎖申請，提升了資源的利用率。

方案缺點

1. 使用 Redisson 實現分散式鎖方案最大的問題就是如果你對某個 Redis Master 例項完成了加鎖，此時 Master 會非同步複製給其對應的 slave 例項。但是這個過程中一旦 Master 宕機，主備切換，slave 變為了 Master。接著就會導致，客戶端 2 來嘗試加鎖的時候，在新的 Master 上完成了加鎖，而客戶端 1 也以為自己成功加了鎖，此時就會導致多個客戶端對一個分散式鎖完成了加鎖，這時系統在業務語義上一定會出現問題，導致各種髒資料的產生。所以這個就是 Redis Cluster 或者說是 Redis Master-Slave 架構的主從非同步複製導致的 Redis 分散式鎖的最大缺陷（在 Redis Master 例項宕機的時候，可能導致多個客戶端同時完成加鎖）。

2. 有個別觀點說使用 Watch Dog 機制開啟一個定時執行緒去不斷延長鎖的時間對系統有所損耗（這裡只是網路上的一種說法，博主查了很多資料並且結合實際生產並不認為有很大系統損耗，這個僅供大家參考）。

總結

以上就是基於 Redis 使用 Redisson 實現分散式鎖的所有原理分析，希望可以幫助小夥伴們對分散式鎖的理解有所加深。其實分析完原始碼後發現基於 Redis 自己手動實現一個簡版的分散式鎖工具也並不是很難，有興趣的小夥伴可以試試。

參考

• https://juejin.im/post/5e828328f265da47cd355a5d#heading-6
• https://juejin.im/post/5e1977dd6fb9a02fbb76e8eb#heading-0
• https://blog.csdn.net/tianyaleixiaowu/article/details/96112684
• https://www.cnblogs.com/qdhxhz/p/11046905.html

最後

歡迎大家關注我的關注我的公眾號一起探討研究技術。

相關推薦