最強分布式鎖工具：Redisson
點擊關(guān)注公眾號，Java干貨及時送達
一、Redisson概述
什么是Redisson？
Redisson是一個在Redis的基礎(chǔ)上實現(xiàn)的Java駐內(nèi)存數(shù)據(jù)網(wǎng)格（In-Memory Data Grid）。它不僅提供了一系列的分布式的Java常用對象，還提供了許多分布式服務(wù)。
其中包括(BitSet, Set, Multimap, SortedSet, Map, List, Queue, BlockingQueue, Deque, BlockingDeque, Semaphore, Lock, AtomicLong, CountDownLatch, Publish / Subscribe, Bloom filter, Remote service, Spring cache, Executor service, Live Object service, Scheduler service) Redisson提供了使用Redis的最簡單和最便捷的方法。
Redisson的宗旨是促進使用者對Redis的關(guān)注分離（Separation of Concern），從而讓使用者能夠?qū)⒕Ω械胤旁谔幚順I(yè)務(wù)邏輯上。
一個基于Redis實現(xiàn)的分布式工具，有基本分布式對象和高級又抽象的分布式服務(wù)，為每個試圖再造分布式輪子的程序員帶來了大部分分布式問題的解決辦法。
Redisson和Jedis、Lettuce有什么區(qū)別？倒也不是雷鋒和雷鋒塔
Redisson和它倆的區(qū)別就像一個用鼠標操作圖形化界面，一個用命令行操作文件。Redisson是更高層的抽象，Jedis和Lettuce是Redis命令的封裝。
Jedis是Redis官方推出的用于通過Java連接Redis客戶端的一個工具包，提供了Redis的各種命令支持
Lettuce是一種可擴展的線程安全的 Redis 客戶端，通訊框架基于Netty，支持高級的 Redis 特性，比如哨兵，集群，管道，自動重新連接和Redis數(shù)據(jù)模型。Spring Boot 2.x 開始 Lettuce 已取代 Jedis 成為首選 Redis 的客戶端。
Redisson是架設(shè)在Redis基礎(chǔ)上，通訊基于Netty的綜合的、新型的中間件，企業(yè)級開發(fā)中使用Redis的最佳范本
Jedis把Redis命令封裝好，Lettuce則進一步有了更豐富的Api，也支持集群等模式。但是兩者也都點到為止，只給了你操作Redis數(shù)據(jù)庫的腳手架，而Redisson則是基于Redis、Lua和Netty建立起了成熟的分布式解決方案，甚至redis官方都推薦的一種工具集。
二、分布式鎖
分布式鎖怎么實現(xiàn)？
分布式鎖是并發(fā)業(yè)務(wù)下的剛需，雖然實現(xiàn)五花八門：ZooKeeper有Znode順序節(jié)點，數(shù)據(jù)庫有表級鎖和樂/悲觀鎖，Redis有setNx，但是殊途同歸，最終還是要回到互斥上來，本篇介紹Redisson，那就以redis為例。
怎么寫一個簡單的Redis分布式鎖？
以Spring Data Redis為例，用RedisTemplate來操作Redis（setIfAbsent已經(jīng)是setNx + expire的合并命令），如下
// 加鎖
public Boolean tryLock(String key, String value, long timeout, TimeUnit unit) {
return redisTemplate.opsForValue().setIfAbsent(key, value, timeout, unit);
}
// 解鎖，防止刪錯別人的鎖，以uuid為value校驗是否自己的鎖
public void unlock(String lockName, String uuid) {
if(uuid.equals(redisTemplate.opsForValue().get(lockName)){ redisTemplate.opsForValue().del(lockName); }
}
// 結(jié)構(gòu)
if(tryLock){
// todo
}finally{
unlock;
}
簡單1.0版本完成，聰明的小張一眼看出，這是鎖沒錯，但get和del操作非原子性，并發(fā)一旦大了，無法保證進程安全。于是小張?zhí)嶙h，用Lua腳本
Lua腳本是什么？
Lua腳本是redis已經(jīng)內(nèi)置的一種輕量小巧語言，其執(zhí)行是通過redis的eval/evalsha命令來運行，把操作封裝成一個Lua腳本，如論如何都是一次執(zhí)行的原子操作。
于是2.0版本通過Lua腳本刪除
lockDel.lua如下
if redis.call('get', KEYS[1]) == ARGV[1]
then
-- 執(zhí)行刪除操作
return redis.call('del', KEYS[1])
else
-- 不成功，返回0
return 0
end
delete操作時執(zhí)行Lua命令
// 解鎖腳本
DefaultRedisScript<Object> unlockScript = new DefaultRedisScript();
unlockScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("lockDel.lua")));
// 執(zhí)行l(wèi)ua腳本解鎖
redisTemplate.execute(unlockScript, Collections.singletonList(keyName), value);
2.0似乎更像一把鎖，但好像又缺少了什么，小張一拍腦袋，synchronized和ReentrantLock都很絲滑，因為他們都是可重入鎖，一個線程多次拿鎖也不會死鎖，我們需要可重入。
怎么保證可重入？
重入就是，同一個線程多次獲取同一把鎖是允許的，不會造成死鎖，這一點synchronized偏向鎖提供了很好的思路，synchronized的實現(xiàn)重入是在JVM層面，JAVA對象頭MARK WORD中便藏有線程ID和計數(shù)器來對當(dāng)前線程做重入判斷，避免每次CAS。
當(dāng)一個線程訪問同步塊并獲取鎖時，會在對象頭和棧幀中的鎖記錄里存儲偏向的線程ID，以后該線程在進入和退出同步塊時不需要進行CAS操作來加鎖和解鎖，只需簡單測試一下對象頭的Mark Word里是否存儲著指向當(dāng)前線程的偏向鎖。如果測試成功，表示線程已經(jīng)獲得了鎖。如果測試失敗，則需要再測試一下Mark Word中偏向鎖標志是否設(shè)置成1：沒有則CAS競爭；設(shè)置了，則CAS將對象頭偏向鎖指向當(dāng)前線程。
再維護一個計數(shù)器，同個線程進入則自增1，離開再減1，直到為0才能釋放
可重入鎖
仿造該方案，我們需改造Lua腳本：
1.需要存儲 鎖名稱lockName、獲得該鎖的線程id和對應(yīng)線程的進入次數(shù)count
2.加鎖
每次線程獲取鎖時，判斷是否已存在該鎖
不存在
設(shè)置hash的key為線程id，value初始化為1
設(shè)置過期時間
返回獲取鎖成功true
存在
繼續(xù)判斷是否存在當(dāng)前線程id的hash key
存在，線程key的value + 1，重入次數(shù)增加1，設(shè)置過期時間
不存在，返回加鎖失敗
3.解鎖
每次線程來解鎖時，判斷是否已存在該鎖
存在
是否有該線程的id的hash key，有則減1，無則返回解鎖失敗
減1后，判斷剩余count是否為0，為0則說明不再需要這把鎖，執(zhí)行del命令刪除
1.存儲結(jié)構(gòu)
為了方便維護這個對象，我們用Hash結(jié)構(gòu)來存儲這些字段。Redis的Hash類似Java的HashMap，適合存儲對象。
hset lockname1 threadId 1
設(shè)置一個名字為lockname1的hash結(jié)構(gòu)，該hash結(jié)構(gòu)key為threadId，值value為1
hget lockname1 threadId
獲取lockname1的threadId的值
存儲結(jié)構(gòu)為
lockname 鎖名稱
key1： threadId 唯一鍵，線程id
value1： count 計數(shù)器，記錄該線程獲取鎖的次數(shù)
redis中的結(jié)構(gòu)
2.計數(shù)器的加減
當(dāng)同一個線程獲取同一把鎖時，我們需要對對應(yīng)線程的計數(shù)器count做加減
判斷一個redis key是否存在，可以用exists，而判斷一個hash的key是否存在，可以用hexists
而redis也有hash自增的命令hincrby
每次自增1時 hincrby lockname1 threadId 1，自減1時 hincrby lockname1 threadId -1
3.解鎖的判斷
當(dāng)一把鎖不再被需要了，每次解鎖一次，count減1，直到為0時，執(zhí)行刪除
綜合上述的存儲結(jié)構(gòu)和判斷流程，加鎖和解鎖Lua如下
加鎖 lock.lua
local key = KEYS[1];
local threadId = ARGV[1];
local releaseTime = ARGV[2];
-- lockname不存在
if(redis.call('exists', key) == 0) then
redis.call('hset', key, threadId, '1');
redis.call('expire', key, releaseTime);
return 1;
end;
-- 當(dāng)前線程已id存在
if(redis.call('hexists', key, threadId) == 1) then
redis.call('hincrby', key, threadId, '1');
redis.call('expire', key, releaseTime);
return 1;
end;
return 0;
解鎖 unlock.lua
local key = KEYS[1];
local threadId = ARGV[1];
-- lockname、threadId不存在
if (redis.call('hexists', key, threadId) == 0) then
return nil;
end;
-- 計數(shù)器-1
local count = redis.call('hincrby', key, threadId, -1);
-- 刪除lock
if (count == 0) then
redis.call('del', key);
return nil;
end;
代碼
/**
* @description 原生redis實現(xiàn)分布式鎖
**/
@Getter
@Setter
public class RedisLock {
private RedisTemplate redisTemplate;
private DefaultRedisScript<Long> lockScript;
private DefaultRedisScript<Object> unlockScript;
public RedisLock(RedisTemplate redisTemplate) {
this.redisTemplate = redisTemplate;
// 加載加鎖的腳本
lockScript = new DefaultRedisScript<>();
this.lockScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("lock.lua")));
this.lockScript.setResultType(Long.class);
// 加載釋放鎖的腳本
unlockScript = new DefaultRedisScript<>();
this.unlockScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("unlock.lua")));
}
/**
* 獲取鎖
*/
public String tryLock(String lockName, long releaseTime) {
// 存入的線程信息的前綴
String key = UUID.randomUUID().toString();
// 執(zhí)行腳本
Long result = (Long) redisTemplate.execute(
lockScript,
Collections.singletonList(lockName),
key + Thread.currentThread().getId(),
releaseTime);
if (result != null && result.intValue() == 1) {
return key;
} else {
return null;
}
}
/**
* 解鎖
* @param lockName
* @param key
*/
public void unlock(String lockName, String key) {
redisTemplate.execute(unlockScript,
Collections.singletonList(lockName),
key + Thread.currentThread().getId()
);
}
}
至此已經(jīng)完成了一把分布式鎖，符合互斥、可重入、防死鎖的基本特點。
嚴謹?shù)男堄X得雖然當(dāng)個普通互斥鎖，已經(jīng)穩(wěn)穩(wěn)夠用，可是業(yè)務(wù)里總是又很多特殊情況的，比如A進程在獲取到鎖的時候，因業(yè)務(wù)操作時間太長，鎖釋放了但是業(yè)務(wù)還在執(zhí)行，而此刻B進程又可以正常拿到鎖做業(yè)務(wù)操作，兩個進程操作就會存在依舊有共享資源的問題。
而且如果負責(zé)儲存這個分布式鎖的Redis節(jié)點宕機以后，而且這個鎖正好處于鎖住的狀態(tài)時，這個鎖會出現(xiàn)鎖死的狀態(tài)。
小張不是杠精，因為庫存操作總有這樣那樣的特殊。
所以我們希望在這種情況時，可以延長鎖的releaseTime延遲釋放鎖來直到完成業(yè)務(wù)期望結(jié)果，這種不斷延長鎖過期時間來保證業(yè)務(wù)執(zhí)行完成的操作就是鎖續(xù)約。
讀寫分離也是常見，一個讀多寫少的業(yè)務(wù)為了性能，常常是有讀鎖和寫鎖的。
而此刻的擴展已經(jīng)超出了一把簡單輪子的復(fù)雜程度，光是處理續(xù)約，就夠小張喝一壺，何況在性能（鎖的最大等待時間）、優(yōu)雅（無效鎖申請）、重試（失敗重試機制）等方面還要下功夫研究。
在小張苦思冥想時，旁邊的小白湊過來看了看小張，很好奇，都2021年了，為什么不直接用redisson呢？
Redisson就有這把你要的鎖。
三、Redisson分布式鎖
號稱簡單的Redisson分布式鎖的使用姿勢是什么？
1.依賴
<!-- 原生，本章使用-->
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.13.6</version>
</dependency>
<!-- 另一種Spring集成starter，本章未使用 -->
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson-spring-boot-starter</artifactId>
<version>3.13.6</version>
</dependency>
2.配置
@Configuration
public class RedissionConfig {
@Value("${spring.redis.host}")
private String redisHost;
@Value("${spring.redis.password}")
private String password;
private int port = 6379;
@Bean
public RedissonClient getRedisson() {
Config config = new Config();
config.useSingleServer().
setAddress("redis://" + redisHost + ":" + port).
setPassword(password);
config.setCodec(new JsonJacksonCodec());
return Redisson.create(config);
}
}
3.啟用分布式鎖
@Resource
private RedissonClient redissonClient;
RLock rLock = redissonClient.getLock(lockName);
try {
boolean isLocked = rLock.tryLock(expireTime, TimeUnit.MILLISECONDS);
if (isLocked) {
// TODO
}
} catch (Exception e) {
rLock.unlock();
}
簡潔明了，只需要一個RLock，既然推薦Redisson，就往里面看看他是怎么實現(xiàn)的。
四、RLock
RLock是Redisson分布式鎖的最核心接口，繼承了concurrent包的Lock接口和自己的RLockAsync接口，RLockAsync的返回值都是RFuture，是Redisson執(zhí)行異步實現(xiàn)的核心邏輯，也是Netty發(fā)揮的主要陣地。
RLock如何加鎖？
從RLock進入，找到RedissonLock類，找到tryLock方法再遞進到干事的tryAcquireOnceAsync方法，這是加鎖的主要代碼（版本不一此處實現(xiàn)有差別，和最新3.15.x有一定出入，但是核心邏輯依然未變。此處以3.13.6為例）
private RFuture<Boolean> tryAcquireOnceAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
if (leaseTime != -1L) {
return this.tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
} else {
RFuture<Boolean> ttlRemainingFuture = this.tryLockInnerAsync(waitTime, this.commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
if (e == null) {
if (ttlRemaining) {
this.scheduleExpirationRenewal(threadId);
}
}
});
return ttlRemainingFuture;
}
}
此處出現(xiàn)leaseTime時間判斷的2個分支，實際上就是加鎖時是否設(shè)置過期時間，未設(shè)置過期時間（-1）時則會有watchDog的鎖續(xù)約（下文），一個注冊了加鎖事件的續(xù)約任務(wù)。我們先來看有過期時間tryLockInnerAsync部分，
evalWriteAsync是eval命令執(zhí)行l(wèi)ua的入口
<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[]{this.internalLockLeaseTime, this.getLockName(threadId)});
}
這里揭開真面目，eval命令執(zhí)行Lua腳本的地方，此處的Lua腳本展開
-- 不存在該key時
if (redis.call('exists', KEYS[1]) == 0) then
-- 新增該鎖并且hash中該線程id對應(yīng)的count置1
redis.call('hincrby', KEYS[1], ARGV[2], 1);
-- 設(shè)置過期時間
redis.call('pexpire', KEYS[1], ARGV[1]);
return nil;
end;
-- 存在該key 并且 hash中線程id的key也存在
if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then
-- 線程重入次數(shù)++
redis.call('hincrby', KEYS[1], ARGV[2], 1);
redis.call('pexpire', KEYS[1], ARGV[1]);
return nil;
end;
return redis.call('pttl', KEYS[1]);
和前面我們寫自定義的分布式鎖的腳本幾乎一致，看來redisson也是一樣的實現(xiàn)，具體參數(shù)分析：
// keyName
KEYS[1] = Collections.singletonList(this.getName())
// leaseTime
ARGV[1] = this.internalLockLeaseTime
// uuid+threadId組合的唯一值
ARGV[2] = this.getLockName(threadId)
總共3個參數(shù)完成了一段邏輯：
判斷該鎖是否已經(jīng)有對應(yīng)hash表存在，
? 沒有對應(yīng)的hash表：則set該hash表中一個entry的key為鎖名稱，value為1，之后設(shè)置該hash表失效時間為leaseTime
? 存在對應(yīng)的hash表：則將該lockName的value執(zhí)行+1操作，也就是計算進入次數(shù)，再設(shè)置失效時間leaseTime
? 最后返回這把鎖的ttl剩余時間
也和上述自定義鎖沒有區(qū)別
既然如此，那解鎖的步驟也肯定有對應(yīng)的-1操作，再看unlock方法，同樣查找方法名，一路到
protected RFuture<Boolean> unlockInnerAsync(long threadId) {
return this.commandExecutor.evalWriteAsync(this.getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, "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;", Arrays.asList(this.getName(), this.getChannelName()), new Object[]{LockPubSub.unlockMessage, this.internalLockLeaseTime, this.getLockName(threadId)});
}
掏出Lua部分
-- 不存在key
if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then
return nil;
end;
-- 計數(shù)器 -1
local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1);
if (counter > 0) then
-- 過期時間重設(shè)
redis.call('pexpire', KEYS[1], ARGV[2]);
return 0;
else
-- 刪除并發(fā)布解鎖消息
redis.call('del', KEYS[1]);
redis.call('publish', KEYS[2], ARGV[1]);
return 1;
end;
return nil;
該Lua KEYS有2個Arrays.asList(getName(), getChannelName())
name 鎖名稱
channelName，用于pubSub發(fā)布消息的channel名稱
ARGV變量有三個LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId)
LockPubSub.UNLOCK_MESSAGE，channel發(fā)送消息的類別，此處解鎖為0
internalLockLeaseTime，watchDog配置的超時時間，默認為30s
lockName 這里的lockName指的是uuid和threadId組合的唯一值
步驟如下：
1.如果該鎖不存在則返回nil；
2.如果該鎖存在則將其線程的hash key計數(shù)器-1，
3.計數(shù)器counter>0，重置下失效時間，返回0；否則，刪除該鎖，發(fā)布解鎖消息unlockMessage，返回1；
其中unLock的時候使用到了Redis發(fā)布訂閱PubSub完成消息通知。
而訂閱的步驟就在RedissonLock的加鎖入口的lock方法里
long threadId = Thread.currentThread().getId();
Long ttl = this.tryAcquire(-1L, leaseTime, unit, threadId);
if (ttl != null) {
// 訂閱
RFuture<RedissonLockEntry> future = this.subscribe(threadId);
if (interruptibly) {
this.commandExecutor.syncSubscriptionInterrupted(future);
} else {
this.commandExecutor.syncSubscription(future);
}
// 省略
當(dāng)鎖被其他線程占用時，通過監(jiān)聽鎖的釋放通知（在其他線程通過RedissonLock釋放鎖時，會通過發(fā)布訂閱pub/sub功能發(fā)起通知），等待鎖被其他線程釋放，也是為了避免自旋的一種常用效率手段。
1.解鎖消息
為了一探究竟通知了什么，通知后又做了什么，進入LockPubSub。
這里只有一個明顯的監(jiān)聽方法onMessage，其訂閱和信號量的釋放都在父類PublishSubscribe，我們只關(guān)注監(jiān)聽事件的實際操作
protected void onMessage(RedissonLockEntry value, Long message) {
Runnable runnableToExecute;
if (message.equals(unlockMessage)) {
// 從監(jiān)聽器隊列取監(jiān)聽線程執(zhí)行監(jiān)聽回調(diào)
runnableToExecute = (Runnable)value.getListeners().poll();
if (runnableToExecute != null) {
runnableToExecute.run();
}
// getLatch()返回的是Semaphore，信號量，此處是釋放信號量
// 釋放信號量后會喚醒等待的entry.getLatch().tryAcquire去再次嘗試申請鎖
value.getLatch().release();
} else if (message.equals(readUnlockMessage)) {
while(true) {
runnableToExecute = (Runnable)value.getListeners().poll();
if (runnableToExecute == null) {
value.getLatch().release(value.getLatch().getQueueLength());
break;
}
runnableToExecute.run();
}
}
}
發(fā)現(xiàn)一個是默認解鎖消息，一個是讀鎖解鎖消息，因為redisson是有提供讀寫鎖的，而讀寫鎖讀讀情況和讀寫、寫寫情況互斥情況不同，我們只看上面的默認解鎖消息unlockMessage分支
LockPubSub監(jiān)聽最終執(zhí)行了2件事
runnableToExecute.run()執(zhí)行監(jiān)聽回調(diào)value.getLatch().release();釋放信號量
Redisson通過LockPubSub監(jiān)聽解鎖消息，執(zhí)行監(jiān)聽回調(diào)和釋放信號量通知等待線程可以重新?lián)屾i。
這時再回來看tryAcquireOnceAsync另一分支
private RFuture<Boolean> tryAcquireOnceAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
if (leaseTime != -1L) {
return this.tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
} else {
RFuture<Boolean> ttlRemainingFuture = this.tryLockInnerAsync(waitTime, this.commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
if (e == null) {
if (ttlRemaining) {
this.scheduleExpirationRenewal(threadId);
}
}
});
return ttlRemainingFuture;
}
}
可以看到，無超時時間時，在執(zhí)行加鎖操作后，還執(zhí)行了一段費解的邏輯
ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
if (e == null) {
if (ttlRemaining) {
this.scheduleExpirationRenewal(threadId);
}
}
}) } } }) 復(fù)制代碼
此處涉及到Netty的Future/Promise-Listener模型，Redisson中幾乎全部以這種方式通信（所以說Redisson是基于Netty通信機制實現(xiàn)的），理解這段邏輯可以試著先理解
在 Java 的 Future 中，業(yè)務(wù)邏輯為一個 Callable 或 Runnable 實現(xiàn)類，該類的 call()或 run()執(zhí)行完畢意味著業(yè)務(wù)邏輯的完結(jié)，在 Promise 機制中，可以在業(yè)務(wù)邏輯中人工設(shè)置業(yè)務(wù)邏輯的成功與失敗，這樣更加方便的監(jiān)控自己的業(yè)務(wù)邏輯。
這塊代碼的表面意義就是，在執(zhí)行異步加鎖的操作后，加鎖成功則根據(jù)加鎖完成返回的ttl是否過期來確認是否執(zhí)行一段定時任務(wù)。
這段定時任務(wù)的就是watchDog的核心。
2.鎖續(xù)約
查看RedissonLock.this.scheduleExpirationRenewal(threadId)
private void scheduleExpirationRenewal(long threadId) {
RedissonLock.ExpirationEntry entry = new RedissonLock.ExpirationEntry();
RedissonLock.ExpirationEntry oldEntry = (RedissonLock.ExpirationEntry)EXPIRATION_RENEWAL_MAP.putIfAbsent(this.getEntryName(), entry);
if (oldEntry != null) {
oldEntry.addThreadId(threadId);
} else {
entry.addThreadId(threadId);
this.renewExpiration();
}
}
private void renewExpiration() {
RedissonLock.ExpirationEntry ee = (RedissonLock.ExpirationEntry)EXPIRATION_RENEWAL_MAP.get(this.getEntryName());
if (ee != null) {
Timeout task = this.commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
public void run(Timeout timeout) throws Exception {
RedissonLock.ExpirationEntry ent = (RedissonLock.ExpirationEntry)RedissonLock.EXPIRATION_RENEWAL_MAP.get(RedissonLock.this.getEntryName());
if (ent != null) {
Long threadId = ent.getFirstThreadId();
if (threadId != null) {
RFuture<Boolean> future = RedissonLock.this.renewExpirationAsync(threadId);
future.onComplete((res, e) -> {
if (e != null) {
RedissonLock.log.error("Can't update lock " + RedissonLock.this.getName() + " expiration", e);
} else {
if (res) {
RedissonLock.this.renewExpiration();
}
}
});
}
}
}
}, this.internalLockLeaseTime / 3L, TimeUnit.MILLISECONDS);
ee.setTimeout(task);
}
}
拆分來看，這段連續(xù)嵌套且冗長的代碼實際上做了幾步
? 添加一個netty的Timeout回調(diào)任務(wù)，每（
internalLockLeaseTime / 3）毫秒執(zhí)行一次，執(zhí)行的方法是renewExpirationAsync?
renewExpirationAsync重置了鎖超時時間，又注冊一個監(jiān)聽器，監(jiān)聽回調(diào)又執(zhí)行了renewExpiration
renewExpirationAsync 的Lua如下
protected RFuture<Boolean> renewExpirationAsync(long threadId) {
return this.commandExecutor.evalWriteAsync(this.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.singletonList(this.getName()), new Object[]{this.internalLockLeaseTime, this.getLockName(threadId)});
}
if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then
redis.call('pexpire', KEYS[1], ARGV[1]);
return 1;
end;
return 0;
重新設(shè)置了超時時間。
Redisson加這段邏輯的目的是什么？
目的是為了某種場景下保證業(yè)務(wù)不影響，如任務(wù)執(zhí)行超時但未結(jié)束，鎖已經(jīng)釋放的問題。
當(dāng)一個線程持有了一把鎖，由于并未設(shè)置超時時間leaseTime，Redisson默認配置了30S，開啟watchDog，每10S對該鎖進行一次續(xù)約，維持30S的超時時間，直到任務(wù)完成再刪除鎖。
這就是Redisson的鎖續(xù)約，也就是WatchDog實現(xiàn)的基本思路。
3.流程概括
通過整體的介紹，流程簡單概括：
A、B線程爭搶一把鎖，A獲取到后，B阻塞
B線程阻塞時并非主動CAS，而是PubSub方式訂閱該鎖的廣播消息
A操作完成釋放了鎖，B線程收到訂閱消息通知
B被喚醒開始繼續(xù)搶鎖，拿到鎖
詳細加鎖解鎖流程總結(jié)如下圖：
五、公平鎖
以上介紹的可重入鎖是非公平鎖，Redisson還基于Redis的隊列（List）和ZSet實現(xiàn)了公平鎖
公平的定義是什么？
公平就是按照客戶端的請求先來后到排隊來獲取鎖，先到先得，也就是FIFO，所以隊列和容器順序編排必不可少
FairSync
回顧JUC的ReentrantLock公平鎖的實現(xiàn)
/**
* Sync object for fair locks
*/
static final class FairSync extends Sync {
private static final long serialVersionUID = -3000897897090466540L;
final void lock() {
acquire(1);
}
/**
* Fair version of tryAcquire. Don't grant access unless
* recursive call or no waiters or is first.
*/
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
if (!hasQueuedPredecessors() &&
compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
}
AQS已經(jīng)提供了整個實現(xiàn)，是否公平取決于實現(xiàn)類取出節(jié)點邏輯是否順序取
AbstractQueuedSynchronizer是用來構(gòu)建鎖或者其他同步組件的基礎(chǔ)框架，通過內(nèi)置FIFO隊列來完成資源獲取線程的排隊工作，他自身沒有實現(xiàn)同步接口，僅僅定義了若干同步狀態(tài)獲取和釋放的方法來供自定義同步組件使用（上圖），支持獨占和共享獲取，這是基于模版方法模式的一種設(shè)計，給公平/非公平提供了土壤。
我們用2張圖來簡單解釋AQS的等待流程（出自《JAVA并發(fā)編程的藝術(shù)》）
一張是同步隊列（FIFO雙向隊列）管理 獲取同步狀態(tài)失敗（搶鎖失敗）的線程引用、等待狀態(tài)和前驅(qū)后繼節(jié)點的流程圖
一張是獨占式獲取同步狀態(tài)的總流程，核心acquire(int arg)方法調(diào)用流程
可以看出鎖的獲取流程
AQS維護一個同步隊列，獲取狀態(tài)失敗的線程都會加入到隊列中進行自旋，移出隊列或停止自旋的條件是前驅(qū)節(jié)點為頭節(jié)點切成功獲取了同步狀態(tài)。
而比較另一段非公平鎖類NonfairSync可以發(fā)現(xiàn)，控制公平和非公平的關(guān)鍵代碼，在于hasQueuedPredecessors方法。
static final class NonfairSync extends Sync {
private static final long serialVersionUID = 7316153563782823691L;
/**
* Performs lock. Try immediate barge, backing up to normal
* acquire on failure.
*/
final void lock() {
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
}
NonfairSync減少了了hasQueuedPredecessors判斷條件，該方法的作用就是
查看同步隊列中當(dāng)前節(jié)點是否有前驅(qū)節(jié)點，如果有比當(dāng)前線程更早請求獲取鎖則返回true。
保證每次都取隊列的第一個節(jié)點（線程）來獲取鎖，這就是公平規(guī)則
為什么JUC以默認非公平鎖呢？
因為當(dāng)一個線程請求鎖時，只要獲取來同步狀態(tài)即成功獲取。在此前提下，剛釋放的線程再次獲取同步狀態(tài)的幾率會非常大，使得其他線程只能在同步隊列中等待。但這樣帶來的好處是，非公平鎖大大減少了系統(tǒng)線程上下文的切換開銷。
可見公平的代價是性能與吞吐量。
Redis里沒有AQS，但是有List和zSet，看看Redisson是怎么實現(xiàn)公平的。
RedissonFairLock
RedissonFairLock 用法依然很簡單
RLock fairLock = redissonClient.getFairLock(lockName);
fairLock.lock();
RedissonFairLock繼承自RedissonLock，同樣一路向下找到加鎖實現(xiàn)方法tryLockInnerAsync。
這里有2段冗長的Lua，但是Debug發(fā)現(xiàn)，公平鎖的入口在 command == RedisCommands.EVAL_LONG 之后，此段Lua較長，參數(shù)也多，我們著重分析Lua的實現(xiàn)規(guī)則
參數(shù)
-- lua中的幾個參數(shù)
KEYS = Arrays.<Object>asList(getName(), threadsQueueName, timeoutSetName)
KEYS[1]: lock_name, 鎖名稱
KEYS[2]: "redisson_lock_queue:{xxx}" 線程隊列
KEYS[3]: "redisson_lock_timeout:{xxx}" 線程id對應(yīng)的超時集合
ARGV = internalLockLeaseTime, getLockName(threadId), currentTime + threadWaitTime, currentTime
ARGV[1]: "{leaseTime}" 過期時間
ARGV[2]: "{Redisson.UUID}:{threadId}"
ARGV[3] = 當(dāng)前時間 + 線程等待時間:（10:00:00） + 5000毫秒 = 10:00:05
ARGV[4] = 當(dāng)前時間（10:00:00） 部署服務(wù)器時間，非redis-server服務(wù)器時間
公平鎖實現(xiàn)的Lua腳本
-- 1.死循環(huán)清除過期key
while true do
-- 獲取頭節(jié)點
local firstThreadId2 = redis.call('lindex', KEYS[2], 0);
-- 首次獲取必空跳出循環(huán)
if firstThreadId2 == false then
break;
end;
-- 清除過期key
local timeout = tonumber(redis.call('zscore', KEYS[3], firstThreadId2));
if timeout <= tonumber(ARGV[4]) then
redis.call('zrem', KEYS[3], firstThreadId2);
redis.call('lpop', KEYS[2]);
else
break;
end;
end;
-- 2.不存在該鎖 && （不存在線程等待隊列 || 存在線程等待隊列而且第一個節(jié)點就是此線程ID)，加鎖部分主要邏輯
if (redis.call('exists', KEYS[1]) == 0) and
((redis.call('exists', KEYS[2]) == 0) or (redis.call('lindex', KEYS[2], 0) == ARGV[2])) then
-- 彈出隊列中線程id元素，刪除Zset中該線程id對應(yīng)的元素
redis.call('lpop', KEYS[2]);
redis.call('zrem', KEYS[3], ARGV[2]);
local keys = redis.call('zrange', KEYS[3], 0, -1);
-- 遍歷zSet所有key，將key的超時時間(score) - 當(dāng)前時間ms
for i = 1, #keys, 1 do
redis.call('zincrby', KEYS[3], -tonumber(ARGV[3]), keys[i]);
end;
-- 加鎖設(shè)置鎖過期時間
redis.call('hset', KEYS[1], ARGV[2], 1);
redis.call('pexpire', KEYS[1], ARGV[1]);
return nil;
end;
-- 3.線程存在，重入判斷
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;
-- 4.返回當(dāng)前線程剩余存活時間
local timeout = redis.call('zscore', KEYS[3], ARGV[2]);
if timeout ~= false then
-- 過期時間timeout的值在下方設(shè)置，此處的減法算出的依舊是當(dāng)前線程的ttl
return timeout - tonumber(ARGV[3]) - tonumber(ARGV[4]);
end;
-- 5.尾節(jié)點剩余存活時間
local lastThreadId = redis.call('lindex', KEYS[2], -1);
local ttl;
-- 尾節(jié)點不空 && 尾節(jié)點非當(dāng)前線程
if lastThreadId ~= false and lastThreadId ~= ARGV[2] then
-- 計算隊尾節(jié)點剩余存活時間
ttl = tonumber(redis.call('zscore', KEYS[3], lastThreadId)) - tonumber(ARGV[4]);
else
-- 獲取lock_name剩余存活時間
ttl = redis.call('pttl', KEYS[1]);
end;
-- 6.末尾排隊
-- zSet 超時時間（score），尾節(jié)點ttl + 當(dāng)前時間 + 5000ms + 當(dāng)前時間，無則新增，有則更新
-- 線程id放入隊列尾部排隊，無則插入，有則不再插入
local timeout = ttl + tonumber(ARGV[3]) + tonumber(ARGV[4]);
if redis.call('zadd', KEYS[3], timeout, ARGV[2]) == 1 then
redis.call('rpush', KEYS[2], ARGV[2]);
end;
return ttl;
1.公平鎖加鎖步驟
通過以上Lua，可以發(fā)現(xiàn)，lua操作的關(guān)鍵結(jié)構(gòu)是列表（list）和有序集合（zSet）。
其中l(wèi)ist維護了一個等待的線程隊列redisson_lock_queue:{xxx}，zSet維護了一個線程超時情況的有序集合redisson_lock_timeout:{xxx}，盡管lua較長，但是可以拆分為6個步驟
1.隊列清理
保證隊列中只有未過期的等待線程
2.首次加鎖
hset加鎖，pexpire過期時間
3.重入判斷
此處同可重入鎖lua
4.返回ttl
5.計算尾節(jié)點ttl
初始值為鎖的剩余過期時間
6.末尾排隊
ttl + 2 * currentTime + waitTime是score的默認值計算公式
2.模擬
如果模擬以下順序，就會明了redisson公平鎖整個加鎖流程
假設(shè) t1 10:00:00 < t2 10:00:10 < t3 10:00:20
t1：當(dāng)線程1初次獲取鎖
1.等待隊列無頭節(jié)點，跳出死循環(huán)->2
2.不存在該鎖 && 不存在線程等待隊列 成立
2.1 lpop和zerm、zincrby都是無效操作，只有加鎖生效，說明是首次加鎖，加鎖后返回nil
加鎖成功，線程1獲取到鎖，結(jié)束
t2：線程2嘗試獲取鎖（線程1未釋放鎖）
1.等待隊列無頭節(jié)點，跳出死循環(huán)->2
2.不存在該鎖 不成立->3
3.非重入線程 ->4
4.score無值 ->5
5.尾節(jié)點為空，設(shè)置ttl初始值為lock_name的ttl -> 6
6.按照ttl + waitTime + currentTime + currentTime 來設(shè)置zSet超時時間score，并且加入等待隊列，線程2為頭節(jié)點
score = 20S + 5000ms + 10:00:10 + 10:00:10 = 10:00:35 + 10:00:10
t3：線程3嘗試獲取鎖（線程1未釋放鎖）
1.等待隊列有頭節(jié)點
1.1未過期->2
2.不存在該鎖不成立->3
3.非重入線程->4
4.score無值 ->5
5.尾節(jié)點不為空 && 尾節(jié)點線程為2，非當(dāng)前線程
5.1取出之前設(shè)置的score，減去當(dāng)前時間：ttl = score - currentTime ->6
6.按照ttl + waitTime + currentTime + currentTime 來設(shè)置zSet超時時間score，并且加入等待隊列
score = 10S + 5000ms + 10:00:20 + 10:00:20 = 10:00:35 + 10:00:20
如此一來，三個需要搶奪一把鎖的線程，完成了一次排隊，在list中排列他們等待線程id，在zSet中存放過期時間（便于排列優(yōu)先級）。其中返回ttl的線程2客戶端、線程3客戶端將會一直按一定間隔自旋重復(fù)執(zhí)行該段Lua，嘗試加鎖，如此一來便和AQS有了異曲同工之處。
而當(dāng)線程1釋放鎖之后（這里依舊有通過Pub/Sub發(fā)布解鎖消息，通知其他線程獲取）
10:00:30 線程2嘗試獲取鎖（線程1已釋放鎖）
1.等待隊列有頭節(jié)點，未過期->2
2.不存在該鎖 & 等待隊列頭節(jié)點是當(dāng)前線程 成立
2.1刪除當(dāng)前線程的隊列信息和zSet信息，超時時間為：
線程2 10:00:35 + 10:00:10 - 10:00:30 = 10:00:15
線程3 10:00:35 + 10:00:20 - 10:00:30 = 10:00:25
2.2線程2獲取到鎖，重新設(shè)置過期時間
加鎖成功，線程2獲取到鎖，結(jié)束
排隊結(jié)構(gòu)如圖
公平鎖的釋放腳本和重入鎖類似，多了一步加鎖開頭的清理過期key的while true邏輯，在此不再展開篇幅描述。
由上可以看出，Redisson公平鎖的玩法類似于延遲隊列的玩法，核心都在Redis的List和zSet結(jié)構(gòu)的搭配，但又借鑒了AQS實現(xiàn)，在定時判斷頭節(jié)點上如出一轍（watchDog），保證了鎖的競爭公平和互斥。并發(fā)場景下，lua腳本里，zSet的score很好地解決了順序插入的問題，排列好優(yōu)先級。
并且為了防止因異常而退出的線程無法清理，每次請求都會判斷頭節(jié)點的過期情況給予清理，最后釋放時通過CHANNEL通知訂閱線程可以來獲取鎖，重復(fù)一開始的步驟，順利交接到下一個順序線程。
六、總結(jié)
Redisson整體實現(xiàn)分布式加解鎖流程的實現(xiàn)稍顯復(fù)雜，作者Rui Gu對Netty和JUC、Redis研究深入，利用了很多高級特性和語義，值得深入學(xué)習(xí)，本次介紹也只是單機Redis下鎖實現(xiàn)。
Redisson也提供了多機情況下的聯(lián)鎖MultiLock：
https://github.com/redisson/redisson/wiki/8.-分布式鎖和同步器#81-可重入鎖reentrant-lock
和官方推薦的紅鎖RedLock：
https://github.com/redisson/redisson/wiki/8.-分布式鎖和同步器#84-紅鎖redlock
所以，當(dāng)你真的需要分布式鎖時，不妨先來Redisson里找找。
來源：juejin.cn/post/6961380552519712798
往 期 推 薦
3、互聯(lián)網(wǎng)人為什么學(xué)不會擺爛
4、為什么國外JetBrains做 IDE 就可以養(yǎng)活自己，國內(nèi)不行？區(qū)別在哪？
點分享
點收藏
點點贊
點在看
