Redis--day9--黑馬點評--分布式鎖（二）

請添加圖片描述
（以下所有內容全部來自上述課程）
在這里插入圖片描述

分布式鎖

1. Redisson功能介紹

基于setnx實現的分布式鎖存在下面的問題：

不可重入：同一個線程無法多次獲取同一把鎖
不可重試：獲取鎖只嘗試一次就返回false，沒有重試機制
超時釋放：鎖超時釋放雖然可以避免死鎖，但如果是業務執行耗時較長，也會導致鎖釋放，存在安全隱患
主從一致性：如果Redis提供了主從集群，主從同步存在延遲，當主宕機時，如果從并同步主中的鎖數據，則會出現鎖實現

Redisson是一個在Redis的基礎上實現的lava駐內存數據網格(In-Memory Data Grid)。它不僅提供了一系列的分布式的Java常用對象，還提供了許多分布式服務，其中就包含了各種分布式鎖的實現。
在這里插入圖片描述
官網地址:https://redisson.org
GitHub地址:https://github.com/redisson/redisson
jar包下載：https://mvnrepository.com/artifact/org.redisson/redisson/3.50.0

2. Redis入門

引入依賴:

<dependency>
<groupId>org.redisson</groupId><artifactId>redisson</artifactId>
<version>3.13.6</version>
</dependency>

配置Redisson客戶端:

