日韩无码专区无码一级三级片|91人人爱网站中日韩无码电影|厨房大战丰满熟妇|AV高清无码在线免费观看|另类AV日韩少妇熟女|中文日本大黄一级黄色片|色情在线视频免费|亚洲成人特黄a片|黄片wwwav色图欧美|欧亚乱色一区二区三区

RELATEED CONSULTING
相關(guān)咨詢
選擇下列產(chǎn)品馬上在線溝通
服務(wù)時間：8:30-17:00
你可能遇到了下面的問題
關(guān)閉右側(cè)工具欄

新聞中心

這里有您想知道的互聯(lián)網(wǎng)營銷解決方案
又長又細，萬字長文帶你解讀Redisson分布式鎖的源碼

又長又細，萬字長文帶你解讀Redisson分布式鎖的源碼

作者：鄙人薛某 2021-02-19 10:42:58

開發(fā)

前端

分布式

Redis 上一篇文章寫了Redis分布式鎖的原理和缺陷，覺得有些不過癮，只是簡單的介紹了下Redisson這個框架，具體的原理什么的還沒說過呢。趁年后暫時沒什么事，反正閑著也是閑著，不如把Redisson的源碼也學(xué)習(xí)一遍好了。

泉州網(wǎng)站制作公司哪家好，找創(chuàng)新互聯(lián)公司！從網(wǎng)頁設(shè)計、網(wǎng)站建設(shè)、微信開發(fā)、APP開發(fā)、響應(yīng)式網(wǎng)站設(shè)計等網(wǎng)站項目制作，到程序開發(fā)，運營維護。創(chuàng)新互聯(lián)公司公司2013年成立到現(xiàn)在10年的時間，我們擁有了豐富的建站經(jīng)驗和運維經(jīng)驗，來保證我們的工作的順利進行。專注于網(wǎng)站建設(shè)就選創(chuàng)新互聯(lián)公司。

 前言

上一篇文章寫了Redis分布式鎖的原理和缺陷，覺得有些不過癮，只是簡單的介紹了下Redisson這個框架，具體的原理什么的還沒說過呢。趁年后暫時沒什么事，反正閑著也是閑著，不如把Redisson的源碼也學(xué)習(xí)一遍好了。

雖說是一時興起，但仔細研究之后發(fā)現(xiàn)Redisson的源碼解讀工作量還是挺大的，其中用到了大量的Java并發(fā)類，并且引用了Netty作為通信工具，實現(xiàn)與Redis組件的遠程調(diào)用，這些知識點如果要全部講解的話不太現(xiàn)實，本文的重點主要是關(guān)于Redisson分布式鎖的實現(xiàn)原理，所以網(wǎng)絡(luò)通信和并發(fā)原理這塊的代碼解讀不會太仔細，有不足之處還望見諒!

Redis 發(fā)布訂閱

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

在學(xué)習(xí)之前，我們有必要先了解一個知識點，就是有關(guān)Redis的發(fā)布訂閱功能。

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

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

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

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

Redisson源碼

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

  
 
 
 
    
   
  
  
  
  1. Config config = new Config(); 
    2. 
   
  
  
  
  2. config.useSingleServer() 
    3. 
   
  
  
  
  3.   .setPassword("") 
    4. 
   
  
  
  
  4.   .setAddress("redis://127.0.0.1:6379"); 
    5. 
   
  
  
  
  5. RedissonClient redisson = Redisson.create(config); 
    6. 
   
  
  
  
  6. // RLock對象 
    7. 
   
  
  
  
  7. RLock lock = redisson.getLock("myLock"); 
    8.

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

這里面初始化了幾個參數(shù)，

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

id：唯一ID，初始化的時候是用UUID創(chuàng)建的;

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

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

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

加鎖

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

