如何理解Redisson分布式鎖的源碼

本篇內容介紹了“如何理解Redisson分布式鎖的源碼”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學有所成！

Redis 發布訂閱

之前說過，分布式鎖的核心功能其實就三個：加鎖、解鎖、設置鎖超時。這三個功能也是我們研究Redisson分布式鎖原理的方向。

在學習之前，我們有必要先了解一個知識點，就是有關Redis的發布訂閱功能。

Redis 發布訂閱 (pub/sub) 是一種消息通信模式：發送者 (pub) 發送消息，訂閱者 (sub) 接收消息，發布者可以向指定的渠道  (channel) 發送消息，訂閱者如果訂閱了該頻道的話就能收到消息，從而實現多個客戶端的通信效果。

訂閱的命令是SUBSCRIBE channel[channel  ...]，可以訂閱一個或多個頻道，當有新消息通過PUBLISH命令發送給頻道時，訂閱者就能收到消息，就好像這樣：

開啟兩個客戶端，一個訂閱了頻道channel1，另一個通過PUBLISH發送消息后，訂閱的那個就能收到了，靠這種模式就能實現不同客戶端之間的通信。

關于這種通信模式有哪些妙用場景我們就不展開了，大家可以自己去網上查閱學習一下，我們的主角還是Redisson，熱身完畢，該上主菜了。

Redisson源碼

在使用Redisson加鎖之前，需要先獲取一個RLock實例對象，有了這個對象就可以調用lock、tryLock方法來完成加鎖的功能

Config config = new Config(); config.useSingleServer()   .setPassword("")   .setAddress("redis://127.0.0.1:6379"); RedissonClient redisson = Redisson.create(config); // RLock對象 RLock lock = redisson.getLock("myLock");

配置好對應的host，然后就可以創建一個RLock對象。RLock是一個接口，具體的同步器需要實現該接口，當我們調用redisson.getLock()時，程序會初始化一個默認的同步執行器RedissonLock

這里面初始化了幾個參數，

commandExecutor：異步的Executor執行器，Redisson中所有的命令都是通過...Executor 執行的 ;

id：唯一ID，初始化的時候是用UUID創建的;

internalLockLeaseTime：等待獲取鎖時間，這里讀的是配置類中默認定義的，時間為30秒;

同時，圖片里我還標注了一個方法getEntryName，返回的是 “ID ：鎖名稱”  的字符串，代表的是當前線程持有對應鎖的一個標識，這些參數有必要留個印象，后面的源碼解析中經常會出現。

說完了初始化的東西，我們就可以開始學習加鎖和解鎖的源碼了。

加鎖

Redisson的加鎖方法有兩個，tryLock和lock，使用上的區別在于tryLock可以設置鎖的過期時長leaseTime和等待時長waitTime，核心處理的邏輯都差不多，我們先從tryLock講起。

tryLock

代碼有點長啊。。。整成圖片不太方便，直接貼上來吧，