@Configuration
public class RedisConfig {@Beanpublic RedissonClient redissonClient(){// 配置類Config config =new Config();//添加redis地址，這里添加了單點的地址，也可以使用config.useClusterServers()添加集群地址	config.useSingleServer().setAddress("redis://192.168.150.101:6379").setPassword("123321");// 創建客戶端return Redisson.create(config);}
}

使用Redisson的分布式鎖

@Resource
private RedissonClient redissonClient;
@Test
void testRedisson()throws InterruptedException {// 獲取鎖(可重入)，指定鎖的名稱RLock lock=redissonClient.getLock("anyLock");//嘗試獲取鎖，參數分別是:獲取鎖的最大等待時間(期間會重試)，鎖自動釋放時間，時間單位boolean isLock =lock.tryLock(1，10,TimeUnit.SECONDS);// 判斷釋放獲取成功if(isLock){try {System.out.println("執行業務");}finally {// 釋放鎖lock.unlock();}}
}

3. Redisson的可重入鎖原理

// 創建鎖對象
RLock lock=redissonClient.getLock("lock");@Test
void method1(){boolean isLock = lock.tryLock();if(!isLock){log.error("獲取鎖失敗,1");return;}try{log.info("獲取鎖成功,1");method2();} finally {log.info("釋放鎖,1");lock.unlock();}
}
void method2(){boolean isLock =lock.tryLock();if(!isLock){log.error("獲取鎖失敗,2");return;}try {log.info("獲取鎖成功,2");} finally {log.info("釋放鎖,2");lock.unlock();}
}

請添加圖片描述
獲取鎖的lua腳本：

local key = KEYS[1]; -- 鎖的key
local threadId= ARGV[1];-- 線程唯一標識
local releaseTime =ARGV[2];-- 鎖的自動釋放時間
--判斷是否存在
if(redis.call('exists',key)== 0)then-- 不存在，獲取鎖redis.call('hset',key, threadId,"1');-- 設置有效期redis.call('expire',key, releaseTime);return 1;--返回結果
end ;
--鎖已經存在，判斷threadId是否是自己
if(redis.call("hexists",key,threadId)== 1) then--不存在，獲取鎖，重入次數+1redis.call("hincrby", key, threadId, '1');--設置有效期redis.call('expire',key, releaseTime);return 1;--返回結果
end ;
return 0; --代碼走到這里，說明獲取鎖的不是自己，獲取鎖失敗

釋放鎖的lua腳本：

local key =KEYS[1];-- 鎖的key
local threadId= ARGV[1];-- 線程唯一標識
local releseTime=ARGV[2];--鎖的自動釋放時間
--判斷當前鎖是否還是被自己持有
if(redis.call("HEXISTs",key,threadId)==0)thenreturn nil;--如果已經不是自己，則直接返回
end ;
--是自己的鎖，則重入次數-1
local count = redis.call('HINCRBY', key, threadId, -1);
--判斷是否重入次數是否已經為0
if(count>0)then--大于0說明不能釋放鎖，重置有效期然后返回redis.call('EXPIRE',key, releaseTime);return nil;
else--等于0說明可以釋放鎖，直接刪除redis.call('DEL', key);return nil;
end ;

4. Redisson的鎖重試和WatchDog機制

tryLock ctrl+alt+B 打開源碼

   private Long tryAcquire(long waitTime, long leaseTime, TimeUnit unit, long threadId) { //第三處查看//阻塞等待Future結果，就是剩余有效期 把Long接著返回-->第二處查看（往下滑）return get(tryAcquireAsync0(waitTime, leaseTime, unit, threadId));  //查看tryAcquireAsync0（往下滑）}private RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {  //第四處查看RFuture<Long> ttlRemainingFuture;if (leaseTime > 0) {ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);} else {  //命令行已經發出去了，但是結果拿沒拿到還不清楚-->FuturettlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime,  //查看tryLockInnerAsync（向下滑）TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);}  //返回第三處查看（向上滑）CompletionStage<Long> s = handleNoSync(threadId, ttlRemainingFuture);ttlRemainingFuture = new CompletableFutureWrapper<>(s);CompletionStage<Long> f = ttlRemainingFuture.thenApply(ttlRemaining -> {// lock acquiredif (ttlRemaining == null) {  //剩余有效期=nullif (leaseTime > 0) {  internalLockLeaseTime = unit.toMillis(leaseTime);} else {scheduleExpirationRenewal(threadId);  //自動更新有效期續約 --> 查看scheduleExpirationRenewal（下一個板塊的代碼）}}return ttlRemaining;});return new CompletableFutureWrapper<>(f);}@Overridepublic boolean tryLock() {return get(tryLockAsync());}<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) " +"or (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]);",  //pttl（毫秒的單位）=ttl(秒的單位) 獲取指定key的剩余有效期Collections.singletonList(getRawName()), unit.toMillis(leaseTime), getLockName(threadId));}  //返回第四處（向上滑）@Overridepublic boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException { //第二處查看long time = unit.toMillis(waitTime);    //把等待時間轉換為毫秒long current = System.currentTimeMillis();  //獲取當前時間long threadId = Thread.currentThread().getId();  //獲取當前線程id 也就是線程標識Long ttl = tryAcquire(waitTime, leaseTime, unit, threadId);  //嘗試獲取鎖 -->看tryAcquire方法（往上滑）// lock acquiredif (ttl == null) {return true;  //獲取成功。直接返回}//獲取失敗，繼續嘗試（重點！）time -= System.currentTimeMillis() - current;  //現在的時間減去之前的當前時間（就是獲取鎖消耗的時間）//然后最大等待時間減去消耗的時間，就是剩余等待時間if (time <= 0) {acquireFailed(waitTime, unit, threadId);return false;  //沒有剩余等待時間，就直接獲取失敗}//獲取失敗，繼續嘗試（但沒有立即-->subscribe）current = System.currentTimeMillis();CompletableFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);  //訂閱了別人釋放鎖的信號（publish）try {subscribeFuture.get(time, TimeUnit.MILLISECONDS);  //因為時間不確定，所以也是Future} catch (TimeoutException e) {if (!subscribeFuture.completeExceptionally(new RedisTimeoutException(  //最大剩余時間結束-->返回false，繼續往下走"Unable to acquire subscription lock after " + time + "ms. " +"Try to increase 'subscriptionsPerConnection' and/or 'subscriptionConnectionPoolSize' parameters."))) {subscribeFuture.whenComplete((res, ex) -> {if (ex == null) {  //判斷有沒有時間了unsubscribe(res, threadId);  //超時，取消訂閱}});}acquireFailed(waitTime, unit, threadId);return false;  } catch (ExecutionException e) {acquireFailed(waitTime, unit, threadId);return false;  //獲取鎖失敗}try {time -= System.currentTimeMillis() - current;  //又計算一次等待消耗時間if (time <= 0) {acquireFailed(waitTime, unit, threadId);return false;  //套娃，沒時間了還是沒獲取到鎖，直接返回獲取失敗}//小總結：上方一直計算剩余時間，有就直接到這兒，沒有就一直重復計算//計算終于有剩余時間（true）while (true) {long currentTime = System.currentTimeMillis();ttl = tryAcquire(waitTime, leaseTime, unit, threadId);  //第一次重試（tryAcquire）// lock acquiredif (ttl == null) {return true;  //成功}//失敗就再計算剩余時間time -= System.currentTimeMillis() - currentTime;if (time <= 0) {acquireFailed(waitTime, unit, threadId);return false;  //沒有就失敗}// waiting for message  失敗就繼續計算時間currentTime = System.currentTimeMillis();if (ttl >= 0 && ttl < time) {  //ttl和time ，哪個小就先等哪個（time是根本底線，沒了也不用等了）//getLatch()  信號量，類似與publish拋出獲取信號//ttl是施放時間，更靈活一些，鎖釋放了就重試（重新獲取鎖）commandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);} else {  //time到期了，還沒釋放也就不用等了commandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);}//再獲取時間time -= System.currentTimeMillis() - currentTime;if (time <= 0) {  //沒時間了，還是直接失敗acquireFailed(waitTime, unit, threadId);return false;}}  //時間充足就while(true)，返回上面又開始重試} finally {unsubscribe(commandExecutor.getNow(subscribeFuture), threadId);}
//        return get(tryLockAsync(waitTime, leaseTime, unit));}  //至此，重試問題解決  -->  去看超時問題：第四處（向上滑）

  protected void scheduleExpirationRenewal(long threadId) {ExpirationEntry entry = new ExpirationEntry();//EntryName：當前鎖的名稱//一個鎖一個entry，第一次來創建新的entry，之后來的返回舊的entry//保證這把鎖不管重復了幾次，返回的都是同一個entryExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);if (oldEntry != null) {oldEntry.addThreadId(threadId);  //同一個線程多次來獲取} else {entry.addThreadId(threadId);try {renewExpiration();  //第一次來：續約，更新有效期  -->看renewExpiration} finally {if (Thread.currentThread().isInterrupted()) {cancelExpirationRenewal(threadId);}}}}
//renewExpiration：
private void renewExpiration() {  //更新有效期ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());  //先得到entryif (ee == null) {return;}Timeout task = getServiceManager().newTimeout(new TimerTask() {@Overridepublic void run(Timeout timeout) throws Exception {ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());  //拿出entryif (ent == null) {return;}Long threadId = ent.getFirstThreadId();  //取出線程idif (threadId == null) {return;}CompletionStage<Boolean> future = renewExpirationAsync(threadId);  //刷新有效期-->查看renewExpirationAsyncfuture.whenComplete((res, e) -> {if (e != null) {log.error("Can't update lock {} expiration", getRawName(), e);EXPIRATION_RENEWAL_MAP.remove(getEntryName());return;}if (res) {// reschedule itselfrenewExpiration();  //自己調用自己：遞歸--重復每十秒更新一次有效期，解決超時問題（重點！）} else {cancelExpirationRenewal(null);}});}}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);  //內部鎖施放時間/3ee.setTimeout(task);}//renewExpirationAsync：
//重置有效期protected CompletionStage<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));}//釋放鎖的邏輯：
@Overridepublic void unlock() {try {get(unlockAsync(Thread.currentThread().getId()));  //查看unlockAsync} catch (RedisException e) {if (e.getCause() instanceof IllegalMonitorStateException) {throw (IllegalMonitorStateException) e.getCause();} else {throw e;}}//unlockAsync：
@Overridepublic RFuture<Void> unlockAsync(long threadId) {return getServiceManager().execute(() -> unlockAsync0(threadId));}private RFuture<Void> unlockAsync0(long threadId) {CompletionStage<Boolean> future = unlockInnerAsync(threadId);CompletionStage<Void> f = future.handle((opStatus, e) -> {cancelExpirationRenewal(threadId);  //取消更新任務 -->查看：cancelExpirationRenewalif (e != null) {if (e instanceof CompletionException) {throw (CompletionException) e;}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);}//cancelExpirationRenewal：
//定時任務的刪除
protected void cancelExpirationRenewal(Long threadId) {ExpirationEntry task = EXPIRATION_RENEWAL_MAP.get(getEntryName());  //從map中取if (task == null) {return;}if (threadId != null) {task.removeThreadId(threadId);  //先把id干掉}if (threadId == null || task.hasNoThreads()) {Timeout timeout = task.getTimeout();if (timeout != null) {timeout.cancel();  //然后再把任務取消}EXPIRATION_RENEWAL_MAP.remove(getEntryName());  //最后再把entry取消掉}}

請添加圖片描述

小總結

Redisson分布式鎖原理

可重入:利用hash結構記錄線程id和重入次數
可重試:利用信號量和PubSub功能實現等待、喚醒，獲取鎖失敗的重試機制
超時續約:利用watchDog，每隔一段時間(releaseTime/3)，重置超時時間

5. Redisson的multiLock原理

請添加圖片描述
RedisConfig.java：

package com.hmdp.config;import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RedissionConfig {@Beanpublic RedissonClient redissonClient(){// 配置類Config config =new Config();//添加redis地址，這里添加了單點的地址，也可以使用config.useClusterServers()添加集群地址config.useSingleServer().setAddress("redis://192.168.150.101:6379").setPassword("123456");// 創建客戶端return Redisson.create(config);}@Beanpublic RedissonClient redissonClient2(){// 配置類Config config =new Config();//添加redis地址，這里添加了單點的地址，也可以使用config.useClusterServers()添加集群地址config.useSingleServer().setAddress("redis://192.168.150.101:6380");// 創建客戶端return Redisson.create(config);}@Beanpublic RedissonClient redissonClient3(){// 配置類Config config =new Config();//添加redis地址，這里添加了單點的地址，也可以使用config.useClusterServers()添加集群地址config.useSingleServer().setAddress("redis://192.168.150.101:6381");// 創建客戶端return Redisson.create(config);}
}

RedissonTest.java：

package com.hmdp;import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import org.junit.jupiter.api.BeforeEach;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.boot.test.context.SpringBootTest;import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;@Slf4j
@SpringBootTest
class RedissonTest {@Resourceprivate RedissonClient redissonClient;@Resourceprivate RedissonClient redissonClient2;@Resourceprivate RedissonClient redissonClient3;private RLock lock;@BeforeEachvoid setUp(){RLock lock1 = redissonClient.getLock("order");RLock lock2 = redissonClient2.getLock("order");RLock lock3 = redissonClient3.getLock("order");//創建聯鎖lock = redissonClient.getMultiLock(lock1,lock2,lock3);}@Testvoid method1() throws InterruptedException{//嘗試獲取鎖boolean isLock = lock.tryLock(1L, TimeUnit.SECONDS);if (!isLock){log.error("獲取鎖失敗....1");return;}try {log.info("獲取鎖成功....1");method2();log.info("開始執行業務....1");}finally {log.warn("準備釋放鎖....1");lock.unlock();}}void method2(){//嘗試獲取鎖boolean isLock = lock.tryLock();if (!isLock){log.error("獲取鎖失敗....2");return;}try {log.info("獲取鎖成功....2");log.info("開始執行業務....2");}finally {log.warn("準備釋放鎖....2");lock.unlock();}}
}

總結

不可重入Redis分布式鎖:
原理:利用setnx的互斥性;利用ex避免死鎖;釋放鎖時判斷線程標示
缺陷:不可重入、無法重試、鎖超時失效
可重入的Redis分布式鎖:
原理:利用hash結構，記錄線程標示和重入次數;利用watchDog延續鎖時間;利用信號量控制鎖重試等待
缺陷:redis宕機引起鎖失效問題
Redisson的multiLock:
原理:多個獨立的Redis節點，必須在所有節點都獲取重入鎖，才算獲取鎖成功
缺陷:運維成本高、實現復雜