tryLock

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

  
 
 
 
    
   
  
  
  
  1. /** 
    2. 
   
  
  
  
  2.  * @param waitTime 等待鎖的時長  
    3. 
   
  
  
  
  3.  * @param leaseTime 鎖的持有時間  
    4. 
   
  
  
  
  4.  * @param unit 時間單位 
    5. 
   
  
  
  
  5.  * @return 
    6. 
   
  
  
  
  6.  * @throws InterruptedException 
    7. 
   
  
  
  
  7.  */ 
    8. 
   
  
  
  
  8. public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {    // 剩余的等待鎖的時間 
    9. 
   
  
  
  
  9.         long time = unit.toMillis(waitTime); 
    10. 
   
  
  
  
  10.         long current = System.currentTimeMillis(); 
    11. 
   
  
  
  
  11.          
    12. 
   
  
  
  
  12.         final long threadId = Thread.currentThread().getId(); 
    13. 
   
  
  
  
  13.         // 嘗試獲取鎖，如果沒取到鎖，則返回鎖的剩余超時時間 
    14. 
   
  
  
  
  14.         Long ttl = tryAcquire(leaseTime, unit, threadId); 
    15. 
   
  
  
  
  15.         // ttl為null，說明可以搶到鎖了，返回true 
    16. 
   
  
  
  
  16.         if (ttl == null) { 
    17. 
   
  
  
  
  17.             return true; 
    18. 
   
  
  
  
  18.         } 
    19. 
   
  
  
  
  19.          
    20. 
   
  
  
  
  20.         // 如果waitTime已經(jīng)超時了，就返回false，代表申請鎖失敗 
    21. 
   
  
  
  
  21.         time -= (System.currentTimeMillis() - current); 
    22. 
   
  
  
  
  22.         if (time <= 0) { 
    23. 
   
  
  
  
  23.             acquireFailed(threadId); 
    24. 
   
  
  
  
  24.             return false; 
    25. 
   
  
  
  
  25.         } 
    26. 
   
  
  
  
  26.          
    27. 
   
  
  
  
  27.         current = System.currentTimeMillis(); 
    28. 
   
  
  
  
  28.         // 訂閱分布式鎖, 解鎖時進行通知，看，這里就用到了我們上面說的發(fā)布-訂閱了吧 
    29. 
   
  
  
  
  29.         final RFuture subscribeFuture = subscribe(threadId); 
    30. 
   
  
  
  
  30.         // 阻塞等待鎖釋放，await()返回false，說明等待超時了 
    31. 
   
  
  
  
  31.         if (!await(subscribeFuture, time, TimeUnit.MILLISECONDS)) { 
    32. 
   
  
  
  
  32.             if (!subscribeFuture.cancel(false)) { 
    33. 
   
  
  
  
  33.                 subscribeFuture.addListener(new FutureListener() { 
    34. 
   
  
  
  
  34.                     @Override 
    35. 
   
  
  
  
  35.                     public void operationComplete(Future future) throws Exception { 
    36. 
   
  
  
  
  36.                         if (subscribeFuture.isSuccess()) { 
    37. 
   
  
  
  
  37.                          // 等待都超時了，直接取消訂閱 
    38. 
   
  
  
  
  38.                             unsubscribe(subscribeFuture, threadId); 
    39. 
   
  
  
  
  39.                         } 
    40. 
   
  
  
  
  40.                     } 
    41. 
   
  
  
  
  41.                 }); 
    42. 
   
  
  
  
  42.             } 
    43. 
   
  
  
  
  43.             acquireFailed(threadId); 
    44. 
   
  
  
  
  44.             return false; 
    45. 
   
  
  
  
  45.         } 
    46. 
   
  
  
  
  46.  
    47. 
   
  
  
  
  47.         try { 
    48. 
   
  
  
  
  48.             time -= (System.currentTimeMillis() - current); 
    49. 
   
  
  
  
  49.             if (time <= 0) { 
    50. 
   
  
  
  
  50.                 acquireFailed(threadId); 
    51. 
   
  
  
  
  51.                 return false; 
    52. 
   
  
  
  
  52.             } 
    53. 
   
  
  
  
  53.          // 進入死循環(huán)，反復(fù)去調(diào)用tryAcquire嘗試獲取鎖，跟上面那一段拿鎖的邏輯一樣 
    54. 
   
  
  
  
  54.             while (true) { 
    55. 
   
  
  
  
  55.                 long currentTime = System.currentTimeMillis(); 
    56. 
   
  
  
  
  56.                 ttl = tryAcquire(leaseTime, unit, threadId); 
    57. 
   
  
  
  
  57.                 // lock acquired 
    58. 
   
  
  
  
  58.                 if (ttl == null) { 
    59. 
   
  
  
  
  59.                     return true; 
    60. 
   
  
  
  
  60.                 } 
    61. 
   
  
  
  
  61.  
    62. 
   
  
  
  
  62.                 time -= (System.currentTimeMillis() - currentTime); 
    63. 
   
  
  
  
  63.                 if (time <= 0) { 
    64. 
   
  
  
  
  64.                     acquireFailed(threadId); 
    65. 
   
  
  
  
  65.                     return false; 
    66. 
   
  
  
  
  66.                 } 
    67. 
   
  
  
  
  67.  
    68. 
   
  
  
  
  68.                 // waiting for message 
    69. 
   
  
  
  
  69.                 currentTime = System.currentTimeMillis(); 
    70. 
   
  
  
  
  70.                 if (ttl >= 0 && ttl < time) { 
    71. 
   
  
  
  
  71.                     getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS); 
    72. 
   
  
  
  
  72.                 } else { 
    73. 
   
  
  
  
  73.                     getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS); 
    74. 
   
  
  
  
  74.                 } 
    75. 
   
  
  
  
  75.  
    76. 
   
  
  
  
  76.                 time -= (System.currentTimeMillis() - currentTime); 
    77. 
   
  
  
  
  77.                 if (time <= 0) { 
    78. 
   
  
  
  
  78.                     acquireFailed(threadId); 
    79. 
   
  
  
  
  79.                     return false; 
    80. 
   
  
  
  
  80.                 } 
    81. 
   
  
  
  
  81.             } 
    82. 
   
  
  
  
  82.         } finally { 
    83. 
   
  
  
  
  83.             unsubscribe(subscribeFuture, threadId); 
    84. 
   
  
  
  
  84.         } 
    85. 
   
  
  
  
  85. //        return get(tryLockAsync(waitTime, leaseTime, unit)); 
    86. 
   
  
  
  
  86.     } 
    87.

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

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

  
 
 
 
    
   
  
  
  
  1. private Long tryAcquire(long leaseTime, TimeUnit unit, long threadId) { 
    2. 
   
  
  
  
  2.     return get(tryAcquireAsync(leaseTime, unit, threadId)); 
    3. 
   
  
  
  
  3. } 
    4. 
   
  
  
  
  4.  
    5. 
   
  
  
  
  5. private  RFuture tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) { 
    6. 
   
  
  
  
  6.         // 如果有設(shè)置鎖的等待時長的話，就直接調(diào)用tryLockInnerAsync方法獲取鎖 
    7. 
   
  
  
  
  7.         if (leaseTime != -1) { 
    8. 
   
  
  
  
  8.             return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG); 
    9. 
   
  
  
  
  9.         } 
    10. 
   
  
  
  
  10.         // 沒有設(shè)置等待鎖的時長的話，加多一個監(jiān)聽器，也就是調(diào)用lock.lock()會跑的邏輯，后面會說 
    11. 
   
  
  
  
  11.         RFuture ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG); 
    12. 
   
  
  
  
  12.         ttlRemainingFuture.addListener(new FutureListener() { 
    13. 
   
  
  
  
  13.             @Override 
    14. 
   
  
  
  
  14.             public void operationComplete(Future future) throws Exception { 
    15. 
   
  
  
  
  15.                 if (!future.isSuccess()) { 
    16. 
   
  
  
  
  16.                     return; 
    17. 
   
  
  
  
  17.                 } 
    18. 
   
  
  
  
  18.  
    19. 
   
  
  
  
  19.                 Long ttlRemaining = future.getNow(); 
    20. 
   
  
  
  
  20.                 // lock acquired 
    21. 
   
  
  
  
  21.                 if (ttlRemaining == null) { 
    22. 
   
  
  
  
  22.                     scheduleExpirationRenewal(threadId); 
    23. 
   
  
  
  
  23.                 } 
    24. 
   
  
  
  
  24.             } 
    25. 
   
  
  
  
  25.         }); 
    26. 
   
  
  
  
  26.         return ttlRemainingFuture; 
    27. 
   
  
  
  
  27.     } 
    28.

