Java redisson實(shí)現(xiàn)分布式鎖原理詳解

更新時(shí)間：2020年02月27日 13:21:31 作者：min.jiang

這篇文章主要介紹了Java redisson實(shí)現(xiàn)分布式鎖原理詳解,文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下

Redisson分布式鎖

之前的基于注解的鎖有一種鎖是基本redis的分布式鎖，鎖的實(shí)現(xiàn)我是基于redisson組件提供的RLock，這篇來(lái)看看redisson是如何實(shí)現(xiàn)鎖的。

不同版本實(shí)現(xiàn)鎖的機(jī)制并不相同

引用的redisson最近發(fā)布的版本3.2.3，不同的版本可能實(shí)現(xiàn)鎖的機(jī)制并不相同，早期版本好像是采用簡(jiǎn)單的setnx，getset等常規(guī)命令來(lái)配置完成，而后期由于redis支持了腳本Lua變更了實(shí)現(xiàn)原理。

<dependency>
	<groupId>org.redisson</groupId>
	<artifactId>redisson</artifactId>
	<version>3.2.3</version>
</dependency>

setnx需要配合getset以及事務(wù)來(lái)完成，這樣才能比較好的避免死鎖問(wèn)題，而新版本由于支持lua腳本，可以避免使用事務(wù)以及操作多個(gè)redis命令，語(yǔ)義表達(dá)更加清晰一些。

RLock接口的特點(diǎn)

繼承標(biāo)準(zhǔn)接口Lock

擁有標(biāo)準(zhǔn)鎖接口的所有特性，比如lock,unlock,trylock等等。

擴(kuò)展標(biāo)準(zhǔn)接口Lock

展了很多方法，常用的主要有：強(qiáng)制鎖釋放，帶有效期的鎖,還有一組異步的方法。其中前面兩個(gè)方法主要是解決標(biāo)準(zhǔn)lock可能造成的死鎖問(wèn)題。比如某個(gè)線程獲取到鎖之后，線程所在機(jī)器死機(jī)，此時(shí)獲取了鎖的線程無(wú)法正常釋放鎖導(dǎo)致其余的等待鎖的線程一直等待下去。

可重入機(jī)制

各版本實(shí)現(xiàn)有差異，可重入主要考慮的是性能，同一線程在未釋放鎖時(shí)如果再次申請(qǐng)鎖資源不需要走申請(qǐng)流程，只需要將已經(jīng)獲取的鎖繼續(xù)返回并且記錄上已經(jīng)重入的次數(shù)即可，與jdk里面的ReentrantLock功能類似。重入次數(shù)靠hincrby命令來(lái)配合使用，詳細(xì)的參數(shù)下面的代碼。

怎么判斷是同一線程？

redisson的方案是，RedissonLock實(shí)例的一個(gè)guid再加當(dāng)前線程的id,通過(guò)getLockName返回

public class RedissonLock extends RedissonExpirable implements RLock {
  
  final UUID id;
  protected RedissonLock(CommandExecutor commandExecutor, String name, UUID id) {
    super(commandExecutor, name);
    this.internalLockLeaseTime = TimeUnit.SECONDS.toMillis(30L);
    this.commandExecutor = commandExecutor;
    this.id = id;
  }

  String getLockName(long threadId) {
    return this.id + ":" + threadId;
  }

RLock獲取鎖的兩種場(chǎng)景

這里拿tryLock的源碼來(lái)看：tryAcquire方法是申請(qǐng)鎖并返回鎖有效期還剩余的時(shí)間，如果為空說(shuō)明鎖未被其它線程申請(qǐng)直接獲取并返回，如果獲取到時(shí)間，則進(jìn)入等待競(jìng)爭(zhēng)邏輯。

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 = this.tryAcquire(leaseTime, unit);
    if(ttl == null) {
      //直接獲取到鎖
      return true;
    } else {
      //有競(jìng)爭(zhēng)的后續(xù)看
    }
  }

無(wú)競(jìng)爭(zhēng)，直接獲取鎖

先看下首先獲取鎖并釋放鎖背后的redis都在做什么，可以利用redis的monitor來(lái)在后臺(tái)監(jiān)控redis的執(zhí)行情況。當(dāng)我們?cè)诜椒嗽黾覢RequestLockable之后，其實(shí)就是調(diào)用lock以及unlock，下面是redis命令：

加鎖

由于高版本的redis支持lua腳本，所以redisson也對(duì)其進(jìn)行了支持，采用了腳本模式，不熟悉lua腳本的可以去查找下。執(zhí)行l(wèi)ua命令的邏輯如下：

<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
    this.internalLockLeaseTime = unit.toMillis(leaseTime);
    return this.commandExecutor.evalWriteAsync(this.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.singletonList(this.getName()), new Object[]{Long.valueOf(this.internalLockLeaseTime), this.getLockName(threadId)});
  }

加鎖的流程：

判斷l(xiāng)ock鍵是否存在，不存在直接調(diào)用hset存儲(chǔ)當(dāng)前線程信息并且設(shè)置過(guò)期時(shí)間,返回nil，告訴客戶端直接獲取到鎖。
判斷l(xiāng)ock鍵是否存在，存在則將重入次數(shù)加1，并重新設(shè)置過(guò)期時(shí)間，返回nil，告訴客戶端直接獲取到鎖。
被其它線程已經(jīng)鎖定，返回鎖有效期的剩余時(shí)間，告訴客戶端需要等待。

"EVAL" 
"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]);"

 "1" "lock.com.csp.product.api.service.ProductAppService.searchProductByPage#0" 
 "1000" "346e1eb8-5bfd-4d49-9870-042df402f248:21"