/**  * @param waitTime 等待鎖的時長   * @param leaseTime 鎖的持有時間   * @param unit 時間單位  * @return  * @throws InterruptedException  */ public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {    // 剩余的等待鎖的時間         long time = unit.toMillis(waitTime);         long current = System.currentTimeMillis();                  final long threadId = Thread.currentThread().getId();         // 嘗試獲取鎖，如果沒取到鎖，則返回鎖的剩余超時時間         Long ttl = tryAcquire(leaseTime, unit, threadId);         // ttl為null，說明可以搶到鎖了，返回true         if (ttl == null) {             return true;         }                  // 如果waitTime已經超時了，就返回false，代表申請鎖失敗         time -= (System.currentTimeMillis() - current);         if (time <= 0) {             acquireFailed(threadId);             return false;         }                  current = System.currentTimeMillis();         // 訂閱分布式鎖, 解鎖時進行通知，看，這里就用到了我們上面說的發布-訂閱了吧         final RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);         // 阻塞等待鎖釋放，await()返回false，說明等待超時了         if (!await(subscribeFuture, time, TimeUnit.MILLISECONDS)) {             if (!subscribeFuture.cancel(false)) {                 subscribeFuture.addListener(new FutureListener<RedissonLockEntry>() {                     @Override                     public void operationComplete(Future<RedissonLockEntry> future) throws Exception {                         if (subscribeFuture.isSuccess()) {                          // 等待都超時了，直接取消訂閱                             unsubscribe(subscribeFuture, threadId);                         }                     }                 });             }             acquireFailed(threadId);             return false;         }          try {             time -= (System.currentTimeMillis() - current);             if (time <= 0) {                 acquireFailed(threadId);                 return false;             }          // 進入死循環，反復去調用tryAcquire嘗試獲取鎖，跟上面那一段拿鎖的邏輯一樣             while (true) {                 long currentTime = System.currentTimeMillis();                 ttl = tryAcquire(leaseTime, unit, threadId);                 // lock acquired                 if (ttl == null) {                     return true;                 }                  time -= (System.currentTimeMillis() - currentTime);                 if (time <= 0) {                     acquireFailed(threadId);                     return false;                 }                  // waiting for message                 currentTime = System.currentTimeMillis();                 if (ttl >= 0 && ttl < time) {                     getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);                 } else {                     getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);                 }                  time -= (System.currentTimeMillis() - currentTime);                 if (time <= 0) {                     acquireFailed(threadId);                     return false;                 }             }         } finally {             unsubscribe(subscribeFuture, threadId);         } //        return get(tryLockAsync(waitTime, leaseTime, unit));     }

代碼還是挺長的，不過流程也就兩步，要么線程拿到鎖返回成功;要么沒拿到鎖并且等待時間還沒過就繼續循環拿鎖，同時監聽鎖是否被釋放。

拿鎖的方法是tryAcquire，傳入的參數分別是鎖的持有時間，時間單位以及代表當前線程的ID，跟進代碼查看調用棧，它會調到一個叫做tryAcquireAsync的方法：

private Long tryAcquire(long leaseTime, TimeUnit unit, long threadId) {     return get(tryAcquireAsync(leaseTime, unit, threadId)); }  private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {         // 如果有設置鎖的等待時長的話，就直接調用tryLockInnerAsync方法獲取鎖         if (leaseTime != -1) {             return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);         }         // 沒有設置等待鎖的時長的話，加多一個監聽器，也就是調用lock.lock()會跑的邏輯，后面會說         RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);         ttlRemainingFuture.addListener(new FutureListener<Long>() {             @Override             public void operationComplete(Future<Long> future) throws Exception {                 if (!future.isSuccess()) {                     return;                 }                  Long ttlRemaining = future.getNow();                 // lock acquired                 if (ttlRemaining == null) {                     scheduleExpirationRenewal(threadId);                 }             }         });         return ttlRemainingFuture;     }

我們繼續跟，看看tryLockInnerAsync方法的源碼：

<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {     internalLockLeaseTime = unit.toMillis(leaseTime);      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)); } String getLockName(long threadId) {     return id + ":" + threadId; }

這里就是底層的調用棧了，直接操作命令，整合成lua腳本后，調用netty的工具類跟redis進行通信，從而實現獲取鎖的功能。

這段腳本命令還是有點意思的，簡單解讀一下：

• 先用exists  key命令判斷是否鎖是否被占據了，沒有的話就用hset命令寫入，key為鎖的名稱，field為“客戶端唯一ID:線程ID”，value為1;

• 鎖被占據了，判斷是否是當前線程占據的，是的話value值加1;

• 鎖不是被當前線程占據，返回鎖剩下的過期時長;

命令的邏輯并不復雜，但不得不說，作者的設計還是很有心的，用了redis的Hash結構存儲數據，如果發現當前線程已經持有鎖了，就用hincrby命令將value值加1，value的值將決定釋放鎖的時候調用解鎖命令的次數，達到實現鎖的可重入性效果。