我們繼續(xù)跟，看看tryLockInnerAsync方法的源碼：

  
 
 
 
    
   
  
  
  
  1.  RFuture tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand command) { 
    2. 
   
  
  
  
  2.     internalLockLeaseTime = unit.toMillis(leaseTime); 
    3. 
   
  
  
  
  3.  
    4. 
   
  
  
  
  4.     return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command, 
    5. 
   
  
  
  
  5.               "if (redis.call('exists', KEYS[1]) == 0) then " + 
    6. 
   
  
  
  
  6.                   "redis.call('hset', KEYS[1], ARGV[2], 1); " + 
    7. 
   
  
  
  
  7.                   "redis.call('pexpire', KEYS[1], ARGV[1]); " + 
    8. 
   
  
  
  
  8.                   "return nil; " + 
    9. 
   
  
  
  
  9.               "end; " + 
    10. 
   
  
  
  
  10.               "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " + 
    11. 
   
  
  
  
  11.                   "redis.call('hincrby', KEYS[1], ARGV[2], 1); " + 
    12. 
   
  
  
  
  12.                   "redis.call('pexpire', KEYS[1], ARGV[1]); " + 
    13. 
   
  
  
  
  13.                   "return nil; " + 
    14. 
   
  
  
  
  14.               "end; " + 
    15. 
   
  
  
  
  15.               "return redis.call('pttl', KEYS[1]);", 
    16. 
   
  
  
  
  16.                 Collections.singletonList(getName()), internalLockLeaseTime, getLockName(threadId)); 
   
  
  
  
  17. } 
    18. 
   
  
  
  
  18. String getLockName(long threadId) { 
    19. 
   
  
  
  
  19.     return id + ":" + threadId; 
    20. 
   
  
  
  
  20. } 
    21. 
  
 
 
 
    這里就是底層的調(diào)用棧了，直接操作命令，整合成lua腳本后，調(diào)用netty的工具類跟redis進行通信，從而實現(xiàn)獲取鎖的功能。
    這段腳本命令還是有點意思的，簡單解讀一下：
      

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

    • 鎖被占據(jù)了，判斷是否是當(dāng)前線程占據(jù)的，是的話value值加1;
      • 

    • 鎖不是被當(dāng)前線程占據(jù)，返回鎖剩下的過期時長;
      • 

    

    命令的邏輯并不復(fù)雜，但不得不說，作者的設(shè)計還是很有心的，用了redis的Hash結(jié)構(gòu)存儲數(shù)據(jù)，如果發(fā)現(xiàn)當(dāng)前線程已經(jīng)持有鎖了，就用hincrby命令將value值加1，value的值將決定釋放鎖的時候調(diào)用解鎖命令的次數(shù)，達到實現(xiàn)鎖的可重入性效果。
    每一步命令對應(yīng)的邏輯我都在下面的圖中標注了，大家可以讀一下：
     
    我們繼續(xù)跟代碼吧，根據(jù)上面的命令可以看出，如果線程拿到鎖的話，tryLock方法會直接返回true，萬事大吉。
    拿不到的話，就會返回鎖的剩余過期時長，這個時長有什么作用呢?我們回到tryLock方法中死循環(huán)的那個地方：
     
    這里有一個針對waitTime和key的剩余過期時間大小的比較，取到二者中比較小的那個值，然后用Java的Semaphore信號量的tryAcquire方法來阻塞線程。
    那么Semaphore信號量又是由誰控制呢，何時才能release呢。這里又需要回到上面來看，各位看官應(yīng)該還記得，我們上面貼的tryLock代碼中還有這一段：
      
 
 
 
      
   
  
  
  
    1. current = System.currentTimeMillis(); 
      2. 
   
  
  
  
    2. // 訂閱分布式鎖, 解鎖時進行通知 
      3. 
   
  
  
  
    3. final RFuture subscribeFuture = subscribe(threadId); 
      4. 
  
 
 
 
    訂閱的邏輯顯然是在subscribe方法里，跟著方法的調(diào)用鏈，它會進入到PublishSubscribe.Java中：
     
    這段代碼的作用在于將當(dāng)前線程的threadId添加到一個AsyncSemaphore中，并且設(shè)置一個redis的監(jiān)聽器，這個監(jiān)聽器是通過redis的發(fā)布、訂閱功能實現(xiàn)的。
    一旦監(jiān)聽器收到redis發(fā)來的消息，就從中獲取與當(dāng)前thread相關(guān)的，如果是鎖被釋放的消息，就立馬通過操作Semaphore(也就是調(diào)用release方法)來讓剛才阻塞的地方釋放。
     
    釋放后線程繼續(xù)執(zhí)行，仍舊是判斷是否已經(jīng)超時。如果還沒超時，就進入下一次循環(huán)再次去獲取鎖，拿到就返回true，沒有拿到的話就繼續(xù)流程。
    這里說明一下，之所以要循環(huán)，是因為鎖可能會被多個客戶端同時爭搶，線程阻塞被釋放之后的那一瞬間很可能還是拿不到鎖，但是線程的等待時間又還沒過，這個時候就需要重新跑循環(huán)去拿鎖。
    這就是tryLock獲取鎖的整個過程了，畫一張流程圖的話表示大概是這樣：
     
    lock
    除了tryLock，一般我們還經(jīng)常直接調(diào)用lock來獲取鎖，lock的拿鎖過程跟tryLock基本是一致的，區(qū)別在于lock沒有手動設(shè)置鎖過期時長的參數(shù)，該方法的調(diào)用鏈也是跑到tryAcquire方法來獲取鎖的，不同的是，它會跑到這部分的邏輯：
     
    這段代碼做了兩件事：
    1、預(yù)設(shè)30秒的過期時長，然后去獲取鎖
    2、開啟一個監(jiān)聽器，如果發(fā)現(xiàn)拿到鎖了，就開啟定時任務(wù)不斷去刷新該鎖的過期時長
    刷新過期時長的方法是scheduleExpirationRenewal，貼一下源碼吧：
      
 
 
 
      
   
  
  
  
    1. private void scheduleExpirationRenewal(final long threadId) { 
      2. 
   
  
  
  
    2.  // expirationRenewalMap是一個ConcurrentMap，存儲標志為"當(dāng)前線程ID:key名稱"的任務(wù) 
      3. 
   
  
  
  
    3.         if (expirationRenewalMap.containsKey(getEntryName())) { 
      4. 
   
  
  
  
    4.             return; 
      5. 
   
  
  
  
    5.         } 
      6. 
   
  
  
  
    6.  
      7. 
   
  
  
  
    7.         Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() { 
      8. 
   
  
  
  
    8.             @Override 
      9. 
   
  
  
  
    9.             public void run(Timeout timeout) throws Exception { 
      10. 
   
  
  
  
    10.                 // 檢測鎖是否存在的lua腳本，存在的話就用pexpire命令刷新過期時長 
      11. 
   
  
  
  
    11.                 RFuture future = commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, 
      12. 
   
  
  
  
    12.                         "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " + 
      13. 
   
  
  
  
    13.                             "redis.call('pexpire', KEYS[1], ARGV[1]); " + 
      14. 
   
  
  
  
    14.                             "return 1; " + 
      15. 
   
  
  
  
    15.                         "end; " + 
      16. 
   
  
  
  
    16.                         "return 0;", 
      17. 
   
  
  
  
    17.                           Collections.singletonList(getName()), internalLockLeaseTime, getLockName(threadId)); 
   
  
  
  
    18.                  
      19. 
   
  
  
  
    19.                 future.addListener(new FutureListener() { 
      20. 
   
  
  
  
    20.                     @Override 
      21. 
   
  
  
  
    21.                     public void operationComplete(Future future) throws Exception { 
      22. 
   
  
  
  
    22.                         expirationRenewalMap.remove(getEntryName()); 
      23. 
   
  
  
  
    23.                         if (!future.isSuccess()) { 
      24. 
   
  
  
  
    24.                             log.error("Can't update lock " + getName() + " expiration", future.cause()); 
      25. 
   
  
  
  
    25.                             return; 
      26. 
   
  
  
  
    26.                         } 
      27. 
   
  
  
  
    27.                          
      28. 
   
  
  
  
    28.                         if (future.getNow()) { 
      29. 
   
  
  
  
    29.                             // reschedule itself 
      30. 
   
  
  
  
    30.                             scheduleExpirationRenewal(threadId); 
      31. 
   
  
  
  
    31.                         } 
      32. 
   
  
  
  
    32.                     } 
      33. 
   
  
  
  
    33.                 }); 
      34. 
   
  
  
  
    34.             } 
      35. 
   
  
  
  
    35.         }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS); 
      36. 
   
  
  
  
    36.  
      37. 
   
  
  
  
    37.         if (expirationRenewalMap.putIfAbsent(getEntryName(), task) != null) { 
      38. 
   
  
  
  
    38.             task.cancel(); 
      39. 
   
  
  
  
    39.         } 
      40. 
   
  
  
  
    40.     } 
      41. 
  
 
 
 
      代碼的流程比較簡單，大概就是開啟一個定時任務(wù)，每隔internalLockLeaseTime / 3的時間(這個時間是10秒)就去檢測鎖是否還被當(dāng)前線程持有，是的話就重新設(shè)置過期時長internalLockLeaseTime，也就是30秒的時間。
      而這些定時任務(wù)會存儲在一個ConcurrentHashMap對象expirationRenewalMap中，存儲的key就為“線程ID:key名稱”，如果發(fā)現(xiàn)expirationRenewalMap中不存在對應(yīng)當(dāng)前線程key的話，定時任務(wù)就不會跑，這也是后面解鎖中的一步重要操作。
      上面這段代碼就是Redisson中所謂的”看門狗“程序，用一個異步線程來定時檢測并執(zhí)行的，以防手動解鎖之前就過期了。
      其他的邏輯就跟tryLock()基本沒什么兩樣啦，大家看一下就知道了
      解鎖
      有拿鎖的方法，自然也就有解鎖。Redisson分布式鎖解鎖的上層調(diào)用方法是unlock()，默認不用傳任何參數(shù)
        
 
 
 
        
   
  
  
  
      1. @Override 
        2. 
   
  
  
  
      2.     public void unlock() { 
        3. 
   
  
  
  
      3.      // 發(fā)起釋放鎖的命令請求 
        4. 
   
  
  
  
      4.         Boolean opStatus = get(unlockInnerAsync(Thread.currentThread().getId())); 
        5. 
   
  
  
  
      5.         if (opStatus == null) { 
        6. 
   
  
  
  
      6.             throw new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: " 
        7. 
   
  
  
  
      7.                     + id + " thread-id: " + Thread.currentThread().getId()); 
        8. 
   
  
  
  
      8.         } 
        9. 
   
  
  
  
      9.         if (opStatus) { 
        10. 
   
  
  
  
      10.          // 成功釋放鎖，取消"看門狗"的續(xù)時線程 
        11. 
   
  
  
  
      11.             cancelExpirationRenewal(); 
        12. 
   
  
  
  
      12.         } 
        13. 
   
  
  
  
      13.     } 
        14. 
  
 
 
 
      解鎖相關(guān)的命令操作在unlockInnerAsync方法中定義，
       又是一大串的lua腳本，比起前面加鎖那段腳本的命令稍微復(fù)雜了點，不過沒關(guān)系，我們簡單梳理一下，命令的邏輯大概是這么幾步：
      1、判斷鎖是否存在，不存在的話用publish命令發(fā)布釋放鎖的消息，訂閱者收到后就能做下一步的拿鎖處理;
      2、鎖存在但不是當(dāng)前線程持有，返回空置nil;
      3、當(dāng)前線程持有鎖，用hincrby命令將鎖的可重入次數(shù)-1，然后判斷重入次數(shù)是否大于0，是的話就重新刷新鎖的過期時長，返回0，否則就刪除鎖，并發(fā)布釋放鎖的消息，返回1;
       當(dāng)線程完全釋放鎖后，就會調(diào)用cancelExpirationRenewal()方法取消"看門狗"的續(xù)時線程
        
 
 
 
        
   
  
  
  
      1. void cancelExpirationRenewal() { 
        2. 
   
  
  
  
      2.  // expirationRenewalMap移除對應(yīng)的key，就不會執(zhí)行當(dāng)前線程對應(yīng)的"看門狗"程序了 
        3. 
   
  
  
  
      3.     Timeout task = expirationRenewalMap.remove(getEntryName()); 
        4. 
   
  
  
  
      4.     if (task != null) { 
        5. 
   
  
  
  
      5.         task.cancel(); 
        6. 
   
  
  
  
      6.     } 
        7. 
   
  
  
  
      7. } 
        8. 
  
 
 
 
      這就是釋放鎖的過程了，怎么樣，是不是還是比較簡單的，閱讀起來比加鎖那份代碼舒服多了，當(dāng)然啦，簡單歸簡單，為了方便你們理清整個分布式鎖的過程，我當(dāng)然還是費心費力的給你們畫流程圖展示下啦(就沖這點，是不是該給我來個三連啊，哈哈)：
       
      RedLock
      以上就是Redisson分布式鎖的原理講解，總的來說，就是簡單的用lua腳本整合基本的set命令實現(xiàn)鎖的功能，這也是很多Redis分布式鎖工具的設(shè)計原理。除此之外，Redisson還支持用"RedLock算法"來實現(xiàn)鎖的效果，這個工具類就是RedissonRedLock。
       
      用法也很簡單，創(chuàng)建多個Redisson Node, 由這些無關(guān)聯(lián)的Node就可以組成一個完整的分布式鎖
        
 
 
 
        
   
  
  
  
      1. RLock lock1 = Redisson.create(config1).getLock(lockKey); 
        2. 
   
  
  
  
      2. RLock lock2 = Redisson.create(config2).getLock(lockKey); 
        3. 
   
  
  
  
      3. RLock lock3 = Redisson.create(config3).getLock(lockKey); 
        4. 
   
  
  
  
      4.  
        5. 
   
  
  
  
      5. RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3); 
        6. 
   
  
  
  
      6. try { 
        7. 
   
  
  
  
      7.    redLock.lock(); 
        8. 
   
  
  
  
      8. } finally { 
        9. 
   
  
  
  
      9.    redLock.unlock(); 
        10. 
   
  
  
  
      10. } 
        11. 
  
 
 
 
      RedLock算法原理方面我就不細說了，大家有興趣可以看我之前的文章，或者是網(wǎng)上搜一下，簡單的說就是能一定程度上能有效防止Redis實例單點故障的問題，但并不完全可靠，不管是哪種設(shè)計，光靠Redis本身都是無法保證鎖的強一致性的。
      還是那句話，魚和熊掌不可兼得，性能和安全方面也往往如此，Redis強大的性能和使用的方便足以滿足日常的分布式鎖需求，如果業(yè)務(wù)場景對鎖的安全隱患無法忍受的話，最保底的方式就是在業(yè)務(wù)層做冪等處理。
      總結(jié)
      看了本文的源碼解析，相信各位看官對Redisson分布式鎖的設(shè)計也有了足夠的了解，當(dāng)然啦，雖然是講解源碼，我們的主要精力還是放在分布式鎖的原理上，一些無關(guān)流程的代碼就沒有帶大家字斟酌句的解讀了，大家有興趣的話可以自己去閱讀看看，源碼中很多地方都展示了一些基礎(chǔ)并發(fā)工具和網(wǎng)絡(luò)通信的妙用之處，學(xué)習(xí)一下還是挺有收獲的。
      最后我還是想吐槽一下，Redisson的注釋是真的少啊。
                                                      
      
                                                標題名稱：又長又細，萬字長文帶你解讀Redisson分布式鎖的源碼                                                
      
                                                文章鏈接：http://www.5511xx.com/article/dpcsdpi.html