一、RLock接口
1、概念
public interface RLock extends Lock, RExpirable, RLockAsync很明顯RLock是繼承Lock鎖，所以他有Lock鎖的所有特性，比如lock、unlock、trylock等特性,同時它還有很多新特性：強制鎖釋放，帶有效期的鎖,。
2、RLock鎖API
這里針對上面做個整理，這里列舉幾個常用的接口說明
public interface RLock {
//----------------------Lock接口方法-----------------------
/**
* 加鎖 鎖的有效期默認30秒
*/
void lock();
/**
* tryLock()方法是有返回值的，它表示用來嘗試獲取鎖，如果獲取成功，則返回true，如果獲取失?。存i已被其他線程獲?。?，則返回false .
*/
boolean tryLock();
/**
* tryLock(long time, TimeUnit unit)方法和tryLock()方法是類似的，只不過區(qū)別在于這個方法在拿不到鎖時會等待一定的時間，
* 在時間期限之內(nèi)如果還拿不到鎖，就返回false。如果如果一開始拿到鎖或者在等待期間內(nèi)拿到了鎖，則返回true。
*
* @param time 等待時間
* @param unit 時間單位 小時、分、秒、毫秒等
*/
boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
/**
* 解鎖
*/
void unlock();
/**
* 中斷鎖 表示該鎖可以被中斷 假如A和B同時調(diào)這個方法，A獲取鎖，B為獲取鎖，那么B線程可以通過
* Thread.currentThread().interrupt(); 方法真正中斷該線程
*/
void lockInterruptibly();
//----------------------RLock接口方法-----------------------
/**
* 加鎖 上面是默認30秒這里可以手動設(shè)置鎖的有效時間
*
* @param leaseTime 鎖有效時間
* @param unit 時間單位 小時、分、秒、毫秒等
*/
void lock(long leaseTime, TimeUnit unit);
/**
* 這里比上面多一個參數(shù)，多添加一個鎖的有效時間
*
* @param waitTime 等待時間
* @param leaseTime 鎖有效時間
* @param unit 時間單位 小時、分、秒、毫秒等
*/
boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException;
/**
* 檢驗該鎖是否被線程使用，如果被使用返回True
*/
boolean isLocked();
/**
* 檢查當前線程是否獲得此鎖（這個和上面的區(qū)別就是該方法可以判斷是否當前線程獲得此鎖，而不是此鎖是否被線程占有）
* 這個比上面那個實用
*/
boolean isHeldByCurrentThread();
/**
* 中斷鎖 和上面中斷鎖差不多，只是這里如果獲得鎖成功,添加鎖的有效時間
* @param leaseTime 鎖有效時間
* @param unit 時間單位 小時、分、秒、毫秒等
*/
void lockInterruptibly(long leaseTime, TimeUnit unit);
}RLock相關(guān)接口，主要是新添加了?leaseTime?屬性字段，主要是用來設(shè)置鎖的過期時間,避免死鎖。
二、RedissonLock實現(xiàn)類
public class RedissonLock extends RedissonExpirable implements RLockRedissonLock實現(xiàn)了RLock接口，所以實現(xiàn)了接口的具體方法。這里我列舉幾個方法說明下
1、void lock()方法
@Override
public void lock() {
try {
lockInterruptibly();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}發(fā)現(xiàn)lock鎖里面進去其實用的是lockInterruptibly（中斷鎖，表示可以被中斷）,而且捕獲異常后用 Thread.currentThread().interrupt()來真正中斷當前線程，其實它們是搭配一起使用的。
具體有關(guān)lockInterruptibly()方法講解推薦一個博客。博客：Lock的lockInterruptibly()
接下來執(zhí)行流程,這里理下關(guān)鍵幾步
/**
* 1、帶上默認值調(diào)另一個中斷鎖方法
*/
@Override
public void lockInterruptibly() throws InterruptedException {
lockInterruptibly(-1, null);
}
/**
* 2、另一個中斷鎖的方法
*/
void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException
/**
* 3、這里已經(jīng)設(shè)置了鎖的有效時間默認為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)
接口的參數(shù)和含義上面已經(jīng)說過了，現(xiàn)在我們開看下源碼，這里只顯示一些重要邏輯。
@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、訂閱監(jiān)聽redis消息，并且創(chuàng)建RedissonLockEntry，其中RedissonLockEntry中比較關(guān)鍵的是一個 Semaphore屬性對象,用來控制本地的鎖請求的信號量同步，返回的是netty框架的Future實現(xiàn)。
final RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
// 阻塞等待subscribe的future的結(jié)果對象，如果subscribe方法調(diào)用超過了time，說明已經(jīng)超過了客戶端設(shè)置的最大wait time，則直接返回false，取消訂閱，不再繼續(xù)申請鎖了。
// 只有await返回true，才進入循環(huán)嘗試獲取鎖
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一直獲取鎖。最終只會有兩種結(jié)果
//1)、在等待時間內(nèi)獲取鎖成功 返回true。2）等待時間結(jié)束了還沒有獲取到鎖那么返回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)。因為從性能上考慮，在高并發(fā)情況下后者效率是前者的好幾倍
3、unlock()
解鎖的邏輯很簡單。
@Override
public void unlock() {
// 1.通過 Lua 腳本執(zhí)行 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.釋放鎖后取消刷新鎖失效時間的調(diào)度任務(wù)
if (opStatus) {
cancelExpirationRenewal();
}
}使用 EVAL 命令執(zhí)行 Lua 腳本來釋放鎖：文章來源：http://www.zghlxwxcb.cn/news/detail-560066.html
- key 不存在，說明鎖已釋放，直接執(zhí)行?
publish?命令發(fā)布釋放鎖消息并返回?1。 - key 存在，但是 field 在 Hash 中不存在，說明自己不是鎖持有者，無權(quán)釋放鎖，返回?
nil。 - 因為鎖可重入，所以釋放鎖時不能把所有已獲取的鎖全都釋放掉，一次只能釋放一把鎖，因此執(zhí)行?
hincrby?對鎖的值減一。 - 釋放一把鎖后，如果還有剩余的鎖，則刷新鎖的失效時間并返回?
0；如果剛才釋放的已經(jīng)是最后一把鎖，則執(zhí)行?del?命令刪除鎖的 key，并發(fā)布鎖釋放消息，返回?1。
注意這里有個實際開發(fā)過程中，容易出現(xiàn)很容易出現(xiàn)上面第二步異常，非鎖的持有者釋放鎖時拋出異常。比如下面這種情況文章來源地址http://www.zghlxwxcb.cn/news/detail-560066.html
//設(shè)置鎖1秒過去
redissonLock.lock("redisson", 1);
/**
* 業(yè)務(wù)邏輯需要咨詢2秒
*/
redissonLock.release("redisson");
/**
* 線程1 進來獲得鎖后，線程一切正常并沒有宕機，但它的業(yè)務(wù)邏輯需要執(zhí)行2秒，這就會有個問題，在 線程1 執(zhí)行1秒后，這個鎖就自動過期了，
* 那么這個時候 線程2 進來了。在線程1去解鎖就會拋上面這個異常（因為解鎖和當前鎖已經(jīng)不是同一線程了）
*/到了這里，關(guān)于微服務(wù)系列文章之 Redisson實現(xiàn)分布式鎖(2)的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
