時間不在於你擁有多少，而在於你怎樣使用。
1：Redisson 是什么
個人理解：一種 可重入、持續阻塞、獨占式的 分布式鎖協調框架，可從 ReentrantLock 去看它。
①：可重入鎖
拿到鎖的線程后續拿鎖可跳過獲取鎖的步驟，只進行value+1的步驟。
②：持續阻塞
獲取不到鎖的線程，會在一定時間內等待鎖。
日常開發中，應該都用過redis 的setnx 進行分布式的操作吧，那setnx 返回了false我們第一時間是不是就結束了？
因此redisson 優化了這個步驟，拿不到鎖會進行等待，直至timeout 。
同時它也支持 公平鎖 和 非公平鎖
③：互斥鎖
很好理解，同一環境下理論上只能有一個線程可以獲取到鎖。
對於redis 集群模式下，若master 的鎖還沒有同步給slave，這時 master 掛掉，然后哨兵選舉出新的master，
由於新的 master 並沒有同步到鎖，所以這個時候其他的線程仍然能獲取到鎖。因此獨占式在一定條件下是會失效的。
這個觀點在另外幾篇參考的文章中也有提到，個人也比較贊同，因此在此寫個筆記。
2：示例代碼
redisson的GitHub地址：https://github.com/redisson/redisson
我用的是boot-starter，配置參考官網給出的就行了。
測試代碼塊：
final String redissonLcokName = "redis-lock";
final RedissonLock redissonLock = (RedissonLock) redissonClient.getLock(redissonLcokName);
try {
redissonLock.lock(100, TimeUnit.SECONDS);
} catch (Exception ignore) {
} finally {
if (redissonLock.isLocked() && redissonLock.isHeldByCurrentThread())
redissonLock.unlock();
}
是不是和ReentrantLock 很像呢？
貼上我畫的草圖再講后面的內容：
3：如何獲取鎖
獲取鎖的操作采用lua腳本的形式，以保證指令的原子性。
從截圖上的序號來說步驟：
①：如果鎖不存在，則進行hincrby 操作（key不存在則value等於1，占鎖），並設置過期時間，然后返回nil。
②：如果鎖存在且 key 也存在，則進行hincrby操作（可重入鎖思想），並以毫秒為單位重新設置過期時間（續命），然后返回nil。
③：如果只存在鎖，key 不存在，則說明有其他線程獲取到了鎖（當前線程需要等待），需要返回鎖的過期時間。
從上述中就可以看出這個鎖是 hash 結構的：
而key的組成應該是：{uuid}:{threadid}
不信？我給你截圖...
①：RedissonBaseLock.getLockName(long threadId)
②：MasterSlaveConnectionManager.MasterSlaveConnectionManager(Config cfg, UUID id)
4：獲取鎖成功
4.1：看門狗
看門狗的存在是為了解決任務沒執行完，鎖就自動釋放了場景。
如默認鎖的釋放時間為30s，但是任務實際執行時間為35s，那么任務在執行到一半的時候鎖就被其他線程給搶占了，這明顯不符合需求。
因此就出現了看門狗，專門進行續命操作~~
private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
// 1：獲取到鎖則返回鎖的過期時間，否則返回null
RFuture<Long> ttlRemainingFuture;
if (leaseTime != -1) {
ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
} else {
ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime,
TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
}
// 2：任務完成之后執行
ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
if (e != null) {
return;
}
// lock acquired
if (ttlRemaining == null) {
if (leaseTime != -1) {
internalLockLeaseTime = unit.toMillis(leaseTime);
} else {
// 3:當鎖沒有預設置釋放時間才會調用看門狗線程
scheduleExpirationRenewal(threadId);
}
}
});
return ttlRemainingFuture;
}
通過分析底層代碼，當鎖沒有設置自動釋放時間才會啟用看門狗線程的。
所以我們要預設置過期時間的話最好還是先預估任務的實際執行時間再進行取值為妙...
4.2：時間輪
看門狗的操作實際上就是基於時間輪的。
①：RedissonBaseLock.renewExpiration()
在此處可以分析到看門狗的執行時間間隔：鎖的默認釋放時間為30s，因此每10s看門狗就會進行一次續命操作。
上述代碼底層點進去后可以看到實際上用了netty的 HashedWheelTimer 類：
②：MasterSlaveConnectionManager.newTimeout(TimerTask task, long delay, TimeUnit unit)
功力不夠，關於netty的細節就不過多描述了~~
借圖說下自己的理解
如上圖為一個時間輪模型，有8個齒輪，指針一秒走一次，那么走完需要8s。
齒輪有兩個屬性：
task：被執行的任務
bound：當bound = 0 時 task才會被執行，當bound > 0 時，指針每過一次bound - 1 直至為0 。
eg：如果你想31s后執行任務，那么bound應該等於3，齒輪處於第7個位置上。因為3*8+7=31。
4.3：解鎖->unlock
底層源碼：
①：若當前線程並沒有持有鎖，則返回nil。
②：當前線程持有鎖，則對value-1，拿到-1之后的vlaue。
③：value>0，以毫秒為單位返回剩下的過期時間。（保證可重入）
④：value<=0，則對key進行刪除操作，return 1 (方法返回 true)。然后進行redis-pub指令。
redis-pub 之后會被其他獲取不到鎖的線程給監聽到，其他線程又進入下一輪的占鎖操作。
5：獲取鎖失敗
5.1：關系類圖
這塊兒比較麻煩，先給一下比較重要的類圖吧...
5.2：訂閱事件
沒獲取到鎖線程后面在干嘛？當然要持續等待啦...
先在redis中發布訂閱消息，等待用完鎖的線程通知我~
看看訂閱主要干了些啥，從源碼上分析一波
①：PublishSubscribe.subscribe(String entryName, String channelName)源碼：
public RFuture<E> subscribe(String entryName, String channelName) {
AsyncSemaphore semaphore = service.getSemaphore(new ChannelName(channelName));
RPromise<E> newPromise = new RedissonPromise<>();
semaphore.acquire(() -> {
if (!newPromise.setUncancellable()) {
semaphore.release();
return;
}
// 1:判斷RedisLockEntry 是否存在
E entry = entries.get(entryName);
if (entry != null) {
entry.acquire();
semaphore.release();
entry.getPromise().onComplete(new TransferListener<E>(newPromise));
return;
}
// 2：創建RedisLockEntry
E value = createEntry(newPromise);
value.acquire();
E oldValue = entries.putIfAbsent(entryName, value);
if (oldValue != null) {
oldValue.acquire();
semaphore.release();
oldValue.getPromise().onComplete(new TransferListener<E>(newPromise));
return;
}
// 3:創建一個監聽器，別的線程進行redis-pub命令之后進行調用
RedisPubSubListener<Object> listener = createListener(channelName, value);
// 4:底層交給netty調用redis-sub命令
service.subscribe(LongCodec.INSTANCE, channelName, semaphore, listener);
});
return newPromise;
}
②：AsyncSemaphore.acquire(Runnable listener)源碼：
public class AsyncSemaphore {
private final AtomicInteger counter;
private final Queue<Runnable> listeners = new ConcurrentLinkedQueue<>();
...
public void acquire(Runnable listener) {
1：將任務假如到阻塞隊列
listeners.add(listener);
tryRun();
}
private void tryRun() {
if (counter.get() == 0
|| listeners.peek() == null) {
return;
}
// 2：counter>0時tryRun 才會取出linstener 中的任務進行執行
if (counter.decrementAndGet() >= 0) {
Runnable listener = listeners.poll();
if (listener == null) {
counter.incrementAndGet();
return;
}
if (removedListeners.remove(listener)) {
counter.incrementAndGet();
tryRun();
} else {
listener.run();
}
} else {
counter.incrementAndGet();
}
}
}
③：PubSubLock.createEntry() 源碼：
@Override
protected RedissonLockEntry createEntry(RPromise<RedissonLockEntry> newPromise) {
return new RedissonLockEntry(newPromise);
}
④：RedisLockEntry 的部分源碼：
public class RedissonLockEntry implements PubSubEntry<RedissonLockEntry> {
private int counter;
private final Semaphore latch;
private final RPromise<RedissonLockEntry> promise;
private final ConcurrentLinkedQueue<Runnable> listeners = new ConcurrentLinkedQueue<Runnable>();
public RedissonLockEntry(RPromise<RedissonLockEntry> promise) {
super();
this.latch = new Semaphore(0);
this.promise = promise;
}
...
public Semaphore getLatch() {
return latch;
}
}
⑤：RedisPubSubConnection.subscribe(Codec codec, ChannelName... channels)源碼：
public ChannelFuture subscribe(Codec codec, ChannelName... channels) {
for (ChannelName ch : channels) {
this.channels.put(ch, codec);
}
return async(new PubSubMessageDecoder(codec.getValueDecoder()), RedisCommands.SUBSCRIBE, channels);
}
...
private <T, R> ChannelFuture async(MultiDecoder<Object> messageDecoder, RedisCommand<T> command, Object... params) {
RPromise<R> promise = new RedissonPromise<R>();
// io.netty.Channel
return channel.writeAndFlush(new CommandData<T, R>(promise, messageDecoder, null, command, params));
}
給張圖可能看起來方便點：
訂閱源碼總結：
①：並不是每次每次都會創建RedisLockEntry，理論上是：當前應用內一個channel 對應一個RedisLockEntry 實例。
②：subscribe 的底層是基於netty進行操作的，並不是基於RedisTemplate。
③：不是每次subscribe都會執行到netty層，只有當屬於該redis-channel的RedisLockEntry 沒有實例化時才會調用到netty層。后續線程的只需要執行RedisLockEntry.acquire 操作即可。
6：redis-pub和redis-sub 是如何遙相呼應的？
6.1：Semaphore.tryAcquire(...)
RedisLockEntry 的latch屬性為Semaphore
我們看看RedisLock.lock() 源碼：
為什么要用while(true) ?
因為只有一個線程能拿到鎖啊，如果第一次拿到的ttl=1433ms，那么線程自旋1433ms就夠了，但是因為只能有一個線程拿到鎖，所以其他線程要進入下一輪的自旋。
紅線區域部分會導致當前線程阻塞。
而每次進行subscirbe后，RedisLockEntry.counter 值就會+1，counter值就代表多少線程正在等待獲取鎖。
6.2：Semaphore.release()
①：RedisPubSubConnection.onMessage(PubSubMessage message) 方法：
public void onMessage(PubSubMessage message) {
for (RedisPubSubListener<Object> redisPubSubListener : listeners) {
redisPubSubListener.onMessage(message.getChannel(), message.getValue());
}
}
會調用到下命這個方法 ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓
②：LockSubPub.onMessage(RedissonLockEntry value, Long message)方法：
@Override
protected void onMessage(RedissonLockEntry value, Long message) {
if (message.equals(UNLOCK_MESSAGE)) {
Runnable runnableToExecute = value.getListeners().poll();
if (runnableToExecute != null) {
runnableToExecute.run();
}
// value 就是線程進行subscribe 操作的時候所使用的RedisLockEntry 對象
// 終於執行release操作了，等待鎖的線程可以再次獲取鎖了
value.getLatch().release();
} else if (message.equals(READ_UNLOCK_MESSAGE)) {
while (true) {
Runnable runnableToExecute = value.getListeners().poll();
if (runnableToExecute == null) {
break;
}
runnableToExecute.run();
}
value.getLatch().release(value.getLatch().getQueueLength());
}
}
還記得我在上面提到過的嗎？下圖這個地方
再往下看就是步驟 LockSubPub.onMessage() 的代碼了。
源碼總結：
因為進行 redis-sub 之后，當前線程實際上會調用Semaphore.tryAcquire 方法去獲取鎖的，此處會導致線程自旋（阻塞）一定時間。
而當前線程在進行subscirbe之后因為會添加個 listner 在 RedisPubSubConnection.listeners（阻塞隊列中），這個listener 就是用來進行Semaphore.release 的。
當收到redis-pub命令時，先遍歷listeners，然后拿到事先傳給 linstener 的 RedisLockEntry 實例進行release 操作。
就這樣，釋放鎖-獲取鎖 就形成了遙相呼應。
看代碼和文字看累了直接看圖吧：
7：總結
實屬吐槽，個人認為redisson 並不是一個易於源碼閱讀的框架，看起來很費勁。
主要分為以下幾點:
1：調用鏈太長，參數一直往下傳遞
2：注釋稀少，很難初步看懂某個api具體的職責
3：各種繼承關系，寫代碼的人爽了，看代碼的人傻了...
第一批源碼文章，實屬不易。
最后，如文章有寫的不對的地方歡迎各位同學指正。
本文參考文章：
雨點的名字博客：https://www.cnblogs.com/qdhxhz/p/11046905.html
why神的文章：https://juejin.cn/post/6844904106461495303#heading-8