又長又細，萬字長文帶你解讀Redisson分佈式鎖的源碼
前言
上一篇文章寫了Redis分佈式鎖的原理和缺陷，以爲有些不過癮，只是簡單的介紹了下Redisson這個框架，具體的原理什麼的還沒說過呢。趁年前項目忙的差很少了，反正閒着也是閒着，不如把Redisson的源碼也學習一遍好了。node
雖然說是一時興起，但仔細研究以後發現Redisson的源碼解讀工做量仍是挺大的，其中用到了大量的Java併發類，而且引用了Netty做爲通訊工具，實現與Redis組件的遠程調用，這些知識點若是要所有講解的話不太現實，本文的重點主要是關於Redisson分佈式鎖的實現原理，因此網絡通訊和併發原理這塊的代碼解讀不會太仔細，有不足之處還望見諒！web
Redis 發佈訂閱
以前說過，分佈式鎖的核心功能其實就三個：加鎖、解鎖、設置鎖超時。這三個功能也是咱們研究Redisson分佈式鎖原理的方向。redis
在學習以前，咱們有必要先了解一個知識點，就是有關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的加鎖方法有兩個，tryLock和lock，使用上的區別在於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方法來獲取鎖的，不一樣的是，它會跑到這部分的邏輯：
這段代碼作了兩件事：
一、預設30秒的過時時長，而後去獲取鎖
二、開啓一個監聽器，若是發現拿到鎖了，就開啓定時任務不斷去刷新該鎖的過時時長
刷新過時時長的方法是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腳本，比起前面加鎖那段腳本的命令稍微複雜了點，不過不要緊，咱們簡單梳理一下，命令的邏輯大概是這麼幾步：
一、判斷鎖是否存在，不存在的話用publish命令發佈釋放鎖的消息，訂閱者收到後就能作下一步的拿鎖處理；
二、鎖存在但不是當前線程持有，返回空置nil；
三、當前線程持有鎖，用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的註釋是真的少啊。。。。。。
若是您以爲文章有用的話，歡迎點個贊支持一下，這將是對我創做的最好鼓勵！
做者：鄙人薛某，一個不拘於技術的互聯網人，喜歡用通俗易懂的語言來解構後端技術的知識點，想看更多精彩文章的能夠關注個人公衆號，微信搜索【鄙人薛某】便可關注
- 1. 帶你全面分析vue-router源碼（萬字長文）
- 2. 萬字長文帶你深度解鎖Webpack系列(進階篇)
- 3. 萬字長文從零解讀Vue3.0源碼響應式系統
- 4. 【GNN】萬字長文帶你入門 GCN
- 5. 萬字長文帶你入門Zookeeper！！！
- 6. 萬字長文：高併發分佈式鎖架構解密，不是全部的鎖都是分佈式鎖
- 7. 分佈式存儲-Redisson&分佈式鎖&源碼分析
- 8. 幹掉可惡的 "try catch "！（又一個萬字長文）
- 9. Redisson 分佈式鎖源碼——公平鎖加鎖
- 10. 【萬字長文】帶你瞭解日誌的前世此生
- 更多相關文章...
- • ARP報文格式詳解 - TCP/IP教程
- • UDP報文格式詳解 - TCP/IP教程
- • 再有人問你分佈式事務，把這篇扔給他
- • 常用的分佈式事務解決方案
-
每一个你不满意的现在，都有一个你没有努力的曾经。