Redisson實現分散式鎖—RedissonLock
阿新 • • 發佈：2019-06-20
Redisson實現分散式鎖—RedissonLock
有關Redisson實現分散式鎖上一篇部落格講了分散式的鎖原理：Redisson實現分散式鎖---原理
這篇主要講RedissonLock和RLock。Redisson分散式鎖的實現是基於RLock介面，RedissonLock實現RLock介面。
一、RLock介面
1、概念
public interface RLock extends Lock, RExpirable, RLockAsync很明顯RLock是繼承Lock鎖，所以他有Lock鎖的所有特性，比如lock、unlock、trylock等特性,同時它還有很多新特性：強制鎖釋放，帶有效期的鎖,。
2、RLock鎖API
這裡針對上面做個整理，這裡列舉幾個常用的介面說明
public interface RRLock { //----------------------Lock介面方法----------------------- /** * 加鎖 鎖的有效期預設30秒 */ void lock(); /** * tryLock()方法是有返回值的，它表示用來嘗試獲取鎖，如果獲取成功，則返回true，如果獲取失敗（即鎖已被其他執行緒獲取），則返回false . */ boolean tryLock(); /** * tryLock(long time, TimeUnit unit)方法和tryLock()方法是類似的，只不過區別在於這個方法在拿不到鎖時會等待一定的時間， * 在時間期限之內如果還拿不到鎖，就返回false。如果如果一開始拿到鎖或者在等待期間內拿到了鎖，則返回true。 * * @param time 等待時間 * @param unit 時間單位 小時、分、秒、毫秒等 */ boolean tryLock(long time, TimeUnit unit) throws InterruptedException; /** * 解鎖 */ void unlock(); /** * 中斷鎖 表示該鎖可以被中斷 假如A和B同時調這個方法，A獲取鎖，B為獲取鎖，那麼B執行緒可以通過 * Thread.currentThread().interrupt(); 方法真正中斷該執行緒 */ void lockInterruptibly(); //----------------------RLock介面方法----------------------- /** * 加鎖 上面是預設30秒這裡可以手動設定鎖的有效時間 * * @param leaseTime 鎖有效時間 * @param unit 時間單位 小時、分、秒、毫秒等 */ void lock(long leaseTime, TimeUnit unit); /** * 這裡比上面多一個引數，多新增一個鎖的有效時間 * * @param waitTime 等待時間 * @param leaseTime 鎖有效時間 * @param unit 時間單位 小時、分、秒、毫秒等 */ boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException; /** * 檢驗該鎖是否被執行緒使用，如果被使用返回True */ boolean isLocked(); /** * 檢查當前執行緒是否獲得此鎖（這個和上面的區別就是該方法可以判斷是否當前執行緒獲得此鎖，而不是此鎖是否被執行緒佔有） * 這個比上面那個實用 */ boolean isHeldByCurrentThread(); /** * 中斷鎖 和上面中斷鎖差不多，只是這裡如果獲得鎖成功,新增鎖的有效時間 * @param leaseTime 鎖有效時間 * @param unit 時間單位 小時、分、秒、毫秒等 */ void lockInterruptibly(long leaseTime, TimeUnit unit); }
RLock相關介面，主要是新添加了 leaseTime 屬性欄位，主要是用來設定鎖的過期時間,避免死鎖。
二、RedissonLock實現類
public class RedissonLock extends RedissonExpirable implements RLockRedissonLock實現了RLock介面，所以實現了介面的具體方法。這裡我列舉幾個方法說明下
1、void lock()方法
@Override public void lock() { try { lockInterruptibly(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }
發現lock鎖裡面進去其實用的是lockInterruptibly（中斷鎖，表示可以被中斷）,而且捕獲異常後用 Thread.currentThread().interrupt()來真正中斷當前執行緒，其實它們是搭配一起使用的。
具體有關lockInterruptibly()方法講解推薦一個部落格。部落格：Lock的lockInterruptibly()
接下來執行流程,這裡理下關鍵幾步
/**
* 1、帶上預設值調另一箇中斷鎖方法
*/
@Override
public void lockInterruptibly() throws InterruptedException {
lockInterruptibly(-1, null);
}
/**
* 2、另一箇中斷鎖的方法
*/
void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException
/**
* 3、這裡已經設定了鎖的有效時間預設為30秒 （commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout()=30）
*/
RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
/**
* 4、最後通過lua指令碼訪問Redis,保證操作的原子性
*/
<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
internalLockLeaseTime = unit.toMillis(leaseTime);
return commandExecutor.evalWriteAsync(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.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}那麼void lock(long leaseTime, TimeUnit unit)方法其實和上面很相似了，就是從上面第二步開始的。
2、tryLock(long waitTime, long leaseTime, TimeUnit unit)
介面的引數和含義上面已經說過了，現在我們開看下原始碼，這裡只顯示一些重要邏輯。
@Override
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
long time = unit.toMillis(waitTime);
long current = System.currentTimeMillis();
long threadId = Thread.currentThread().getId();
Long ttl = tryAcquire(leaseTime, unit, threadId);
//1、 獲取鎖同時獲取成功的情況下，和lock(...)方法是一樣的 直接返回True，獲取鎖False再往下走
if (ttl == null) {
return true;
}
//2、如果超過了嘗試獲取鎖的等待時間,當然返回false 了。
time -= System.currentTimeMillis() - current;
if (time <= 0) {
acquireFailed(threadId);
return false;
}
// 3、訂閱監聽redis訊息，並且建立RedissonLockEntry，其中RedissonLockEntry中比較關鍵的是一個 Semaphore屬性物件,用來控制本地的鎖請求的訊號量同步，返回的是netty框架的Future實現。
final RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
// 阻塞等待subscribe的future的結果物件，如果subscribe方法呼叫超過了time，說明已經超過了客戶端設定的最大wait time，則直接返回false，取消訂閱，不再繼續申請鎖了。
// 只有await返回true，才進入迴圈嘗試獲取鎖
if (!await(subscribeFuture, time, TimeUnit.MILLISECONDS)) {
if (!subscribeFuture.cancel(false)) {
subscribeFuture.addListener(new FutureListener<RedissonLockEntry>() {
@Override
public void operationComplete(Future<RedissonLockEntry> future) throws Exception {
if (subscribeFuture.isSuccess()) {
unsubscribe(subscribeFuture, threadId);
}
}
});
}
acquireFailed(threadId);
return false;
}
//4、如果沒有超過嘗試獲取鎖的等待時間，那麼通過While一直獲取鎖。最終只會有兩種結果
//1)、在等待時間內獲取鎖成功 返回true。2）等待時間結束了還沒有獲取到鎖那麼返回false。
while (true) {
long currentTime = System.currentTimeMillis();
ttl = tryAcquire(leaseTime, unit, threadId);
// 獲取鎖成功
if (ttl == null) {
return true;
}
// 獲取鎖失敗
time -= System.currentTimeMillis() - currentTime;
if (time <= 0) {
acquireFailed(threadId);
return false;
}
}
}重點 tryLock一般用於特定滿足需求的場合，但不建議作為一般需求的分散式鎖，一般分散式鎖建議用void lock(long leaseTime, TimeUnit unit)。因為從效能上考慮，在高併發情況下後者效率是前者的好幾倍
3、unlock()
解鎖的邏輯很簡單。
@Override
public void unlock() {
// 1.通過 Lua 指令碼執行 Redis 命令釋放鎖
Boolean opStatus = commandExecutor.evalWrite(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.<Object>asList(getName(), getChannelName()),
LockPubSub.unlockMessage, internalLockLeaseTime,
getLockName(Thread.currentThread().getId()));
// 2.非鎖的持有者釋放鎖時丟擲異常
if (opStatus == null) {
throw new IllegalMonitorStateException(
"attempt to unlock lock, not locked by current thread by node id: "
+ id + " thread-id: " + Thread.currentThread().getId());
}
// 3.釋放鎖後取消重新整理鎖失效時間的排程任務
if (opStatus) {
cancelExpirationRenewal();
}
}使用 EVAL 命令執行 Lua 指令碼來釋放鎖：
- key 不存在，說明鎖已釋放，直接執行
publish命令釋出釋放鎖訊息並返回1。 - key 存在，但是 field 在 Hash 中不存在，說明自己不是鎖持有者，無權釋放鎖，返回
nil。 - 因為鎖可重入，所以釋放鎖時不能把所有已獲取的鎖全都釋放掉，一次只能釋放一把鎖，因此執行
hincrby對鎖的值減一。 - 釋放一把鎖後，如果還有剩餘的鎖，則重新整理鎖的失效時間並返回
0；如果剛才釋放的已經是最後一把鎖，則執行del命令刪除鎖的 key，併發布鎖釋放訊息，返回1。
注意這裡有個實際開發過程中，容易出現很容易出現上面第二步異常，非鎖的持有者釋放鎖時丟擲異常。比如下面這種情況
//設定鎖1秒過去
redissonLock.lock("redisson", 1);
/**
* 業務邏輯需要諮詢2秒
*/
redissonLock.release("redisson");
/**
* 執行緒1 進來獲得鎖後，執行緒一切正常並沒有宕機，但它的業務邏輯需要執行2秒，這就會有個問題，在 執行緒1 執行1秒後，這個鎖就自動過期了，
* 那麼這個時候 執行緒2 進來了。線上程1去解鎖就會拋上面這個異常（因為解鎖和當前鎖已經不是同一執行緒了）
*/
只要自己變優秀了，其他的事情才會跟著好起來（中將6）