redisson實現分佈式鎖原理

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接口的特色

繼承標準接口Lock

擁有標準鎖接口的全部特性，好比lock,unlock,trylock等等。

擴展標準接口Lock

擴展了不少方法，經常使用的主要有：強制鎖釋放，帶有效期的鎖,還有一組異步的方法。其中前面兩個方法主要是解決標準lock可能形成的死鎖問題。好比某個線程獲取到鎖以後，線程所在機器死機，此時獲取了鎖的線程沒法正常釋放鎖致使其他的等待鎖的線程一直等待下去。

可重入機制

各版本實現有差別，可重入主要考慮的是性能，同一線程在未釋放鎖時若是再次申請鎖資源不須要走申請流程，只須要將已經獲取的鎖繼續返回而且記錄上已經重入的次數便可，與jdk裏面的ReentrantLock功能相似。重入次數靠hincrby命令來配合使用，詳細的參數下面的代碼。

怎麼判斷是同一線程？
redisson的方案是，RedissonLock實例的一個guid再加當前線程的id,經過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獲取鎖的兩種場景

這裏拿tryLock的源碼來看：tryAcquire方法是申請鎖並返回鎖有效期還剩餘的時間，若是爲空說明鎖未被其它線程申請直接獲取並返回，若是獲取到時間，則進入等待競爭邏輯。

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 {
            //有競爭的後續看
        }
    }

無競爭，直接獲取鎖

先看下首先獲取鎖並釋放鎖背後的redis都在作什麼，能夠利用redis的monitor來在後臺監控redis的執行狀況。當咱們在方法了增長@RequestLockable以後，其實就是調用lock以及unlock，下面是redis命令：

• 加鎖
  因爲高版本的redis支持lua腳本，因此redisson也對其進行了支持，採用了腳本模式，不熟悉lua腳本的能夠去查找下。執行lua命令的邏輯以下：

<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)});
    }

加鎖的流程：

1. 判斷lock鍵是否存在，不存在直接調用hset存儲當前線程信息而且設置過時時間,返回nil，告訴客戶端直接獲取到鎖。
2. 判斷lock鍵是否存在，存在則將重入次數加1，並從新設置過時時間，返回nil，告訴客戶端直接獲取到鎖。
3. 被其它線程已經鎖定，返回鎖有效期的剩餘時間，告訴客戶端須要等待。

"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腳本會轉換成真正的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"

• 解鎖
  解鎖的流程看起來複雜些：

1. 若是lock鍵不存在，發消息說鎖已經可用
2. 若是鎖不是被當前線程鎖定，則返回nil
3. 因爲支持可重入，在解鎖時將重入次數須要減1
4. 若是計算後的重入次數>0，則從新設置過時時間
5. 若是計算後的重入次數<=0，則發消息說鎖已經可用

"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"

無競爭狀況下解鎖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;
                }
            }
        }
    }

循環嘗試通常有以下幾種方法：

1. while循環，一次接着一次的嘗試，這個方法的缺點是會形成大量無效的鎖申請。
2. Thread.sleep，在上面的while方案中增長睡眠時間以下降鎖申請次數，缺點是這個睡眠的時間設置比較難控制。
3. 基於信息量，當鎖被其它資源佔用時，當前線程訂閱鎖的釋放事件，一旦鎖釋放會發消息通知待等待的鎖進行競爭，有效的解決了無效的鎖申請狀況。核心邏輯是this.getEntry(threadId).getLatch().tryAcquire，this.getEntry(threadId).getLatch()返回的是一個信號量，有興趣能夠再研究研究。

redisson依賴

因爲redisson不光是針對鎖，提供了不少客戶端操做redis的方法，因此會依賴一些其它的框架，好比netty，若是隻是簡單的使用鎖也能夠本身去實現。