ReentrantLock 重入鎖

在說 Redisson 之前我們先來說一下 JDK 可重入鎖: ReentrantLock

ReentrantLock 保證了 JVM 共享資源同一時刻只允許單個線程進行操作

實現思路

ReentrantLock 內部公平鎖與非公平鎖繼承了 AQS[AbstractQueuedSynchronizer]

1、AQS 內部通過 volatil 修飾的 int 類型變量 state 控制并發情況下線程安全問題及鎖重入

2、將未競爭到鎖的線程放入 AQS 的隊列中通過 LockSupport#park、unPark 掛起喚醒

簡要描述哈, 詳情可以查看具體的文章

Redisson

可以直接查看 Github Redisson官網 介紹, 沒有了解過的小伙伴, 看一下 Redisson 的 WIKI 目錄, 仔細瞅瞅 Redis 是如何被 Redisson 武裝到牙齒的

上下滾動查看更多

這里先過一下和文章有關的一部分內容

通過項目簡介可以看出來, 寫這個項目介紹的人水平非常哇塞哈, 從第一段咱們就知道了兩個問題

Redisson 是什么

Redisson 是架設在 Redis 基礎上的一個 Java 駐內存數據網格框架, 充分利用 Redis 鍵值數據庫提供的一系列優勢, 基于 Java 實用工具包中常用接口, 為使用者提供了 一系列具有分布式特性的常用工具類

Redisson 的優勢

使得原本作為協調單機多線程并發程序的工具包 獲得了協調分布式多機多線程并發系統的能力, 大大降低了設計和研發大規模分布式系統的難度

同時結合各富特色的分布式服務, 更進一步 簡化了分布式環境中程序相互之間的協作

“

了解到這里就差不多了, 就不向下擴展了, 想要了解詳細用途的, 翻一下上面的目錄

Redisson 重入鎖

由于 Redisson 太過于復雜, 設計的 API 調用大多用 Netty 相關, 所以這里只對 如何加鎖、如何實現重入鎖進行分析以及如何鎖續時進行分析

創建鎖

我這里是將 Redisson 的源碼下載到本地了

下面這個簡單的程序, 就是使用 Redisson 創建了一個非公平的可重入鎖

lock() 方法加鎖成功 默認過期時間 30 秒, 并且支持 "看門狗" 續時功能

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();//?業務邏輯}?finally?{lock.unlock();} }

我們先來看一下 RLock 接口的聲明

public?interface?RLock?extends?Lock,?RLockAsync?{}

RLock 繼承了 JDK 源碼 JUC 包下的 Lock 接口, 同時也繼承了 RLockAsync

RLockAsync 從字面意思看是 支持異步的鎖, 證明獲取鎖時可以異步獲取

“

看了 Redisson 的源碼會知道, 注釋比黃金貴 ?????

由于獲取鎖的 API 較多, 我們這里以 lock() 做源碼講解, 看接口定義相當簡單

/***?lock?并沒有指定鎖過期時間,?默認?30?秒*?如果獲取到鎖,?會對鎖進行續時*/ void?lock();

獲取鎖實例

根據上面的小 Demo, 看下第一步獲取鎖是如何做的

RLock?lock?=?redisson.getLock("myLock");// name 就是鎖名稱 public?RLock?getLock(String?name)?{//?默認創建的同步執行器,?(存在異步執行器,?因為鎖的獲取和釋放是有強一致性要求,?默認同步)return?new?RedissonLock(connectionManager.getCommandExecutor(),?name); }

Redisson 中所有 Redis 命令都是通過 ...Executor 執行的

獲取到默認的同步執行器后, 就要初始化 RedissonLock

