redisson實現分布式鎖原理

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

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


org.redissongroupid>
redissonartifactid>
3.2.3version>
dependency>

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

繼承標準介面lock

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

擴充套件標準介面lock

擴充套件了很多方法，常用的主要有：強制鎖釋放，帶有效期的鎖,還有一組非同步的方法。其中前面兩個方法主要是解決標準lock可能造成的死鎖問題。比如某個執行緒獲取到鎖之後，執行緒所在機器宕機，此時獲取了鎖的執行緒無法正常釋放鎖導致其餘的等待鎖的執行緒一直等待下去。

可重入機制

各版本實現有差異，可重入主要考慮的是效能，同一執行緒在未釋放鎖時如果再次申請鎖資源不需要走申請流程，只需要將已經獲取的鎖繼續返回並且記錄上已經重入的次數即可，與jdk裡面的reentrantlock功能類似。重入次數靠hincrby命令來配合使用，詳細的引數下面的**。

怎麼判斷是同一執行緒？

redisson的方案是，redissonlock例項的乙個guid再加當前執行緒的id,通過getlockname返回

public

class

redissonlock

extends

redissonexpirable

implements

rlock

string

getlockname(long

threadid)

這裡拿trylock的原始碼來看：tryacquire方法是申請鎖並返回鎖有效期還剩餘的時間，如果為空說明鎖未被其它執行緒申請直接獲取並返回，如果獲取到時間，則進入等待競爭邏輯。

public
boolean trylock(long waittime, long leasetime, timeunit unit) throws interruptedexception else 
}

無競爭，直接獲取鎖

先看下首先獲取鎖並釋放鎖背後的redis都在做什麼，可以利用redis的monitor來在後台監控redis的執**況。當我們在方法了增加@requestlockable之後，其實就是呼叫lock以及unlock，下面是redis命令：

<
t>
rfuture trylockinnerasync(long leasetime, timeunit unit, long threadid, redisstrictcommand command) );
}

加鎖的流程：

判斷lock鍵是否存在，不存在直接呼叫hset儲存當前執行緒資訊並且設定過期時間,返回nil，告訴客戶端直接獲取到鎖。

判斷lock鍵是否存在，存在則將重入次數加1，並重新設定過期時間，返回nil，告訴客戶端直接獲取到鎖。

被其它執行緒已經鎖定，返回鎖有效期的剩餘時間，告訴客戶端需要等待。

"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""1000"
"346e1eb8-5bfd-4d49-9870-042df402f248:21"

上面的lua指令碼會轉換成真正的redis命令，下面的是經過lua指令碼運算之後實際執行的redis命令。

1486642677.053488 [0 lua] "exists"
1486642677.053515 [0 lua] "hset"
"346e1eb8-5bfd-4d49-9870-042df402f248:21"
"1"1486642677.053540 [0 lua] "pexpire"
"1000"

如果lock鍵不存在，發訊息說鎖已經可用

如果鎖不是被當前執行緒鎖定，則返回nil

由於支援可重入，在解鎖時將重入次數需要減1

如果計算後的重入次數》0，則重新設定過期時間

如果計算後的重入次數<=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" 
"0""1000"
"346e1eb8-5bfd-4d49-9870-042df402f248:21"

無競爭情況下解鎖redis命令：

主要是傳送乙個解鎖的訊息，以此喚醒等待佇列中的執行緒重新競爭鎖。

1486642678.493691 [0 lua] "exists"
1486642678.493712 [0 lua] "publish"
"0"

有競爭，等待

有競爭的情況在redis端的lua指令碼是相同的，只是不同的條件執行不同的redis命令，複雜的在redisson的原始碼上。當通過tryacquire發現鎖被其它執行緒申請時，需要進入等待競爭邏輯中。

public
boolean trylock(long waittime, long leasetime, timeunit unit) throws interruptedexception else else 
}});
}return
false;
} else 
do time -=
system
.currenttimemillis() - currenttime;
if(time <=
0l) 
currenttime =
system
.currenttimemillis();
if(ttl.longvalue() >=
0l&& ttl.longvalue() < time) else 
time -=
system
.currenttimemillis() - currenttime;
} while(time >
0l);
var16 =
false;
} finally 
return var16;}}
}}

迴圈嘗試一般有如下幾種方法：

while迴圈，一次接著一次的嘗試，這個方法的缺點是會造成大量無效的鎖申請。

thread.sleep，在上面的while方案中增加睡眠時間以降低鎖申請次數，缺點是這個睡眠的時間設定比較難控制。

基於資訊量，當鎖被其它資源占用時，當前執行緒訂閱鎖的釋放事件，一旦鎖釋放會發訊息通知待等待的鎖進行競爭，有效的解決了無效的鎖申請情況。核心邏輯是this.getentry(threadid).getlatch().tryacquire，this.getentry(threadid).getlatch()返回的是乙個訊號量，有興趣可以再研究研究。

由於redisson不光是針對鎖，提供了很多客戶端操作redis的方法，所以會依賴一些其它的框架，比如netty，如果只是簡單的使用鎖也可以自己去實現。

redisson實現分布式鎖原理

Redisson實現分布式鎖

RedisSon實現分布式鎖

redisson實現分布式鎖

redisson實現分布式鎖原理

Redisson實現分布式鎖

RedisSon實現分布式鎖

redisson實現分布式鎖

相關推薦