每一步命令對應的邏輯我都在下面的圖中標注了，大家可以讀一下：

我們繼續跟代碼吧，根據上面的命令可以看出，如果線程拿到鎖的話，tryLock方法會直接返回true，萬事大吉。

拿不到的話，就會返回鎖的剩余過期時長，這個時長有什么作用呢?我們回到tryLock方法中死循環的那個地方：

這里有一個針對waitTime和key的剩余過期時間大小的比較，取到二者中比較小的那個值，然后用Java的Semaphore信號量的tryAcquire方法來阻塞線程。

那么Semaphore信號量又是由誰控制呢，何時才能release呢。這里又需要回到上面來看，各位看官應該還記得，我們上面貼的tryLock代碼中還有這一段：

current = System.currentTimeMillis(); // 訂閱分布式鎖, 解鎖時進行通知 final RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);

訂閱的邏輯顯然是在subscribe方法里，跟著方法的調用鏈，它會進入到PublishSubscribe.Java中：

這段代碼的作用在于將當前線程的threadId添加到一個AsyncSemaphore中，并且設置一個redis的監聽器，這個監聽器是通過redis的發布、訂閱功能實現的。

一旦監聽器收到redis發來的消息，就從中獲取與當前thread相關的，如果是鎖被釋放的消息，就立馬通過操作Semaphore(也就是調用release方法)來讓剛才阻塞的地方釋放。

釋放后線程繼續執行，仍舊是判斷是否已經超時。如果還沒超時，就進入下一次循環再次去獲取鎖，拿到就返回true，沒有拿到的話就繼續流程。

這里說明一下，之所以要循環，是因為鎖可能會被多個客戶端同時爭搶，線程阻塞被釋放之后的那一瞬間很可能還是拿不到鎖，但是線程的等待時間又還沒過，這個時候就需要重新跑循環去拿鎖。

這就是tryLock獲取鎖的整個過程了，畫一張流程圖的話表示大概是這樣：

lock

除了tryLock，一般我們還經常直接調用lock來獲取鎖，lock的拿鎖過程跟tryLock基本是一致的，區別在于lock沒有手動設置鎖過期時長的參數，該方法的調用鏈也是跑到tryAcquire方法來獲取鎖的，不同的是，它會跑到這部分的邏輯：

這段代碼做了兩件事：

1、預設30秒的過期時長，然后去獲取鎖

2、開啟一個監聽器，如果發現拿到鎖了，就開啟定時任務不斷去刷新該鎖的過期時長

刷新過期時長的方法是scheduleExpirationRenewal，貼一下源碼吧：