上面的lua腳本會(huì)轉(zhuǎn)換成真正的redis命令，下面的是經(jīng)過(guò)lua腳本運(yùn)算之后實(shí)際執(zhí)行的redis命令。

1486642677.053488 [0 lua] "exists" "lock.com.csp.product.api.service.ProductAppService.searchProductByPage#0"
1486642677.053515 [0 lua] "hset" "lock.com.csp.product.api.service.ProductAppService.searchProductByPage#0" 
"346e1eb8-5bfd-4d49-9870-042df402f248:21" "1"
1486642677.053540 [0 lua] "pexpire" "lock.com.csp.product.api.service.ProductAppService.searchProductByPage#0" "1000"

解鎖

解鎖的流程看起來(lái)復(fù)雜些：

如果lock鍵不存在，發(fā)消息說(shuō)鎖已經(jīng)可用
如果鎖不是被當(dāng)前線程鎖定，則返回nil
由于支持可重入，在解鎖時(shí)將重入次數(shù)需要減1
如果計(jì)算后的重入次數(shù)>0，則重新設(shè)置過(guò)期時(shí)間
如果計(jì)算后的重入次數(shù)<=0，則發(fā)消息說(shuō)鎖已經(jīng)可用

"EVAL" 
"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;"
"2" "lock.com.csp.product.api.service.ProductAppService.searchProductByPage#0" 
"redisson_lock__channel:{lock.com.csp.product.api.service.ProductAppService.searchProductByPage#0}"
 "0" "1000"
 "346e1eb8-5bfd-4d49-9870-042df402f248:21"

無(wú)競(jìng)爭(zhēng)情況下解鎖redis命令：
主要是發(fā)送一個(gè)解鎖的消息，以此喚醒等待隊(duì)列中的線程重新競(jìng)爭(zhēng)鎖。

1486642678.493691 [0 lua] "exists" "lock.com.csp.product.api.service.ProductAppService.searchProductByPage#0"
1486642678.493712 [0 lua] "publish" "redisson_lock__channel:{lock.com.csp.product.api.service.ProductAppService.searchProductByPage#0}" "0"

有競(jìng)爭(zhēng)，等待

有競(jìng)爭(zhēng)的情況在redis端的lua腳本是相同的，只是不同的條件執(zhí)行不同的redis命令，復(fù)雜的在redisson的源碼上。當(dāng)通過(guò)tryAcquire發(fā)現(xiàn)鎖被其它線程申請(qǐng)時(shí)，需要進(jìn)入等待競(jìng)爭(zhēng)邏輯中。

this.await返回false，說(shuō)明等待時(shí)間已經(jīng)超出獲取鎖最大等待時(shí)間，取消訂閱并返回獲取鎖失敗
this.await返回true，進(jìn)入循環(huán)嘗試獲取鎖。

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 = this.tryAcquire(leaseTime, unit);
    if(ttl == null) {
      return true;
    } else {
      //重點(diǎn)是這段
      time -= System.currentTimeMillis() - current;
      if(time <= 0L) {
        return false;
      } else {
        current = System.currentTimeMillis();
        final RFuture subscribeFuture = this.subscribe(threadId);
        if(!this.await(subscribeFuture, time, TimeUnit.MILLISECONDS)) {
          if(!subscribeFuture.cancel(false)) {
            subscribeFuture.addListener(new FutureListener() {
              public void operationComplete(Future<RedissonLockEntry> future) throws Exception {
                if(subscribeFuture.isSuccess()) {
                  RedissonLock.this.unsubscribe(subscribeFuture, threadId);
                }

              }
            });
          }

          return false;
        } else {
          boolean var16;
          try {
            time -= System.currentTimeMillis() - current;
            if(time <= 0L) {
              boolean currentTime1 = false;
              return currentTime1;
            }

            do {
              long currentTime = System.currentTimeMillis();
              ttl = this.tryAcquire(leaseTime, unit);
              if(ttl == null) {
                var16 = true;
                return var16;
              }

              time -= System.currentTimeMillis() - currentTime;
              if(time <= 0L) {
                var16 = false;
                return var16;
              }

              currentTime = System.currentTimeMillis();
              if(ttl.longValue() >= 0L && ttl.longValue() < time) {
                this.getEntry(threadId).getLatch().tryAcquire(ttl.longValue(), TimeUnit.MILLISECONDS);
              } else {
                this.getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
              }

              time -= System.currentTimeMillis() - currentTime;
            } while(time > 0L);

            var16 = false;
          } finally {
            this.unsubscribe(subscribeFuture, threadId);
          }

          return var16;
        }
      }
    }
  }

循環(huán)嘗試一般有如下幾種方法：

while循環(huán)，一次接著一次的嘗試，這個(gè)方法的缺點(diǎn)是會(huì)造成大量無(wú)效的鎖申請(qǐng)。
Thread.sleep，在上面的while方案中增加睡眠時(shí)間以降低鎖申請(qǐng)次數(shù)，缺點(diǎn)是這個(gè)睡眠的時(shí)間設(shè)置比較難控制。
基于信息量，當(dāng)鎖被其它資源占用時(shí)，當(dāng)前線程訂閱鎖的釋放事件，一旦鎖釋放會(huì)發(fā)消息通知待等待的鎖進(jìn)行競(jìng)爭(zhēng)，有效的解決了無(wú)效的鎖申請(qǐng)情況。核心邏輯是this.getEntry(threadId).getLatch().tryAcquire，this.getEntry(threadId).getLatch()返回的是一個(gè)信號(hào)量，有興趣可以再研究研究。

redisson依賴

由于redisson不光是針對(duì)鎖，提供了很多客戶端操作redis的方法，所以會(huì)依賴一些其它的框架，比如netty，如果只是簡(jiǎn)單的使用鎖也可以自己去實(shí)現(xiàn)。以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: