詳解redisson實現分散式鎖定方法原理-java教程-PHP中文網

Redisson分散式鎖定

之前的基於註解的鎖定有一種鎖是基本redis的分散式鎖，鎖的實作我是基於redisson元件提供的RLock，這篇來看看redisson是如何實現鎖的。

不同版本實現鎖定的機制並不相同

引用的redisson最近發布的版本3.2.3，不同的版本可能實現鎖的機制並不相同，早期版本好像是採用簡單的setnx，getset等常規命令來配置完成，而後期由於redis支援了腳本Lua變更了實現原理。

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

setnx需要配合getset以及交易來完成，這樣才能比較好的避免死鎖問題，而新版本由於支援lua腳本，可以避免使用事務以及操作多個redis指令，語意表達更加清晰一些。

#RLock介面的特性##QySHTML5中文學習網- HTML5先行者學習網

繼承標準介面LockQySHTML5中文學習網- HTML5先行者學習網

擁有標準鎖定介面的所有特性，如lock,unlock,trylock等等。

QySHTML5中文學習網- HTML5先行者學習網

#擴充標準介面Lock##QySHTML5中文學習網- HTML5先行者學習網擴展了很多方法，常用的主要有：強制鎖釋放，帶有效期的鎖,還有一組非同步的方法。其中前面兩個方法主要是解決標準lock可能造成的死鎖問題。例如某個執行緒取得到鎖之後，執行緒所在機器死機，此時取得了鎖的執行緒無法正常釋放鎖導致其餘的等待鎖的執行緒一直等待下去。

QySHTML5中文學習網- HTML5先行者學習網

#可重入機制##QySHTML5中文學習網- HTML5先行者學習網QySHTML5中文學習網 - HTML5先行者學習網

#怎麼判斷是相同執行緒？

QySHTML5中文學習網- HTML5先行者學習網

redisson的方案是，RedissonLock實例的一個guid再加當前執行緒的id,透過getLockName返回

QySHTML5中文學習網- HTML5先行者學習網

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取得鎖定的兩個場景

QySHTML5中文學習網- HTML5先行者學習網QySHTML5中文學習網- HTML5先行者學習網

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 {   //有竞争的后续看  } }

無競爭，直接取得鎖定

#QySHTML5中文學習網- HTML5先行者學習網QySHTML5中文學習網- HTML5先行者學習網

#加鎖定

QySHTML5中文學習網- HTML5先行者學習網QySHTML5中文學習網 - HTML5先行者學習網

<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(/&#39;exists/&#39;, KEYS[1]) == 0) then redis.call(/&#39;hset/&#39;, KEYS[1], ARGV[2], 1); redis.call(/&#39;pexpire/&#39;, KEYS[1], ARGV[1]); return nil; end; if (redis.call(/&#39;hexists/&#39;, KEYS[1], ARGV[2]) == 1) then redis.call(/&#39;hincrby/&#39;, KEYS[1], ARGV[2], 1); redis.call(/&#39;pexpire/&#39;, KEYS[1], ARGV[1]); return nil; end; return redis.call(/&#39;pttl/&#39;, KEYS[1]);", Collections.singletonList(this.getName()), new Object[]{Long.valueOf(this.internalLockLeaseTime), this.getLockName(threadId)});  }

加锁的流程：

判断lock键是否存在，不存在直接调用hset存储当前线程信息并且设置过期时间,返回nil，告诉客户端直接获取到锁。
判断lock键是否存在，存在则将重入次数加1，并重新设置过期时间，返回nil，告诉客户端直接获取到锁。
被其它线程已经锁定，返回锁有效期的剩余时间，告诉客户端需要等待。

"EVAL" "if (redis.call(&#39;exists&#39;, KEYS[1]) == 0) then redis.call(&#39;hset&#39;, KEYS[1], ARGV[2], 1); redis.call(&#39;pexpire&#39;, KEYS[1], ARGV[1]); return nil; end;if (redis.call(&#39;hexists&#39;, KEYS[1], ARGV[2]) == 1) then redis.call(&#39;hincrby&#39;, KEYS[1], ARGV[2], 1); redis.call(&#39;pexpire&#39;, KEYS[1], ARGV[1]); return nil; end;return redis.call(&#39;pttl&#39;, KEYS[1]);" "1" "lock.com.csp.product.api.service.ProductAppService.searchProductByPage#0"  "1000" "346e1eb8-5bfd-4d49-9870-042df402f248:21"

上面的lua脚本会转换成真正的redis命令，下面的是经过lua脚本运算之后实际执行的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"

解锁

解锁的流程看起来复杂些：

如果lock键不存在，发消息说锁已经可用
如果锁不是被当前线程锁定，则返回nil
由于支持可重入，在解锁时将重入次数需要减1
如果计算后的重入次数>0，则重新设置过期时间
如果计算后的重入次数<=0，则发消息说锁已经可用

"EVAL" "if (redis.call(&#39;exists&#39;, KEYS[1]) == 0) then redis.call(&#39;publish&#39;, KEYS[2], ARGV[1]); return 1; end;if (redis.call(&#39;hexists&#39;, KEYS[1], ARGV[3]) == 0) then return nil;end; local counter = redis.call(&#39;hincrby&#39;, KEYS[1], ARGV[3], -1); if (counter > 0) then redis.call(&#39;pexpire&#39;, KEYS[1], ARGV[2]); return 0; else redis.call(&#39;del&#39;, KEYS[1]); redis.call(&#39;publish&#39;, 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"

无竞争情况下解锁redis命令：

主要是发送一个解锁的消息，以此唤醒等待队列中的线程重新竞争锁。

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"

有竞争，等待

有竞争的情况在redis端的lua脚本是相同的，只是不同的条件执行不同的redis命令，复杂的在redisson的源码上。当通过tryAcquire发现锁被其它线程申请时，需要进入等待竞争逻辑中。

this.await返回false，说明等待时间已经超出获取锁最大等待时间，取消订阅并返回获取锁失败
this.await返回true，进入循环尝试获取锁。

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 {      //重点是这段      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;        }      }    }  }

循环尝试一般有如下几种方法：

while循环，一次接着一次的尝试，这个方法的缺点是会造成大量无效的锁申请。
Thread.sleep，在上面的while方案中增加睡眠时间以降低锁申请次数，缺点是这个睡眠的时间设置比较难控制。
基于信息量，当锁被其它资源占用时，当前线程订阅锁的释放事件，一旦锁释放会发消息通知待等待的锁进行竞争，有效的解决了无效的锁申请情况。核心逻辑是this.getEntry(threadId).getLatch().tryAcquire，this.getEntry(threadId).getLatch()返回的是一个信号量，有兴趣可以再研究研究。

redisson依赖

由于redisson不光是针对锁，提供了很多客户端操作redis的方法，所以会依赖一些其它的框架，比如netty，如果只是简单的使用锁也可以自己去实现。

以上是詳解redisson實現分散式鎖定方法原理的詳細內容。更多資訊請關注PHP中文網其他相關文章！