Why 分布式鎖

java.util.concurrent.locks 中包含了 JDK 提供的在多線程情況下對共享資源的訪問控制的一系列工具，它們可以幫助我們解決進程內多線程并發時的數據一致性問題。

但是在分布式系統中，JDK 原生的并發鎖工具在一些場景就無法滿足我們的要求了，這就是為什么要使用分布式鎖。我總結了一句話，分布式鎖是用于解決分布式系統中操作共享資源時的數據一致性問題。

設計分布式鎖要注意的問題

互斥

分布式系統中運行著多個節點，必須確保在同一時刻只能有一個節點的一個線程獲得鎖，這是最基本的一點。

死鎖

分布式系統中，可能產生死鎖的情況要相對復雜一些。分布式系統是處在復雜網絡環境中的，當一個節點獲取到鎖，如果它在釋放鎖之前掛掉了，或者因網絡故障無法執行釋放鎖的命令，都會導致其他節點無法申請到鎖。

因此分布式鎖有必要設置時效，確保在未來的一定時間內，無論獲得鎖的節點發生了什么問題，最終鎖都能被釋放掉。

性能

對于訪問量大的共享資源，如果針對其獲取鎖時造成長時間的等待，導致大量節點阻塞，是絕對不能接受的。

所以設計分布式鎖時要能夠掌握鎖持有者的動態，若判斷鎖持有者處于不活動狀態，要能夠強制釋放其持有的鎖。
此外，排隊等待鎖的節點如果不知道鎖何時會被釋放，則只能隔一段時間嘗試獲取一次鎖，這樣無法保證資源的高效利用，因此當鎖釋放時，要能夠通知等待隊列，使一個等待節點能夠立刻獲得鎖。

重入

考慮到一些應用場景和資源的高效利用，鎖要設計成可重入的，就像 JDK 中的 ReentrantLock 一樣，同一個線程可以重復拿到同一個資源的鎖。

RedissonLock 實現解讀

本文中 Redisson 的代碼版本為 2.2.17-SNAPSHOT。

這里以 lock() 方法為例，其他一系列方法與其核心實現基本一致。

先來看 lock() 的基本用法

RLock lock = redisson.getLock("foobar"); // 1.獲得鎖對象實例 lock.lock(); // 2.獲取分布式鎖 try {// do sth. } finally {lock.unlock(); // 3.釋放鎖 }

通過 RedissonClient 的 getLock() 方法取得一個 RLock 實例。

lock() 方法嘗試獲取鎖，如果成功獲得鎖，則繼續往下執行，否則等待鎖被釋放，然后再繼續嘗試獲取鎖，直到成功獲得鎖。

unlock() 方法釋放獲得的鎖，并通知等待的節點鎖已釋放。

下面來看看 RedissonLock 的具體實現

org.redisson.Redisson#getLock()

@Override public RLock getLock(String name) {return new RedissonLock(commandExecutor, name, id); }

這里的 RLock 是繼承自 java.util.concurrent.locks.Lock 的一個 interface，getLock 返回的實際上是其實現類 RedissonLock 的實例。

來看看構造 RedissonLock 的參數

• commandExecutor: 與 Redis 節點通信并發送指令的真正實現。需要說明一下，Redisson 缺省的 CommandExecutor 實現是通過 eval 命令來執行 Lua 腳本，所以要求 Redis 的版本必須為 2.6 或以上，否則你可能要自己來實現 CommandExecutor。關于 Redisson 的 CommandExecutor 以后會專門解讀，所以本次就不多說了。
• name: 鎖的全局名稱，例如上面代碼中的 "foobar"，具體業務中通常可能使用共享資源的唯一標識作為該名稱。
• id: Redisson 客戶端唯一標識，實際上就是一個 UUID.randomUUID()。

org.redisson.RedissonLock#lock()

此處略過前面幾個方法的層層調用，直接看最核心部分的方法 lockInterruptibly()，該方法在 RLock 中聲明，支持對獲取鎖的線程進行中斷操作。在直接使用 lock() 方法獲取鎖時，最后實際執行的是 lockInterruptibly(-1, null)。

