redisson 實現(xiàn)分布式鎖的源碼解析
更新時間：2022年05月05日 16:06:59 作者：心城以北
這篇文章主要介紹了redisson 實現(xiàn)分布式鎖的源碼解析,通過模擬一個商品秒殺的場景結合示例代碼給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下
redisson
redisson 實現(xiàn)分布式鎖的機制如下：
依賴版本
implementation 'org.redisson:redisson-spring-boot-starter:3.17.0'
測試代碼
下面是模擬一個商品秒殺的場景，示例代碼如下：
public class RedissonTest {
public static void main(String[] args) {
//1. 配置部分
Config config = new Config();
String address = "redis://127.0.0.1:6379";
SingleServerConfig serverConfig = config.useSingleServer();
serverConfig.setAddress(address);
serverConfig.setDatabase(0);
config.setLockWatchdogTimeout(5000);
Redisson redisson = (Redisson) Redisson.create(config);
RLock rLock = redisson.getLock("goods:1000:1");
//2. 加鎖
rLock.lock();
try {
System.out.println("todo 邏輯處理 1000000.");
} finally {
if (rLock.isLocked() && rLock.isHeldByCurrentThread()) {
//3. 解鎖
rLock.unlock();
}
}
}
}
加鎖設計
rLock.lock();是加鎖的核心代碼，我們一起來看看調用棧
加鎖的核心方法是：org.redisson.RedissonLock#tryLockInnerAsync
<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
return evalWriteAsync(getRawName(), 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(getRawName()), unit.toMillis(leaseTime), getLockName(threadId));
}其實它的本質是調用一段 LUA 腳本進行加鎖。
鎖續(xù)期設計
鎖的續(xù)期是在 org.redisson.RedissonLock#tryAcquireAsync方法中調用 scheduleExpirationRenewal實現(xiàn)的。
續(xù)期需要注意的是，看門狗是設置在主線程的延遲隊列的線程中。
tryAcquireAsync 代碼如下：
private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
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);
}
CompletionStage<Long> f = ttlRemainingFuture.thenApply(ttlRemaining -> {
// lock acquired
if (ttlRemaining == null) {
if (leaseTime != -1) {
internalLockLeaseTime = unit.toMillis(leaseTime);
} else {
// 鎖過期時間續(xù)期
scheduleExpirationRenewal(threadId);
}
}
return ttlRemaining;
});
return new CompletableFutureWrapper<>(f);
}
鎖續(xù)期 scheduleExpirationRenewal代碼如下：
protected 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);
try {
renewExpiration();
} finally {
if (Thread.currentThread().isInterrupted()) {
cancelExpirationRenewal(threadId);
}
}
}
}然后在調用 renewExpiration();執(zhí)行續(xù)期邏輯
private void renewExpiration() {
ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
if (ee == null) {
return;
}
// 創(chuàng)建延遲任務
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;
}
// 真正的續(xù)期，調用 LUA 腳本續(xù)期
RFuture<Boolean> future = renewExpirationAsync(threadId);
future.whenComplete((res, e) -> {
if (e != null) {
log.error("Can't update lock " + getRawName() + " expiration", e);
EXPIRATION_RENEWAL_MAP.remove(getEntryName());
return;
}
// 如果續(xù)期成功
if (res) {
// reschedule itself
renewExpiration();
} else {
cancelExpirationRenewal(null);
}
});
}
}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
ee.setTimeout(task);
}
renewExpirationAsync方法, 里面還是一段 LUA 腳本，進行重新設置鎖的過期時間。
protected RFuture<Boolean> renewExpirationAsync(long threadId) {
return evalWriteAsync(getRawName(), 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(getRawName()),
internalLockLeaseTime, getLockName(threadId));
}鎖的自旋重試
org.redisson.RedissonLock#lock(long, java.util.concurrent.TimeUnit, boolean)在執(zhí)行獲取鎖失敗的時候，會進入重試。其實這里就會執(zhí)行 18 行以后的 while (true) 邏輯
private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
long threadId = Thread.currentThread().getId();
Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
return;
}
CompletableFuture<RedissonLockEntry> future = subscribe(threadId);
RedissonLockEntry entry;
if (interruptibly) {
entry = commandExecutor.getInterrupted(future);
} else {
entry = commandExecutor.get(future);
}
try {
while (true) {
ttl = tryAcquire(-1, leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
break;
}
// waiting for message
if (ttl >= 0) {
try {
// 阻塞鎖的超時時間，等鎖過期后再嘗試加鎖
entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
if (interruptibly) {
throw e;
}
entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
}
} else {
if (interruptibly) {
entry.getLatch().acquire();
} else {
entry.getLatch().acquireUninterruptibly();
}
}
}
} finally {
unsubscribe(entry, threadId);
}
// get(lockAsync(leaseTime, unit));
}
entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);其實這里就是一個間歇性自旋。 等到上次鎖過期的時間，在喚醒進行搶鎖 entry.getLatch().acquire();
還有一個邏輯就是
CompletableFuture future = subscribe(threadId);
這里其實是會訂閱一個消息，如果解鎖過后，會發(fā)布解鎖的消息。
解鎖設計
rLock.unlock(); 的核心就是釋放鎖，撤銷續(xù)期和喚醒在等待加鎖的線程（發(fā)布解鎖成功消息）。
核心方法（解鎖）： org.redisson.RedissonLock#unlockInnerAsync
protected RFuture<Boolean> unlockInnerAsync(long threadId) {
return evalWriteAsync(getRawName(), 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]); " +
// 發(fā)布解鎖成功消息
"redis.call('publish', KEYS[2], ARGV[1]); " +
"return 1; " +
"end; " +
"return nil;",
Arrays.asList(getRawName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));
}還是 LUA 的執(zhí)行方式。
撤銷鎖續(xù)期
核心方法 org.redisson.RedissonBaseLock#unlockAsync(long)
@Override
public RFuture<Void> unlockAsync(long threadId) {
// 解鎖
RFuture<Boolean> future = unlockInnerAsync(threadId);
// 撤銷續(xù)期
CompletionStage<Void> f = future.handle((opStatus, e) -> {
cancelExpirationRenewal(threadId);
if (e != null) {
throw new CompletionException(e);
}
if (opStatus == null) {
IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
+ id + " thread-id: " + threadId);
throw new CompletionException(cause);
}
return null;
});
return new CompletableFutureWrapper<>(f);
}解鎖成功喚排隊線程
在 org.redisson.pubsub.LockPubSub#onMessage中回去喚醒阻塞的線程，讓執(zhí)行前面的鎖自旋邏輯，具體代碼如下：
@Override
protected void onMessage(RedissonLockEntry value, Long message) {
if (message.equals(UNLOCK_MESSAGE)) {
Runnable runnableToExecute = value.getListeners().poll();
if (runnableToExecute != null) {
runnableToExecute.run();
}
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());
}
}到此這篇關于redisson 實現(xiàn)分布式鎖的文章就介紹到這了,更多相關redisson 分布式鎖內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
this可能是幾乎所有有一點面向對象思想的語言都會引用到的變量，this有多少種用法。下面小編給大家?guī)砹薺ava中this的n種使用方法，感興趣的朋友一起看看吧2018-08-08
SpringBoot創(chuàng)建RSocket服務器的全過程記錄
RSocket應用層協(xié)議支持 Reactive Streams語義， 例如：用RSocket作為HTTP的一種替代方案。這篇文章主要給大家介紹了關于SpringBoot創(chuàng)建RSocket服務器的相關資料,需要的朋友可以參考下2021-05-05
這篇文章主要介紹了Java經(jīng)典排序算法之插入排序代碼實例,插入排序是一種最簡單直觀的排序算法,它的工作原理是通過構建有序序列,對于未排序數(shù)據(jù),在已排序序列中從后向前掃描,找到相應位置并插入,需要的朋友可以參考下2023-10-10
Java 創(chuàng)建動態(tài)類和查看方法列表信息的實例
這篇文章主要介紹了 Java 創(chuàng)建動態(tài)類和查看方法列表信息的實例的相關資料,需要的朋友可以參考下2017-06-06
Can''t use Subversion command line client:svn 報錯處理
這篇文章主要介紹了Can't use Subversion command line client:svn 報錯處理的相關資料,需要的朋友可以參考下2016-09-09
這篇文章主要介紹了java懶惰評估如何實現(xiàn)的相關內(nèi)容及實例，有興趣的朋友們可以學習參考下。2021-05-05
這篇文章主要介紹了淺談基于Consul創(chuàng)建分布式鎖,Consul是HashiCorp公司推出的開源工具，用于實現(xiàn)分布式系統(tǒng)的服務發(fā)現(xiàn)與配置Consul是分布式的、高可用的、可橫向擴展的,需要的朋友可以參考下2023-07-07
Spring?Boot?2.6.x整合Swagger啟動失敗報錯問題的完美解決辦法
這篇文章主要給大家介紹了關于Spring?Boot?2.6.x整合Swagger啟動失敗報錯問題的完美解決辦法,文中通過實例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2022-03-03
本文主要介紹了JAVA中的基本數(shù)據(jù)類型。具有很好的參考價值，下面跟著小編一起來看下吧2017-02-02
這篇文章主要介紹了基于jmeter錄制瀏覽器Https請求過程圖解,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2020-04-04