一文掌握Redisson分布式鎖原理|干貨推薦
ReentrantLock 重入鎖
在說 Redisson 之前我們先來說一下 JDK 可重入鎖: ReentrantLock
ReentrantLock 保證了 JVM 共享資源同一時刻只允許單個線程進行操作
實現(xiàn)思路
ReentrantLock 內部公平鎖與非公平鎖繼承了 AQS[AbstractQueuedSynchronizer]
1、AQS 內部通過 volatil 修飾的 int 類型變量 state 控制并發(fā)情況下線程安全問題及鎖重入
2、將未競爭到鎖的線程放入 AQS 的隊列中通過 LockSupport#park、unPark 掛起喚醒
簡要描述哈, 詳情可以查看具體的文章
Redisson
可以直接查看 Github Redisson官網 介紹, 沒有了解過的小伙伴, 看一下 Redisson 的 WIKI 目錄, 仔細瞅瞅 Redis 是如何被 Redisson 武裝到牙齒的
上下滾動查看更多
這里先過一下和文章有關的一部分內容
通過項目簡介可以看出來, 寫這個項目介紹的人水平非常哇塞哈, 從第一段咱們就知道了兩個問題
Redisson 是什么
Redisson 是架設在 Redis 基礎上的一個 Java 駐內存數(shù)據網格框架, 充分利用 Redis 鍵值數(shù)據庫提供的一系列優(yōu)勢, 基于 Java 實用工具包中常用接口, 為使用者提供了 一系列具有分布式特性的常用工具類
Redisson 的優(yōu)勢
使得原本作為協(xié)調單機多線程并發(fā)程序的工具包 獲得了協(xié)調分布式多機多線程并發(fā)系統(tǒng)的能力, 大大降低了設計和研發(fā)大規(guī)模分布式系統(tǒng)的難度
同時結合各富特色的分布式服務, 更進一步 簡化了分布式環(huán)境中程序相互之間的協(xié)作
“了解到這里就差不多了, 就不向下擴展了, 想要了解詳細用途的, 翻一下上面的目錄
Redisson 重入鎖
由于 Redisson 太過于復雜, 設計的 API 調用大多用 Netty 相關, 所以這里只對 如何加鎖、如何實現(xiàn)重入鎖進行分析以及如何鎖續(xù)時進行分析
創(chuàng)建鎖
我這里是將 Redisson 的源碼下載到本地了
下面這個簡單的程序, 就是使用 Redisson 創(chuàng)建了一個非公平的可重入鎖
lock() 方法加鎖成功 默認過期時間 30 秒, 并且支持 "看門狗" 續(xù)時功能
public static void main(String[] args) {
Config config = new Config();
config.useSingleServer()
.setPassword("123456")
.setAddress("redis://127.0.0.1:6379");
RedissonClient redisson = Redisson.create(config);
RLock lock = redisson.getLock("myLock");
try {
lock.lock();
// 業(yè)務邏輯
} finally {
lock.unlock();
}
}
我們先來看一下 RLock 接口的聲明
public interface RLock extends Lock, RLockAsync {}
RLock 繼承了 JDK 源碼 JUC 包下的 Lock 接口, 同時也繼承了 RLockAsync
RLockAsync 從字面意思看是 支持異步的鎖, 證明獲取鎖時可以異步獲取
“看了 Redisson 的源碼會知道, 注釋比黃金貴 ???
由于獲取鎖的 API 較多, 我們這里以 lock() 做源碼講解, 看接口定義相當簡單
/**
* lock 并沒有指定鎖過期時間, 默認 30 秒
* 如果獲取到鎖, 會對鎖進行續(xù)時
*/
void lock();
獲取鎖實例
根據上面的小 Demo, 看下第一步獲取鎖是如何做的
RLock lock = redisson.getLock("myLock");
// name 就是鎖名稱
public RLock getLock(String name) {
// 默認創(chuàng)建的同步執(zhí)行器, (存在異步執(zhí)行器, 因為鎖的獲取和釋放是有強一致性要求, 默認同步)
return new RedissonLock(connectionManager.getCommandExecutor(), name);
}
Redisson 中所有 Redis 命令都是通過 ...Executor 執(zhí)行的
獲取到默認的同步執(zhí)行器后, 就要初始化 RedissonLock
public RedissonLock(CommandAsyncExecutor commandExecutor, String name) {
super(commandExecutor, name);
this.commandExecutor = commandExecutor;
// 唯一ID
this.id = commandExecutor.getConnectionManager().getId();
// 等待獲取鎖時間
this.internalLockLeaseTime = commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout();
// ID + 鎖名稱
this.entryName = id + ":" + name;
// 發(fā)布訂閱, 后面關于加、解鎖流程會用到
this.pubSub = commandExecutor.getConnectionManager().getSubscribeService().getLockPubSub();
}
嘗試獲取鎖
我們來看一下 RLock#lock() 底層是如何獲取鎖的
@Override
public void lock() {
try {
lock(-1, null, false);
} catch (InterruptedException e) {
throw new IllegalStateException();
}
}
leaseTime: 加鎖到期時間, -1 使用默認值 30 秒
unit: 時間單位, 毫秒、秒、分鐘、小時...
interruptibly: 是否可被中斷標示
private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
// 獲取當前線程ID
long threadId = Thread.currentThread().getId();
// ?? 嘗試獲取鎖, 下面重點分析
Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
// 成功獲取鎖, 過期時間為空
if (ttl == null) {
return;
}
// 訂閱分布式鎖, 解鎖時進行通知
RFuture<RedissonLockEntry> future = subscribe(threadId);
if (interruptibly) {
commandExecutor.syncSubscriptionInterrupted(future);
} else {
commandExecutor.syncSubscription(future);
}
try {
while (true) {
// 再次嘗試獲取鎖
ttl = tryAcquire(-1, leaseTime, unit, threadId);
// 成功獲取鎖, 過期時間為空, 成功返回
if (ttl == null) {
break;
}
// 鎖過期時間如果大于零, 則進行帶過期時間的阻塞獲取
if (ttl >= 0) {
try {
// 獲取不到鎖會在這里進行阻塞, Semaphore, 解鎖時釋放信號量通知
future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
if (interruptibly) {
throw e;
}
future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
}
// 鎖過期時間小于零, 則死等, 區(qū)分可中斷及不可中斷
} else {
if (interruptibly) {
future.getNow().getLatch().acquire();
} else {
future.getNow().getLatch().acquireUninterruptibly();
}
}
}
} finally {
// 取消訂閱
unsubscribe(future, threadId);
}
}
這一段代碼是用來執(zhí)行加鎖, 繼續(xù)看下方法實現(xiàn)
Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
private Long tryAcquire(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
return get(tryAcquireAsync(waitTime, leaseTime, unit, threadId));
}
lock() 以及 tryLock(...) 方法最終都會調用此方法, 分為兩個流程分支
1、tryLock(...) API 異步加鎖返回
2、lock() & tryLock() API 異步加鎖并進行鎖續(xù)時
private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
// 執(zhí)行 tryLock(...) 才會進入
if (leaseTime != -1) {
// 進行異步獲取鎖
return tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
}
// 嘗試異步獲取鎖, 獲取鎖成功返回空, 否則返回鎖剩余過期時間
RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(waitTime,
commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),
TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
// ttlRemainingFuture 執(zhí)行完成后觸發(fā)此操作
ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
if (e != null) {
return;
}
// ttlRemaining == null 代表獲取了鎖
// 獲取到鎖后執(zhí)行續(xù)時操作
if (ttlRemaining == null) {
scheduleExpirationRenewal(threadId);
}
});
return ttlRemainingFuture;
}
繼續(xù)看一下 tryLockInnerAsync(...) 詳細的加鎖流程, 內部采用的 Lua 腳本形式, 保證了原子性操作
到這一步大家就很明了了, 將 Lua 腳本被 Redisoon 包裝最后通過 Netty 進行傳輸
<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
internalLockLeaseTime = unit.toMillis(leaseTime);
return evalWriteAsync(getName(), LongCodec.INSTANCE, command,
"if (redis.call('exists', KEYS[1]) == 0) then " +
"redis.call('hincrby', 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(getName()), internalLockLeaseTime, getLockName(threadId));
}
“evalWriteAsync(...) 是對 Eval 命令的封裝以及 Netty 的應用就不繼續(xù)跟進了
加鎖 Lua
執(zhí)行 Redis 加鎖的 Lua 腳本, 截個圖讓大家看一下參數(shù)以及具體含義
KEYS[1]: myLock
ARGV[1]: 36000... 這個是過期時間, 自己測試的, 單位毫秒
ARGV[2]: UUID + 線程 ID
# KEYS[1] 代表上面的 myLock
# 判斷 KEYS[1] 是否存在, 存在返回 1, 不存在返回 0
if (redis.call('exists', KEYS[1]) == 0) then
# 當 KEYS[1] == 0 時代表當前沒有鎖
# 使用 hincrby 命令發(fā)現(xiàn) KEYS[1] 不存在并新建一個 hash
# ARGV[2] 就作為 hash 的第一個key, val 為 1
# 相當于執(zhí)行了 hincrby myLock 91089b45... 1
redis.call('hincrby', KEYS[1], ARGV[2], 1);
# 設置 KEYS[1] 過期時間, 單位毫秒
redis.call('pexpire', KEYS[1], ARGV[1]);
return nil;
end;
# 查找 KEYS[1] 中 key ARGV[2] 是否存在, 存在回返回 1
if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then
# 同上, ARGV[2] 為 key 的 val +1
redis.call('hincrby', KEYS[1], ARGV[2], 1);
# 同上
redis.call('pexpire', KEYS[1], ARGV[1]);
return nil;
end;
# 返回 KEYS[1] 過期時間, 單位毫秒
return redis.call('pttl', KEYS[1]);
整個 Lua 腳本加鎖的流程畫圖如下:
現(xiàn)在回過頭看一下獲取到鎖之后, 是如何為鎖進行延期操作的
鎖續(xù)時
之前有和軍哥聊過這個話題, 他說的思路和 Redisson 中體現(xiàn)的基本一致
說一下 Redisson 的具體實現(xiàn)思路吧, 中文翻譯叫做 "看門狗"
1、獲取到鎖之后執(zhí)行 "看門狗" 流程
2、使用 Netty 的 Timeout 實現(xiàn)定時延時
3、比如鎖過期 30 秒, 每過 1/3 時間也就是 10 秒會檢查鎖是否存在, 存在則更新鎖的超時時間
可能會有小伙伴會提出這么一個疑問, 如果檢查返回存在, 設置鎖過期時剛好鎖被釋放了怎么辦?
有這樣的疑問, 代表確實用心去考慮所有可能發(fā)生的情況了, 但是不必擔心哈
Redisson 中使用的 Lua 腳本做的檢查及設置過期時間操作, 本身是原子性的不會出現(xiàn)上面情況
“如果不想要引用 Netty 的包, 使用延時隊列等包工具也是可以完成 "看門狗"
這里也貼一哈相關代碼, 能夠讓小伙伴更直觀的了解如何鎖續(xù)時的
我可真是個暖男, 上代碼 RedissonLock#tryAcquireAsync(...)
private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
// ...
// 嘗試異步獲取鎖, 獲取鎖成功返回空, 否則返回鎖剩余過期時間
RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(waitTime,
commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),
TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
// ttlRemainingFuture 執(zhí)行完成后觸發(fā)此操作
ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
if (e != null) {
return;
}
// 獲取到鎖后執(zhí)行續(xù)時操作
if (ttlRemaining == null) {
scheduleExpirationRenewal(threadId);
}
});
return ttlRemainingFuture;
}
可以看到續(xù)時方法將 threadId 當作標識符進行續(xù)時
private void scheduleExpirationRenewal(long threadId) {
ExpirationEntry entry = new ExpirationEntry();
ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);
if (oldEntry != null) {
oldEntry.addThreadId(threadId);
} else {
entry.addThreadId(threadId);
renewExpiration();
}
}
知道核心理念就好了, 沒必要研究每一行代碼哈
private void renewExpiration() {
ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
if (ee == null) {
return;
}
Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
@Override
public void run(Timeout timeout) throws Exception {
ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());
if (ent == null) {
return;
}
Long threadId = ent.getFirstThreadId();
if (threadId == null) {
return;
}
RFuture<Boolean> future = renewExpirationAsync(threadId);
future.onComplete((res, e) -> {
if (e != null) {
log.error("Can't update lock " + getName() + " expiration", e);
return;
}
if (res) {
// 調用本身
renewExpiration();
}
});
}
}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
ee.setTimeout(task);
}
解鎖操作
解鎖時的操作相對加鎖還是比較簡單的
@Override
public void unlock() {
try {
get(unlockAsync(Thread.currentThread().getId()));
} catch (RedisException e) {
if (e.getCause() instanceof IllegalMonitorStateException) {
throw (IllegalMonitorStateException) e.getCause();
} else {
throw e;
}
}
}
解鎖成功后會將之前的"看門狗" Timeout 續(xù)時取消, 并返回成功
@Override
public RFuture<Void> unlockAsync(long threadId) {
RPromise<Void> result = new RedissonPromise<Void>();
RFuture<Boolean> future = unlockInnerAsync(threadId);
future.onComplete((opStatus, e) -> {
// 取消自動續(xù)時功能
cancelExpirationRenewal(threadId);
if (e != null) {
// 失敗
result.tryFailure(e);
return;
}
if (opStatus == null) {
IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
+ id + " thread-id: " + threadId);
result.tryFailure(cause);
return;
}
// 解鎖成功
result.trySuccess(null);
});
return result;
}
又是一個精髓點, 解鎖的 Lua 腳本定義
protected RFuture<Boolean> unlockInnerAsync(long threadId) {
return evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
"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(getName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));
}
還是來張圖理解哈, Lua 腳本會詳細分析
解鎖 Lua
老規(guī)矩, 圖片加參數(shù)說明
KEYS[1]: myLock
KEYS[2]: redisson_lock_channel:{myLock}
ARGV[1]: 0
ARGV[2]: 360000... (過期時間)
ARGV[3]: 7f0c54e2...(Hash 中的鎖 Key)
# 判斷 KEYS[1] 中是否存在 ARGV[3]
if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then
return nil;
end;
# 將 KEYS[1] 中 ARGV[3] Val - 1
local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1);
# 如果返回大于0 證明是一把重入鎖
if (counter > 0) then
# 重制過期時間
redis.call('pexpire', KEYS[1], ARGV[2]);
return 0;
else
# 刪除 KEYS[1]
redis.call('del', KEYS[1]);
# 通知阻塞等待線程或進程資源可用
redis.call('publish', KEYS[2], ARGV[1]);
return 1;
end;
return nil;
Redlock 算法
不可否認, Redisson 設計的分布式鎖真的很 NB, 但是還是沒有解決 主從節(jié)點下異步同步數(shù)據導致鎖丟失問題
所以 Redis 作者 Antirez 推出 紅鎖算法, 這個算法的精髓就是: 沒有從節(jié)點, 如果部署多臺 Redis, 各實例之間相互獨立, 不存在主從復制或者其他集群協(xié)調機制
如何使用
創(chuàng)建多個 Redisson Node, 由這些無關聯(lián)的 Node 組成一個完整的分布式鎖
public static void main(String[] args) {
String lockKey = "myLock";
Config config = new Config();
config.useSingleServer().setPassword("123456").setAddress("redis://127.0.0.1:6379");
Config config2 = new Config();
config.useSingleServer().setPassword("123456").setAddress("redis://127.0.0.1:6380");
Config config3 = new Config();
config.useSingleServer().setPassword("123456").setAddress("redis://127.0.0.1:6381");
RLock lock = Redisson.create(config).getLock(lockKey);
RLock lock2 = Redisson.create(config2).getLock(lockKey);
RLock lock3 = Redisson.create(config3).getLock(lockKey);
RedissonRedLock redLock = new RedissonRedLock(lock, lock2, lock3);
try {
redLock.lock();
} finally {
redLock.unlock();
}
}
當然, 對于 Redlock 算法不是沒有質疑聲, 大家可以去 Redis 官網查看Martin Kleppmann 與 Redis 作者Antirez 的辯論
CAP 原則之間的取舍
CAP 原則又稱 CAP 定理, 指的是在一個分布式系統(tǒng)中, Consistency（一致性）、 Availability（可用性）、Partition tolerance（分區(qū)容錯性）, 三者不可得兼
一致性(C) : 在分布式系統(tǒng)中的所有數(shù)據備份, 在同一時刻是否同樣的值(等同于所有節(jié)點訪問同一份最新的數(shù)據副本)
可用性(A): 在集群中一部分節(jié)點故障后, 集群整體是否還能響應客戶端的讀寫請求(對數(shù)據更新具備高可用性)
分區(qū)容忍性(P): 以實際效果而言, 分區(qū)相當于對通信的時限要求. 系統(tǒng)如果不能在時限內達成數(shù)據一致性, 就意味著發(fā)生了分區(qū)的情況, 必須就當前操作在 C 和 A 之間做出選擇
分布式鎖選型
如果要滿足上述分布式鎖之間的強一致性, 可以采用 Zookeeper 的分布式鎖, 因為它底層的 ZAB協(xié)議(原子廣播協(xié)議), 天然滿足 CP
但是這也意味著性能的下降, 所以不站在具體數(shù)據下看 Redis 和 Zookeeper, 代表著性能和一致性的取舍
如果項目沒有強依賴 ZK, 使用 Redis 就好了, 因為現(xiàn)在 Redis 用途很廣, 大部分項目中都引用了 Redis
沒必要對此再引入一個新的組件, 如果業(yè)務場景對于 Redis 異步方式的同步數(shù)據造成鎖丟失無法忍受, 在業(yè)務層處理就好了
往期推薦
Redis為什么變慢了？一文詳解Redis性能問題 | 萬字長文
Redis 消息隊列的三種方案（List、Streams、Pub/Sub）
硬核Redis總結，看這篇就夠了！