Redisson實現分布式鎖(2)—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 RLock
RedissonLock實現了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）