private void scheduleExpirationRenewal(final long threadId) {  // expirationRenewalMap是一個ConcurrentMap，存儲標志為"當前線程ID:key名稱"的任務         if (expirationRenewalMap.containsKey(getEntryName())) {             return;         }          Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {             @Override             public void run(Timeout timeout) throws Exception {                 // 檢測鎖是否存在的lua腳本，存在的話就用pexpire命令刷新過期時長                 RFuture<Boolean> future = 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));                                  future.addListener(new FutureListener<Boolean>() {                     @Override                     public void operationComplete(Future<Boolean> future) throws Exception {                         expirationRenewalMap.remove(getEntryName());                         if (!future.isSuccess()) {                             log.error("Can't update lock " + getName() + " expiration", future.cause());                             return;                         }                                                  if (future.getNow()) {                             // reschedule itself                             scheduleExpirationRenewal(threadId);                         }                     }                 });             }         }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);          if (expirationRenewalMap.putIfAbsent(getEntryName(), task) != null) {             task.cancel();         }     }

代碼的流程比較簡單，大概就是開啟一個定時任務，每隔internalLockLeaseTime /  3的時間(這個時間是10秒)就去檢測鎖是否還被當前線程持有，是的話就重新設置過期時長internalLockLeaseTime，也就是30秒的時間。

而這些定時任務會存儲在一個ConcurrentHashMap對象expirationRenewalMap中，存儲的key就為“線程ID:key名稱”，如果發現expirationRenewalMap中不存在對應當前線程key的話，定時任務就不會跑，這也是后面解鎖中的一步重要操作。

上面這段代碼就是Redisson中所謂的”看門狗“程序，用一個異步線程來定時檢測并執行的，以防手動解鎖之前就過期了。

其他的邏輯就跟tryLock()基本沒什么兩樣啦，大家看一下就知道了

解鎖

有拿鎖的方法，自然也就有解鎖。Redisson分布式鎖解鎖的上層調用方法是unlock()，默認不用傳任何參數

@Override     public void unlock() {      // 發起釋放鎖的命令請求         Boolean opStatus = get(unlockInnerAsync(Thread.currentThread().getId()));         if (opStatus == null) {             throw new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "                     + id + " thread-id: " + Thread.currentThread().getId());         }         if (opStatus) {          // 成功釋放鎖，取消"看門狗"的續時線程             cancelExpirationRenewal();         }     }

解鎖相關的命令操作在unlockInnerAsync方法中定義，

又是一大串的lua腳本，比起前面加鎖那段腳本的命令稍微復雜了點，不過沒關系，我們簡單梳理一下，命令的邏輯大概是這么幾步：

1、判斷鎖是否存在，不存在的話用publish命令發布釋放鎖的消息，訂閱者收到后就能做下一步的拿鎖處理;

2、鎖存在但不是當前線程持有，返回空置nil;

3、當前線程持有鎖，用hincrby命令將鎖的可重入次數-1，然后判斷重入次數是否大于0，是的話就重新刷新鎖的過期時長，返回0，否則就刪除鎖，并發布釋放鎖的消息，返回1;

當線程完全釋放鎖后，就會調用cancelExpirationRenewal()方法取消"看門狗"的續時線程

void cancelExpirationRenewal() {  // expirationRenewalMap移除對應的key，就不會執行當前線程對應的"看門狗"程序了     Timeout task = expirationRenewalMap.remove(getEntryName());     if (task != null) {         task.cancel();     } }

這就是釋放鎖的過程了，怎么樣，是不是還是比較簡單的，閱讀起來比加鎖那份代碼舒服多了，當然啦，簡單歸簡單，為了方便你們理清整個分布式鎖的過程，我當然還是費心費力的給你們畫流程圖展示下啦(就沖這點，是不是該給我來個三連啊，哈哈)：

RedLock

以上就是Redisson分布式鎖的原理講解，總的來說，就是簡單的用lua腳本整合基本的set命令實現鎖的功能，這也是很多Redis分布式鎖工具的設計原理。除此之外，Redisson還支持用"RedLock算法"來實現鎖的效果，這個工具類就是RedissonRedLock。

用法也很簡單，創建多個Redisson  Node, 由這些無關聯的Node就可以組成一個完整的分布式鎖

RLock lock1 = Redisson.create(config1).getLock(lockKey); RLock lock2 = Redisson.create(config2).getLock(lockKey); RLock lock3 = Redisson.create(config3).getLock(lockKey);  RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3); try {    redLock.lock(); } finally {    redLock.unlock(); }

RedLock算法原理方面我就不細說了，大家有興趣可以看我之前的文章，或者是網上搜一下，簡單的說就是能一定程度上能有效防止Redis實例單點故障的問題，但并不完全可靠，不管是哪種設計，光靠Redis本身都是無法保證鎖的強一致性的。

還是那句話，魚和熊掌不可兼得，性能和安全方面也往往如此，Redis強大的性能和使用的方便足以滿足日常的分布式鎖需求，如果業務場景對鎖的安全隱患無法忍受的話，最保底的方式就是在業務層做冪等處理。

“如何理解Redisson分布式鎖的源碼”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識可以關注億速云網站，小編將為大家輸出更多高質量的實用文章！