分布式鎖工具Redisson，太香了！！
文章來源：https://juejin.cn/post/6961380552519712798
一、Redisson概述
二、分布式鎖
三、Redisson分布式鎖
四、RLock
五、公平鎖
六、總結(jié)
一、Redisson概述
什么是Redisson？—— Redisson Wiki
Redisson是一個(gè)在Redis的基礎(chǔ)上實(shí)現(xiàn)的Java駐內(nèi)存數(shù)據(jù)網(wǎng)格（In-Memory Data Grid）。它不僅提供了一系列的分布式的Java常用對(duì)象，還提供了許多分布式服務(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的宗旨是促進(jìn)使用者對(duì)Redis的關(guān)注分離（Separation of Concern），從而讓使用者能夠?qū)⒕Ω械胤旁谔幚順I(yè)務(wù)邏輯上。
一個(gè)基于Redis實(shí)現(xiàn)的分布式工具，有基本分布式對(duì)象和高級(jí)又抽象的分布式服務(wù)，為每個(gè)試圖再造分布式輪子的程序員帶來了大部分分布式問題的解決辦法。
Redisson和Jedis、Lettuce有什么區(qū)別？倒也不是雷鋒和雷鋒塔
Redisson和它倆的區(qū)別就像一個(gè)用鼠標(biāo)操作圖形化界面，一個(gè)用命令行操作文件。Redisson是更高層的抽象，Jedis和Lettuce是Redis命令的封裝。
Jedis是Redis官方推出的用于通過Java連接Redis客戶端的一個(gè)工具包，提供了Redis的各種命令支持
Lettuce是一種可擴(kuò)展的線程安全的 Redis 客戶端，通訊框架基于Netty，支持高級(jí)的 Redis 特性，比如哨兵，集群，管道，自動(dòng)重新連接和Redis數(shù)據(jù)模型。Spring Boot 2.x 開始 Lettuce 已取代 Jedis 成為首選 Redis 的客戶端。
Redisson是架設(shè)在Redis基礎(chǔ)上，通訊基于Netty的綜合的、新型的中間件，企業(yè)級(jí)開發(fā)中使用Redis的最佳范本
Jedis把Redis命令封裝好，Lettuce則進(jìn)一步有了更豐富的Api，也支持集群等模式。但是兩者也都點(diǎn)到為止，只給了你操作Redis數(shù)據(jù)庫的腳手架，而Redisson則是基于Redis、Lua和Netty建立起了成熟的分布式解決方案，甚至redis官方都推薦的一種工具集。
二、分布式鎖
分布式鎖怎么實(shí)現(xiàn)？
分布式鎖是并發(fā)業(yè)務(wù)下的剛需，雖然實(shí)現(xiàn)五花八門：ZooKeeper有Znode順序節(jié)點(diǎn)，數(shù)據(jù)庫有表級(jí)鎖和樂/悲觀鎖，Redis有setNx，但是殊途同歸，最終還是要回到互斥上來，本篇介紹Redisson，那就以redis為例。
怎么寫一個(gè)簡單的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);
}
// 解鎖，防止刪錯(cuò)別人的鎖，以u(píng)uid為value校驗(yàn)是否自己的鎖
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版本完成，聰明的小張一眼看出，這是鎖沒錯(cuò)，但get和del操作非原子性，并發(fā)一旦大了，無法保證進(jìn)程安全。于是小張?zhí)嶙h，用Lua腳本
Lua腳本是什么？
Lua腳本是redis已經(jīng)內(nèi)置的一種輕量小巧語言，其執(zhí)行是通過redis的eval/evalsha命令來運(yùn)行，把操作封裝成一個(gè)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
enddelete操作時(shí)執(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都很絲滑，因?yàn)樗麄兌际强芍厝腈i，一個(gè)線程多次拿鎖也不會(huì)死鎖，我們需要可重入。
怎么保證可重入？
重入就是，同一個(gè)線程多次獲取同一把鎖是允許的，不會(huì)造成死鎖，這一點(diǎn)synchronized偏向鎖提供了很好的思路，synchronized的實(shí)現(xiàn)重入是在JVM層面，JAVA對(duì)象頭MARK WORD中便藏有線程ID和計(jì)數(shù)器來對(duì)當(dāng)前線程做重入判斷，避免每次CAS。
當(dāng)一個(gè)線程訪問同步塊并獲取鎖時(shí)，會(huì)在對(duì)象頭和棧幀中的鎖記錄里存儲(chǔ)偏向的線程ID，以后該線程在進(jìn)入和退出同步塊時(shí)不需要進(jìn)行CAS操作來加鎖和解鎖，只需簡單測試一下對(duì)象頭的Mark Word里是否存儲(chǔ)著指向當(dāng)前線程的偏向鎖。如果測試成功，表示線程已經(jīng)獲得了鎖。如果測試失敗，則需要再測試一下Mark Word中偏向鎖標(biāo)志是否設(shè)置成1：沒有則CAS競爭；設(shè)置了，則CAS將對(duì)象頭偏向鎖指向當(dāng)前線程。
再維護(hù)一個(gè)計(jì)數(shù)器，同個(gè)線程進(jìn)入則自增1，離開再減1，直到為0才能釋放
可重入鎖
仿造該方案，我們需改造Lua腳本：
需要存儲(chǔ) 鎖名稱lockName、獲得該鎖的線程id和對(duì)應(yīng)線程的進(jìn)入次數(shù)count
2.加鎖
每次線程獲取鎖時(shí)，判斷是否已存在該鎖
不存在
設(shè)置hash的key為線程id，value初始化為1
設(shè)置過期時(shí)間
返回獲取鎖成功true
存在
繼續(xù)判斷是否存在當(dāng)前線程id的hash key
存在，線程key的value + 1，重入次數(shù)增加1，設(shè)置過期時(shí)間
不存在，返回加鎖失敗
3.解鎖
每次線程來解鎖時(shí)，判斷是否已存在該鎖
存在
是否有該線程的id的hash key，有則減1，無則返回解鎖失敗
減1后，判斷剩余count是否為0，為0則說明不再需要這把鎖，執(zhí)行del命令刪除
1.存儲(chǔ)結(jié)構(gòu)
為了方便維護(hù)這個(gè)對(duì)象，我們用Hash結(jié)構(gòu)來存儲(chǔ)這些字段。Redis的Hash類似Java的HashMap，適合存儲(chǔ)對(duì)象。
設(shè)置一個(gè)名字為lockname1的hash結(jié)構(gòu)，該hash結(jié)構(gòu)key為threadId，值value為1
hget lockname1 threadId
獲取lockname1的threadId的值
存儲(chǔ)結(jié)構(gòu)為
lockname 鎖名稱
key1：threadId 唯一鍵，線程id
value1：count 計(jì)數(shù)器，記錄該線程獲取鎖的次數(shù)
redis中的結(jié)構(gòu)
2.計(jì)數(shù)器的加減
當(dāng)同一個(gè)線程獲取同一把鎖時(shí)，我們需要對(duì)對(duì)應(yīng)線程的計(jì)數(shù)器count做加減
判斷一個(gè)redis key是否存在，可以用exists，而判斷一個(gè)hash的key是否存在，可以用hexists
而redis也有hash自增的命令hincrby
每次自增1時(shí) hincrby lockname1 threadId 1，自減1時(shí) hincrby lockname1 threadId -1
3.解鎖的判斷
當(dāng)一把鎖不再被需要了，每次解鎖一次，count減1，直到為0時(shí)，執(zhí)行刪除
綜合上述的存儲(chǔ)結(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;
-- 計(jì)數(shù)器-1
local count = redis.call('hincrby', key, threadId, -1);
-- 刪除lock
if (count == 0) then
redis.call('del', key);
return nil;
end;代碼
/**
* @description 原生redis實(shí)現(xiàn)分布式鎖
* @date 2021/2/6 10:51 下午
**/
@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)完成了一把分布式鎖，符合互斥、可重入、防死鎖的基本特點(diǎn)。
嚴(yán)謹(jǐn)?shù)男堄X得雖然當(dāng)個(gè)普通互斥鎖，已經(jīng)穩(wěn)穩(wěn)夠用，可是業(yè)務(wù)里總是又很多特殊情況的，比如A進(jìn)程在獲取到鎖的時(shí)候，因業(yè)務(wù)操作時(shí)間太長，鎖釋放了但是業(yè)務(wù)還在執(zhí)行，而此刻B進(jìn)程又可以正常拿到鎖做業(yè)務(wù)操作，兩個(gè)進(jìn)程操作就會(huì)存在依舊有共享資源的問題。
而且如果負(fù)責(zé)儲(chǔ)存這個(gè)分布式鎖的Redis節(jié)點(diǎn)宕機(jī)以后，而且這個(gè)鎖正好處于鎖住的狀態(tài)時(shí)，這個(gè)鎖會(huì)出現(xiàn)鎖死的狀態(tài)。
小張不是杠精，因?yàn)閹齑娌僮骺傆羞@樣那樣的特殊。
所以我們希望在這種情況時(shí)，可以延長鎖的releaseTime延遲釋放鎖來直到完成業(yè)務(wù)期望結(jié)果，這種不斷延長鎖過期時(shí)間來保證業(yè)務(wù)執(zhí)行完成的操作就是鎖續(xù)約。
讀寫分離也是常見，一個(gè)讀多寫少的業(yè)務(wù)為了性能，常常是有讀鎖和寫鎖的。
而此刻的擴(kuò)展已經(jīng)超出了一把簡單輪子的復(fù)雜程度，光是處理續(xù)約，就夠小張喝一壺，何況在性能（鎖的最大等待時(shí)間）、優(yōu)雅（無效鎖申請(qǐng)）、重試（失敗重試機(jī)制）等方面還要下功夫研究。在小張苦思冥想時(shí)，旁邊的小白湊過來看了看小張，很好奇，都2021年了，為什么不直接用redisson呢？
Redisson就有這把你要的鎖。
三、Redisson分布式鎖
號(hào)稱簡單的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();
}簡潔明了，只需要一個(gè)RLock，既然推薦Redisson，就往里面看看他是怎么實(shí)現(xiàn)的。
四、RLock
RLock是Redisson分布式鎖的最核心接口，繼承了concurrent包的Lock接口和自己的RLockAsync接口，RLockAsync的返回值都是RFuture，是Redisson執(zhí)行異步實(shí)現(xiàn)的核心邏輯，也是Netty發(fā)揮的主要陣地。
RLock如何加鎖？
從RLock進(jìn)入，找到RedissonLock類，找到tryLock方法再遞進(jìn)到干事的tryAcquireOnceAsync方法，這是加鎖的主要代碼（版本不一此處實(shí)現(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時(shí)間判斷的2個(gè)分支，實(shí)際上就是加鎖時(shí)是否設(shè)置過期時(shí)間，未設(shè)置過期時(shí)間（-1）時(shí)則會(huì)有watchDog的鎖續(xù)約（下文），一個(gè)注冊了加鎖事件的續(xù)約任務(wù)。我們先來看有過期時(shí)間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時(shí)
if (redis.call('exists', KEYS[1]) == 0) then
-- 新增該鎖并且hash中該線程id對(duì)應(yīng)的count置1
redis.call('hincrby', KEYS[1], ARGV[2], 1);
-- 設(shè)置過期時(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也是一樣的實(shí)現(xiàn)，具體參數(shù)分析：
// keyName
KEYS[1] = Collections.singletonList(this.getName())
// leaseTime
ARGV[1] = this.internalLockLeaseTime
// uuid+threadId組合的唯一值
ARGV[2] = this.getLockName(threadId)
總共3個(gè)參數(shù)完成了一段邏輯：
判斷該鎖是否已經(jīng)有對(duì)應(yīng)hash表存在，
? 沒有對(duì)應(yīng)的hash表：則set該hash表中一個(gè)entry的key為鎖名稱，value為1，之后設(shè)置該hash表失效時(shí)間為leaseTime
? 存在對(duì)應(yīng)的hash表：則將該lockName的value執(zhí)行+1操作，也就是計(jì)算進(jìn)入次數(shù)，再設(shè)置失效時(shí)間leaseTime
? 最后返回這把鎖的ttl剩余時(shí)間
也和上述自定義鎖沒有區(qū)別
既然如此，那解鎖的步驟也肯定有對(duì)應(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;
-- 計(jì)數(shù)器 -1
local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1);
if (counter > 0) then
-- 過期時(shí)間重設(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個(gè)Arrays.asList(getName(), getChannelName())
name 鎖名稱
channelName，用于pubSub發(fā)布消息的channel名稱
ARGV變量有三個(gè)LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId)
LockPubSub.UNLOCK_MESSAGE，channel發(fā)送消息的類別，此處解鎖為0
internalLockLeaseTime，watchDog配置的超時(shí)時(shí)間，默認(rèn)為30s
lockName 這里的lockName指的是uuid和threadId組合的唯一值
步驟如下：
1.如果該鎖不存在則返回nil；
2.如果該鎖存在則將其線程的hash key計(jì)數(shù)器-1，
3.計(jì)數(shù)器counter>0，重置下失效時(shí)間，返回0；否則，刪除該鎖，發(fā)布解鎖消息unlockMessage，返回1；
其中unLock的時(shí)候使用到了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)鎖被其他線程占用時(shí)，通過監(jiān)聽鎖的釋放通知（在其他線程通過RedissonLock釋放鎖時(shí)，會(huì)通過發(fā)布訂閱pub/sub功能發(fā)起通知），等待鎖被其他線程釋放，也是為了避免自旋的一種常用效率手段。
1.解鎖消息
為了一探究竟通知了什么，通知后又做了什么，進(jìn)入LockPubSub。
這里只有一個(gè)明顯的監(jiān)聽方法onMessage，其訂閱和信號(hào)量的釋放都在父類PublishSubscribe，我們只關(guān)注監(jiān)聽事件的實(shí)際操作
protected void onMessage(RedissonLockEntry value, Long message) {
Runnable runnableToExecute;
if (message.equals(unlockMessage)) {
// 從監(jiān)聽器隊(duì)列取監(jiān)聽線程執(zhí)行監(jiān)聽回調(diào)
runnableToExecute = (Runnable)value.getListeners().poll();
if (runnableToExecute != null) {
runnableToExecute.run();
}
// getLatch()返回的是Semaphore，信號(hào)量，此處是釋放信號(hào)量
// 釋放信號(hào)量后會(huì)喚醒等待的entry.getLatch().tryAcquire去再次嘗試申請(qǐng)鎖
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)一個(gè)是默認(rèn)解鎖消息，一個(gè)是**讀鎖解鎖消息****，**因?yàn)閞edisson是有提供讀寫鎖的，而讀寫鎖讀讀情況和讀寫、寫寫情況互斥情況不同，我們只看上面的默認(rèn)解鎖消息unlockMessage分支
LockPubSub監(jiān)聽最終執(zhí)行了2件事
runnableToExecute.run() 執(zhí)行監(jiān)聽回調(diào)
value.getLatch().release(); 釋放信號(hào)量
Redisson通過LockPubSub監(jiān)聽解鎖消息，執(zhí)行監(jiān)聽回調(diào)和釋放信號(hào)量通知等待線程可以重新?lián)屾i。
這時(shí)再回來看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;
}
}可以看到，無超時(shí)時(shí)間時(shí)，在執(zhí)行加鎖操作后，還執(zhí)行了一段費(fèi)解的邏輯
ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
if (e == null) {
if (ttlRemaining) {
this.scheduleExpirationRenewal(threadId);
}
}
})
此處涉及到Netty的Future/Promise-Listener模型（參考Netty中的異步編程），Redisson中幾乎全部以這種方式通信（所以說Redisson是基于Netty通信機(jī)制實(shí)現(xiàn)的），理解這段邏輯可以試著先理解
在 Java 的 Future 中，業(yè)務(wù)邏輯為一個(gè) Callable 或 Runnable 實(shí)現(xiàn)類，該類的 call()或 run()執(zhí)行完畢意味著業(yè)務(wù)邏輯的完結(jié)，在 Promise 機(jī)制中，可以在業(yè)務(wù)邏輯中人工設(shè)置業(yè)務(wù)邏輯的成功與失敗，這樣更加方便的監(jiān)控自己的業(yè)務(wù)邏輯。
這塊代碼的表面意義就是，在執(zhí)行異步加鎖的操作后，加鎖成功則根據(jù)加鎖完成返回的ttl是否過期來確認(rèn)是否執(zhí)行一段定時(shí)任務(wù)。
這段定時(shí)任務(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ù)嵌套且冗長的代碼實(shí)際上做了幾步
? 添加一個(gè)netty的Timeout回調(diào)任務(wù)，每（internalLockLeaseTime / 3）毫秒執(zhí)行一次，執(zhí)行的方法是renewExpirationAsync
? renewExpirationAsync重置了鎖超時(shí)時(shí)間，又注冊一個(gè)監(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è)置了超時(shí)時(shí)間。
Redisson加這段邏輯的目的是什么？
目的是為了某種場景下保證業(yè)務(wù)不影響，如任務(wù)執(zhí)行超時(shí)但未結(jié)束，鎖已經(jīng)釋放的問題。
當(dāng)一個(gè)線程持有了一把鎖，由于并未設(shè)置超時(shí)時(shí)間leaseTime，Redisson默認(rèn)配置了30S，開啟watchDog，每10S對(duì)該鎖進(jìn)行一次續(xù)約，維持30S的超時(shí)時(shí)間，直到任務(wù)完成再刪除鎖。
這就是Redisson的鎖續(xù)約，也就是WatchDog實(shí)現(xiàn)的基本思路。
3.流程概括
通過整體的介紹，流程簡單概括：
A、B線程爭搶一把鎖，A獲取到后，B阻塞
B線程阻塞時(shí)并非主動(dòng)CAS，而是PubSub方式訂閱該鎖的廣播消息
A操作完成釋放了鎖，B線程收到訂閱消息通知
B被喚醒開始繼續(xù)搶鎖，拿到鎖
詳細(xì)加鎖解鎖流程總結(jié)如下圖：
五、公平鎖
以上介紹的可重入鎖是非公平鎖，Redisson還基于Redis的隊(duì)列（List）和ZSet實(shí)現(xiàn)了公平鎖
公平的定義是什么？
公平就是按照客戶端的請(qǐng)求先來后到排隊(duì)來獲取鎖，先到先得，也就是FIFO，所以隊(duì)列和容器順序編排必不可少
FairSync
回顧JUC的ReentrantLock公平鎖的實(shí)現(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)提供了整個(gè)實(shí)現(xiàn)，是否公平取決于實(shí)現(xiàn)類取出節(jié)點(diǎn)邏輯是否順序取
AbstractQueuedSynchronizer是用來構(gòu)建鎖或者其他同步組件的基礎(chǔ)框架，通過內(nèi)置FIFO隊(duì)列來完成資源獲取線程的排隊(duì)工作，他自身沒有實(shí)現(xiàn)同步接口，僅僅定義了若干同步狀態(tài)獲取和釋放的方法來供自定義同步組件使用（上圖），支持獨(dú)占和共享獲取，這是基于模版方法模式的一種設(shè)計(jì)，給公平/非公平提供了土壤。
我們用2張圖來簡單解釋AQS的等待流程（出自《JAVA并發(fā)編程的藝術(shù)》）
一張是同步隊(duì)列（FIFO雙向隊(duì)列）管理 獲取同步狀態(tài)失?。〒屾i失?。┑木€程引用、等待狀態(tài)和前驅(qū)后繼節(jié)點(diǎn)的流程圖
一張是獨(dú)占式獲取同步狀態(tài)的總流程，核心acquire(int arg)方法調(diào)用流程
可以看出鎖的獲取流程
AQS維護(hù)一個(gè)同步隊(duì)列，獲取狀態(tài)失敗的線程都會(huì)加入到隊(duì)列中進(jìn)行自旋，移出隊(duì)列或停止自旋的條件是前驅(qū)節(jié)點(diǎn)為頭節(jié)點(diǎn)切成功獲取了同步狀態(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判斷條件，該方法的作用就是
查看同步隊(duì)列中當(dāng)前節(jié)點(diǎn)是否有前驅(qū)節(jié)點(diǎn)，如果有比當(dāng)前線程更早請(qǐng)求獲取鎖則返回true。
保證每次都取隊(duì)列的第一個(gè)節(jié)點(diǎn)（線程）來獲取鎖，這就是公平規(guī)則
為什么JUC以默認(rèn)非公平鎖呢？
因?yàn)楫?dāng)一個(gè)線程請(qǐng)求鎖時(shí)，只要獲取來同步狀態(tài)即成功獲取。在此前提下，剛釋放的線程再次獲取同步狀態(tài)的幾率會(huì)非常大，使得其他線程只能在同步隊(duì)列中等待。但這樣帶來的好處是，非公平鎖大大減少了系統(tǒng)線程上下文的切換開銷。
可見公平的代價(jià)是性能與吞吐量。
Redis里沒有AQS，但是有List和zSet，看看Redisson是怎么實(shí)現(xiàn)公平的。
RedissonFairLock
RedissonFairLock 用法依然很簡單
RLock fairLock = redissonClient.getFairLock(lockName);
fairLock.lock();
RedissonFairLock繼承自RedissonLock，同樣一路向下找到加鎖實(shí)現(xiàn)方法tryLockInnerAsync。
這里有2段冗長的Lua，但是Debug發(fā)現(xiàn)，公平鎖的入口在 command == RedisCommands.EVAL_LONG 之后，此段Lua較長，參數(shù)也多，我們著重分析Lua的實(shí)現(xiàn)規(guī)則
參數(shù)
-- lua中的幾個(gè)參數(shù)
KEYS = Arrays.<Object>asList(getName(), threadsQueueName, timeoutSetName)
KEYS[1]: lock_name, 鎖名稱
KEYS[2]: "redisson_lock_queue:{xxx}" 線程隊(duì)列
KEYS[3]: "redisson_lock_timeout:{xxx}" 線程id對(duì)應(yīng)的超時(shí)集合
ARGV = internalLockLeaseTime, getLockName(threadId), currentTime + threadWaitTime, currentTime
ARGV[1]: "{leaseTime}" 過期時(shí)間
ARGV[2]: "{Redisson.UUID}:{threadId}"
ARGV[3] = 當(dāng)前時(shí)間 + 線程等待時(shí)間:（10:00:00） + 5000毫秒 = 10:00:05
ARGV[4] = 當(dāng)前時(shí)間（10:00:00） 部署服務(wù)器時(shí)間，非redis-server服務(wù)器時(shí)間公平鎖實(shí)現(xiàn)的Lua腳本
-- 1.死循環(huán)清除過期key
while true do
-- 獲取頭節(jié)點(diǎn)
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.不存在該鎖 && （不存在線程等待隊(duì)列 || 存在線程等待隊(duì)列而且第一個(gè)節(jié)點(diǎn)就是此線程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
-- 彈出隊(duì)列中線程id元素，刪除Zset中該線程id對(duì)應(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的超時(shí)時(shí)間(score) - 當(dāng)前時(shí)間ms
for i = 1, #keys, 1 do
redis.call('zincrby', KEYS[3], -tonumber(ARGV[3]), keys[i]);
end;
-- 加鎖設(shè)置鎖過期時(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)前線程剩余存活時(shí)間
local timeout = redis.call('zscore', KEYS[3], ARGV[2]);
if timeout ~= false then
-- 過期時(shí)間timeout的值在下方設(shè)置，此處的減法算出的依舊是當(dāng)前線程的ttl
return timeout - tonumber(ARGV[3]) - tonumber(ARGV[4]);
end;
-- 5.尾節(jié)點(diǎn)剩余存活時(shí)間
local lastThreadId = redis.call('lindex', KEYS[2], -1);
local ttl;
-- 尾節(jié)點(diǎn)不空 && 尾節(jié)點(diǎn)非當(dāng)前線程
if lastThreadId ~= false and lastThreadId ~= ARGV[2] then
-- 計(jì)算隊(duì)尾節(jié)點(diǎn)剩余存活時(shí)間
ttl = tonumber(redis.call('zscore', KEYS[3], lastThreadId)) - tonumber(ARGV[4]);
else
-- 獲取lock_name剩余存活時(shí)間
ttl = redis.call('pttl', KEYS[1]);
end;
-- 6.末尾排隊(duì)
-- zSet 超時(shí)時(shí)間（score），尾節(jié)點(diǎn)ttl + 當(dāng)前時(shí)間 + 5000ms + 當(dāng)前時(shí)間，無則新增，有則更新
-- 線程id放入隊(duì)列尾部排隊(duì)，無則插入，有則不再插入
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維護(hù)了一個(gè)等待的線程隊(duì)列redisson_lock_queue:{xxx}，zSet維護(hù)了一個(gè)線程超時(shí)情況的有序集合redisson_lock_timeout:{xxx}，盡管lua較長，但是可以拆分為6個(gè)步驟
1. 隊(duì)列清理
保證隊(duì)列中只有未過期的等待線程
2. 首次加鎖
hset加鎖，pexpire過期時(shí)間
3. 重入判斷
此處同可重入鎖lua
4. 返回ttl
5. 計(jì)算尾節(jié)點(diǎn)ttl
5. 初始值為鎖的剩余過期時(shí)間
6. 末尾排隊(duì)
ttl + 2 * currentTime + waitTime是score的默認(rèn)值計(jì)算公式
2.模擬
如果模擬以下順序，就會(huì)明了redisson公平鎖整個(gè)加鎖流程
假設(shè) t1 10:00:00 < t2 10:00:10 < t3 10:00:20
t1：當(dāng)線程1初次獲取鎖
1.等待隊(duì)列無頭節(jié)點(diǎn)，跳出死循環(huán)->2
2.不存在該鎖 && 不存在線程等待隊(duì)列 成立
2.1 lpop和zerm、zincrby都是無效操作，只有加鎖生效，說明是首次加鎖，加鎖后返回nil
加鎖成功，線程1獲取到鎖，結(jié)束
t2：線程2嘗試獲取鎖（線程1未釋放鎖）
1.等待隊(duì)列無頭節(jié)點(diǎn)，跳出死循環(huán)->2
2.不存在該鎖 不成立->3
3.非重入線程 ->4
4.score無值 ->5
5.尾節(jié)點(diǎn)為空，設(shè)置ttl初始值為lock_name的ttl -> 6
6.按照ttl + waitTime + currentTime + currentTime 來設(shè)置zSet超時(shí)時(shí)間score，并且加入等待隊(duì)列，線程2為頭節(jié)點(diǎn)
score = 20S + 5000ms + 10:00:10 + 10:00:10 = 10:00:35 + 10:00:10
t3：線程3嘗試獲取鎖（線程1未釋放鎖）
1.等待隊(duì)列有頭節(jié)點(diǎn)
1.1未過期->2
2.不存在該鎖不成立->3
3.非重入線程->4
4.score無值 ->5
5.尾節(jié)點(diǎn)不為空 && 尾節(jié)點(diǎn)線程為2，非當(dāng)前線程
5.1取出之前設(shè)置的score，減去當(dāng)前時(shí)間：ttl = score - currentTime ->6
6.按照ttl + waitTime + currentTime + currentTime 來設(shè)置zSet超時(shí)時(shí)間score，并且加入等待隊(duì)列
score = 10S + 5000ms + 10:00:20 + 10:00:20 = 10:00:35 + 10:00:20
如此一來，三個(gè)需要搶奪一把鎖的線程，完成了一次排隊(duì)，在list中排列他們等待線程id，在zSet中存放過期時(shí)間（便于排列優(yōu)先級(jí)）。其中返回ttl的線程2客戶端、線程3客戶端將會(huì)一直按一定間隔自旋重復(fù)執(zhí)行該段Lua，嘗試加鎖，如此一來便和AQS有了異曲同工之處。
而當(dāng)線程1釋放鎖之后（這里依舊有通過Pub/Sub發(fā)布解鎖消息，通知其他線程獲?。?/p>
10:00:30 線程2嘗試獲取鎖（線程1已釋放鎖）
1.等待隊(duì)列有頭節(jié)點(diǎn)，未過期->2
2.不存在該鎖 & 等待隊(duì)列頭節(jié)點(diǎn)是當(dāng)前線程 成立
2.1刪除當(dāng)前線程的隊(duì)列信息和zSet信息，超時(shí)時(shí)間為：
線程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è)置過期時(shí)間
加鎖成功，線程2獲取到鎖，結(jié)束
排隊(duì)結(jié)構(gòu)如圖
公平鎖的釋放腳本和重入鎖類似，多了一步加鎖開頭的清理過期key的while true邏輯，在此不再展開篇幅描述。
由上可以看出，Redisson公平鎖的玩法類似于延遲隊(duì)列的玩法，核心都在Redis的List和zSet結(jié)構(gòu)的搭配，但又借鑒了AQS實(shí)現(xiàn)，在定時(shí)判斷頭節(jié)點(diǎn)上如出一轍（watchDog），保證了鎖的競爭公平和互斥。并發(fā)場景下，lua腳本里，zSet的score很好地解決了順序插入的問題，排列好優(yōu)先級(jí)。并且為了防止因異常而退出的線程無法清理，每次請(qǐng)求都會(huì)判斷頭節(jié)點(diǎn)的過期情況給予清理，最后釋放時(shí)通過CHANNEL通知訂閱線程可以來獲取鎖，重復(fù)一開始的步驟，順利交接到下一個(gè)順序線程。
六、總結(jié)
Redisson整體實(shí)現(xiàn)分布式加解鎖流程的實(shí)現(xiàn)稍顯復(fù)雜，作者Rui Gu對(duì)Netty和JUC、Redis研究深入，利用了很多高級(jí)特性和語義，值得深入學(xué)習(xí)，本次介紹也只是單機(jī)Redis下鎖實(shí)現(xiàn)，Redisson也提供了多機(jī)情況下的聯(lián)鎖（MultiLock)和官方推薦的紅鎖（RedLock），下一章再詳細(xì)介紹。
所以，當(dāng)你真的需要分布式鎖時(shí)，不妨先來Redisson里找找。
-End-
最近有一些小伙伴，讓我?guī)兔φ乙恍?nbsp;面試題 資料，于是我翻遍了收藏的 5T 資料后，匯總整理出來，可以說是程序員面試必備！所有資料都整理到網(wǎng)盤了，歡迎下載！
面試題】即可獲取
好文分享給更多人↓↓