public?RedissonLock(CommandAsyncExecutor?commandExecutor,?String?name)?{super(commandExecutor,?name);this.commandExecutor?=?commandExecutor;//?唯一IDthis.id?=?commandExecutor.getConnectionManager().getId();//?等待獲取鎖時間this.internalLockLeaseTime?=?commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout();//?ID?+?鎖名稱this.entryName?=?id?+?":"?+?name;//?發布訂閱,?后面關于加、解鎖流程會用到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?{//?獲取當前線程IDlong?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);}//?鎖過期時間小于零,?則死等,?區分可中斷及不可中斷}?else?{if?(interruptibly)?{future.getNow().getLatch().acquire();}?else?{future.getNow().getLatch().acquireUninterruptibly();}}}}?finally?{//?取消訂閱unsubscribe(future,?threadId);} }

這一段代碼是用來執行加鎖, 繼續看下方法實現

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 異步加鎖并進行鎖續時

private?<T>?RFuture<Long>?tryAcquireAsync(long?waitTime,?long?leaseTime,?TimeUnit?unit,?long?threadId)?{//?執行?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?執行完成后觸發此操作ttlRemainingFuture.onComplete((ttlRemaining,?e)?->?{if?(e?!=?null)?{return;}//?ttlRemaining?==?null?代表獲取了鎖//?獲取到鎖后執行續時操作if?(ttlRemaining?==?null)?{scheduleExpirationRenewal(threadId);}});return?ttlRemainingFuture; }

繼續看一下 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 的應用就不繼續跟進了

加鎖 Lua

執行 Redis 加鎖的 Lua 腳本, 截個圖讓大家看一下參數以及具體含義

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?命令發現?KEYS[1]?不存在并新建一個?hash#?ARGV[2]?就作為?hash?的第一個key,?val?為?1#?相當于執行了?hincrby?myLock?91089b45...?1redis.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?+1redis.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 腳本加鎖的流程畫圖如下:

現在回過頭看一下獲取到鎖之后, 是如何為鎖進行延期操作的

鎖續時

之前有和軍哥聊過這個話題, 他說的思路和 Redisson 中體現的基本一致

說一下 Redisson 的具體實現思路吧, 中文翻譯叫做 "看門狗"

1、獲取到鎖之后執行 "看門狗" 流程

2、使用 Netty 的 Timeout 實現定時延時

3、比如鎖過期 30 秒, 每過 1/3 時間也就是 10 秒會檢查鎖是否存在, 存在則更新鎖的超時時間

可能會有小伙伴會提出這么一個疑問, 如果檢查返回存在, 設置鎖過期時剛好鎖被釋放了怎么辦?

有這樣的疑問, 代表確實用心去考慮所有可能發生的情況了, 但是不必擔心哈

Redisson 中使用的 Lua 腳本做的檢查及設置過期時間操作, 本身是原子性的不會出現上面情況

“

如果不想要引用 Netty 的包, 使用延時隊列等包工具也是可以完成 "看門狗"

這里也貼一哈相關代碼, 能夠讓小伙伴更直觀的了解如何鎖續時的

我可真是個暖男, 上代碼 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?執行完成后觸發此操作ttlRemainingFuture.onComplete((ttlRemaining,?e)?->?{if?(e?!=?null)?{return;}//?獲取到鎖后執行續時操作if?(ttlRemaining?==?null)?{scheduleExpirationRenewal(threadId);}});return?ttlRemainingFuture; }

可以看到續時方法將 threadId 當作標識符進行續時

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()?{@Overridepublic?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 續時取消, 并返回成功

@Override public?RFuture<Void>?unlockAsync(long?threadId)?{RPromise<Void>?result?=?new?RedissonPromise<Void>();RFuture<Boolean>?future?=?unlockInnerAsync(threadId);future.onComplete((opStatus,?e)?->?{//?取消自動續時功能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

老規矩, 圖片加參數說明

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, 但是還是沒有解決 主從節點下異步同步數據導致鎖丟失問題

所以 Redis 作者 Antirez 推出 紅鎖算法, 這個算法的精髓就是: 沒有從節點, 如果部署多臺 Redis, 各實例之間相互獨立, 不存在主從復制或者其他集群協調機制

如何使用

創建多個 Redisson Node, 由這些無關聯的 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 定理, 指的是在一個分布式系統中, ?Consistency（一致性）、 Availability（可用性）、Partition tolerance（分區容錯性）, 三者不可得兼

一致性(C) : 在分布式系統中的所有數據備份, 在同一時刻是否同樣的值(等同于所有節點訪問同一份最新的數據副本)

可用性(A): 在集群中一部分節點故障后, 集群整體是否還能響應客戶端的讀寫請求(對數據更新具備高可用性)

分區容忍性(P): 以實際效果而言, 分區相當于對通信的時限要求. 系統如果不能在時限內達成數據一致性, 就意味著發生了分區的情況, 必須就當前操作在 C 和 A 之間做出選擇

分布式鎖選型

如果要滿足上述分布式鎖之間的強一致性, 可以采用 Zookeeper 的分布式鎖, 因為它底層的 ZAB協議(原子廣播協議), 天然滿足 CP

但是這也意味著性能的下降, 所以不站在具體數據下看 Redis 和 Zookeeper, 代表著性能和一致性的取舍

如果項目沒有強依賴 ZK, 使用 Redis 就好了, 因為現在 Redis 用途很廣, 大部分項目中都引用了 Redis

沒必要對此再引入一個新的組件, 如果業務場景對于 Redis 異步方式的同步數據造成鎖丟失無法忍受, 在業務層處理就好了

往期推薦

Redis為什么變慢了？一文詳解Redis性能問題 | 萬字長文

Redis 消息隊列的三種方案（List、Streams、Pub/Sub）

硬核Redis總結，看這篇就夠了！

總結

以上是生活随笔為你收集整理的一文掌握Redisson分布式锁原理|干货推荐的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。