前言
上一篇文章寫了Redis分布式鎖的原理和缺陷，覺得有些不過癮，只是簡單的介紹了下Redisson這個框架，具體的原理什么的還沒說過呢，趁年前專案忙的差不多了，反正閑著也是閑著，不如把Redisson的原始碼也學習一遍好了，
雖說是一時興起，但仔細研究之后發現Redisson的原始碼解讀作業量還是挺大的，其中用到了大量的Java并發類，并且參考了Netty作為通信工具，實作與Redis組件的遠程呼叫，這些知識點如果要全部講解的話不太現實，本文的重點主要是關于Redisson分布式鎖的實作原理，所以網路通信和并發原理這塊的代碼解讀不會太仔細，有不足之處還望見諒！
Redis 發布訂閱
之前說過，分布式鎖的核心功能其實就三個：加鎖、解鎖、設定鎖超時，這三個功能也是我們研究Redisson分布式鎖原理的方向，
在學習之前，我們有必要先了解一個知識點，就是有關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方法來獲取鎖的，不同的是，它會跑到這部分的邏輯：
這段代碼做了兩件事：
1、預設30秒的過期時長，然后去獲取鎖
2、開啟一個監聽器，如果發現拿到鎖了，就開啟定時任務不斷去重繪該鎖的過期時長
重繪過期時長的方法是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腳本，比起前面加鎖那段腳本的命令稍微復雜了點，不過沒關系，我們簡單梳理一下，命令的邏輯大概是這么幾步：
1、判斷鎖是否存在，不存在的話用publish命令發布釋放鎖的訊息，訂閱者收到后就能做下一步的拿鎖處理；
2、鎖存在但不是當前執行緒持有，回傳空置nil；
3、當前執行緒持有鎖，用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的注釋是真的少啊，，，，，，
如果您覺得文章有用的話，歡迎點個贊支持一下，這將是對我創作的最好鼓勵！
作者：鄙人薛某，一個不拘于技術的互聯網人，喜歡用通俗易懂的語言來解構后端技術的知識點，想看更多精彩文章的可以關注我的公眾號，微信搜索【鄙人薛某】即可關注
轉載請註明出處，本文鏈接：https://www.uj5u.com/ruanti/263692.html
標籤：其他
上一篇：又長又細，萬字長文帶你解讀Redisson分布式鎖的原始碼
下一篇：如何介紹自己測驗過的專案