<sub id="gqw76"><listing id="gqw76"></listing></sub>
      <sub id="gqw76"><listing id="gqw76"></listing></sub>

    1. <form id="gqw76"><legend id="gqw76"></legend></form>
    2. 又長又細,萬字長文帶你解讀Redisson分布式鎖的源碼

      前言

      上一篇文章寫了Redis分布式鎖的原理和缺陷,覺得有些不過癮,只是簡單的介紹了下Redisson這個框架,具體的原理什么的還沒說過呢。趁年前項目忙的差不多了,反正閑著也是閑著,不如把Redisson的源碼也學習一遍好了。

      雖說是一時興起,但仔細研究之后發現Redisson的源碼解讀工作量還是挺大的,其中用到了大量的Java并發類,并且引用了Netty作為通信工具,實現與Redis組件的遠程調用,這些知識點如果要全部講解的話不太現實,本文的重點主要是關于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的加鎖方法有兩個,tryLocklock,使用上的區別在于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分布式鎖的設計也有了足夠的了解,當然啦,雖然是講解源碼,我們的主要精力還是放在分布式鎖的原理上,一些無關流程的代碼就沒有帶大家字斟酌句的解讀了,大家有興趣的話可以自己去閱讀看看,源碼中很多地方都展示了一些基礎并發工具和網絡通信的妙用之處,學習一下還是挺有收獲的。

      最后我還是想吐槽一下,Redisson的注釋是真的少啊。。。。。。

      如果您覺得文章有用的話,歡迎點個贊支持一下,這將是對我創作的最好鼓勵!

      作者:鄙人薛某,一個不拘于技術的互聯網人,喜歡用通俗易懂的語言來解構后端技術的知識點,想看更多精彩文章的可以關注我的公眾號,微信搜索【鄙人薛某】即可關注

      posted @ 2021-02-25 09:56  鄙人薛某  閱讀(1220)  評論(5編輯  收藏
      最新chease0ldman老人|无码亚洲人妻下载|大香蕉在线看好吊妞视频这里有精品www|亚洲色情综合网

        <sub id="gqw76"><listing id="gqw76"></listing></sub>
        <sub id="gqw76"><listing id="gqw76"></listing></sub>

      1. <form id="gqw76"><legend id="gqw76"></legend></form>