@Override public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException {// 1.嘗試獲取鎖Long ttl = tryAcquire(leaseTime, unit);// 2.獲得鎖成功if (ttl == null) {return;}// 3.等待鎖釋放，并訂閱鎖long threadId = Thread.currentThread().getId();Future<RedissonLockEntry> future = subscribe(threadId);get(future);try {while (true) {// 4.重試獲取鎖ttl = tryAcquire(leaseTime, unit);// 5.成功獲得鎖if (ttl == null) {break;}// 6.等待鎖釋放if (ttl >= 0) {getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);} else {getEntry(threadId).getLatch().acquire();}}} finally {// 7.取消訂閱unsubscribe(future, threadId);} }

首先嘗試獲取鎖，具體代碼下面再看，返回結果是已存在的鎖的剩余存活時間，為 null 則說明沒有已存在的鎖并成功獲得鎖。

如果獲得鎖則結束流程，回去執行業務邏輯。

如果沒有獲得鎖，則需等待鎖被釋放，并通過 Redis 的 channel 訂閱鎖釋放的消息，這里的具體實現本文也不深入，只是簡單提一下 Redisson 在執行 Redis 命令時提供了同步和異步的兩種實現，但實際上同步的實現都是基于異步的，具體做法是使用 Netty 中的異步工具 Future 和 FutureListener 結合 JDK 中的 CountDownLatch 一起實現。

訂閱鎖的釋放消息成功后，進入一個不斷重試獲取鎖的循環，循環中每次都先試著獲取鎖，并得到已存在的鎖的剩余存活時間。

如果在重試中拿到了鎖，則結束循環，跳過第 6 步。

如果鎖當前是被占用的，那么等待釋放鎖的消息，具體實現使用了 JDK 并發的信號量工具 Semaphore 來阻塞線程，當鎖釋放并發布釋放鎖的消息后，信號量的 release() 方法會被調用，此時被信號量阻塞的等待隊列中的一個線程就可以繼續嘗試獲取鎖了。

在成功獲得鎖后，就沒必要繼續訂閱鎖的釋放消息了，因此要取消對 Redis 上相應 channel 的訂閱。

下面著重看看 tryAcquire() 方法的實現，

private Long tryAcquire(long leaseTime, TimeUnit unit) {// 1.將異步執行的結果以同步的形式返回return get(tryAcquireAsync(leaseTime, unit, Thread.currentThread().getId())); }private <T> Future<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, long threadId) {if (leaseTime != -1) {return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);}// 2.用默認的鎖超時時間去獲取鎖Future<Long> ttlRemainingFuture = tryLockInnerAsync(LOCK_EXPIRATION_INTERVAL_SECONDS,TimeUnit.SECONDS, threadId, RedisCommands.EVAL_LONG);ttlRemainingFuture.addListener(new FutureListener<Long>() {@Overridepublic void operationComplete(Future<Long> future) throws Exception {if (!future.isSuccess()) {return;}Long ttlRemaining = future.getNow();// 成功獲得鎖if (ttlRemaining == null) {// 3.鎖過期時間刷新任務調度scheduleExpirationRenewal();}}});return ttlRemainingFuture; }<T> Future<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId,RedisStrictCommand<T> command) {internalLockLeaseTime = unit.toMillis(leaseTime);// 3.使用 EVAL 命令執行 Lua 腳本獲取鎖return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,"if (redis.call('exists', KEYS[1]) == 0) then " +"redis.call('hset', 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]);",Collections.<Object>singletonList(getName()), internalLockLeaseTime,getLockName(threadId)); }

上面說過 Redisson 實現的執行 Redis 命令都是異步的，但是它在異步的基礎上提供了以同步的方式獲得執行結果的封裝。

前面提到分布式鎖要確保未來的一段時間內鎖一定能夠被釋放，因此要對鎖設置超時釋放的時間，在我們沒有指定該時間的情況下，Redisson 默認指定為30秒。

在成功獲取到鎖的情況下，為了避免業務中對共享資源的操作還未完成，鎖就被釋放掉了，需要定期（鎖失效時間的三分之一）刷新鎖失效的時間，這里 Redisson 使用了 Netty 的 TimerTask、Timeout 工具來實現該任務調度。

獲取鎖真正執行的命令，Redisson 使用 EVAL 命令執行上面的 Lua 腳本來完成獲取鎖的操作：

如果通過 exists 命令發現當前 key 不存在，即鎖沒被占用，則執行 hset 寫入 Hash 類型數據 key:全局鎖名稱（例如共享資源ID）, field:鎖實例名稱（Redisson客戶端ID:線程ID）, value:1，并執行 pexpire 對該 key 設置失效時間，返回空值 nil，至此獲取鎖成功。

如果通過 hexists 命令發現 Redis 中已經存在當前 key 和 field 的 Hash 數據，說明當前線程之前已經獲取到鎖，因為這里的鎖是可重入的，則執行 hincrby 對當前 key field 的值加一，并重新設置失效時間，返回空值，至此重入獲取鎖成功。

最后是鎖已被占用的情況，即當前 key 已經存在，但是 Hash 中的 Field 與當前值不同，則執行 pttl 獲取鎖的剩余存活時間并返回，至此獲取鎖失敗。

以上就是對 lock() 的解讀，不過在實際業務中我們可能還會經常使用 tryLock()，雖然兩者有一定差別，但核心部分的實現都是相同的，另外還有其他一些方法可以支持更多自定義參數，本文中就不一一詳述了。

org.redisson.RedissonLock#unlock()

最后來看鎖的釋放，

@Override public void unlock() {// 1.通過 EVAL 和 Lua 腳本執行 Redis 命令釋放鎖Boolean opStatus = commandExecutor.evalWrite(getName(), LongCodec.INSTANCE,RedisCommands.EVAL_BOOLEAN,"if (redis.call('exists', KEYS[1]) == 0) then " +"redis.call('publish', KEYS[2], ARGV[1]); " +"return 1; " +"end;" +"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +"return nil;" +"end; " +"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.unlockMessage, internalLockLeaseTime, getLockName(Thread.currentThread().getId()));// 2.非鎖的持有者釋放鎖時拋出異常if (opStatus == null) {throw new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "+ id + " thread-id: " + Thread.currentThread().getId());}// 3.釋放鎖后取消刷新鎖失效時間的調度任務if (opStatus) {cancelExpirationRenewal();} }

使用 EVAL 命令執行 Lua 腳本來釋放鎖：

key 不存在，說明鎖已釋放，直接執行 publish 命令發布釋放鎖消息并返回 1。

key 存在，但是 field 在 Hash 中不存在，說明自己不是鎖持有者，無權釋放鎖，返回 nil。

因為鎖可重入，所以釋放鎖時不能把所有已獲取的鎖全都釋放掉，一次只能釋放一把鎖，因此執行 hincrby 對鎖的值減一。

釋放一把鎖后，如果還有剩余的鎖，則刷新鎖的失效時間并返回 0；如果剛才釋放的已經是最后一把鎖，則執行 del 命令刪除鎖的 key，并發布鎖釋放消息，返回 1。

上面執行結果返回 nil 的情況（即第2中情況），因為自己不是鎖的持有者，不允許釋放別人的鎖，故拋出異常。

執行結果返回 1 的情況，該鎖的所有實例都已全部釋放，所以不需要再刷新鎖的失效時間。

總結

寫了這么多，其實最主要的就是上面的兩段 Lua 腳本，基于 Redis 的分布式鎖的設計完全體現在其中，看完這兩段腳本，再回顧一下前面的 設計分布式鎖要注意的問題 就豁然開朗了。

作者：Raymond_Z
鏈接：https://www.jianshu.com/p/de5a69622e49
來源：簡書
簡書著作權歸作者所有，任何形式的轉載都請聯系作者獲得授權并注明出處。

總結

以上是生活随笔為你收集整理的Redisson 分布式锁实现分析（一）的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。