最近的自己,一直都在做些老年的技術(shù),沒有啥升級,自己也快麻木了,自己該怎么說,那必須行動起來啊!~來來,我們一起增長自己的內(nèi)功

分布式鎖的最強(qiáng)實(shí)現(xiàn): Redisson

1.概念

在介紹之前,我們要知道這個(gè)Redisson是啥? 難道就是Redis的son?(我第一次就這么認(rèn)為的哈哈!) 事實(shí)也的確如此–>看下面解釋

首先來點(diǎn)專業(yè)點(diǎn)解釋:

1. Redisson是In-momery data Grid(建立在Redis基礎(chǔ)上的java駐內(nèi)存網(wǎng)格)(一種針對分布式的緩沖技術(shù)_挺棒的!)

2. 通訊基于Netty,面向企業(yè)級開發(fā)

3. 提供一系列的分布式java常用對象,以及分布式服務(wù)等

   大白話:

   ? a.就是特么封裝許多分布式常用功能的分布式工具,

   ? b.簡化開發(fā)者開發(fā),更專注于業(yè)務(wù)開發(fā) ,而不是話大多時(shí)間的Redis上面

   ? c.避免分布式造輪子

2.分布式鎖

這玩意兒我也說下吧?就是并發(fā)業(yè)務(wù)的剛需;俗話說: 并發(fā)好不好,就看這把鎖管的好不好!

有啥example呢? (有啊!騷等~ ) 我們就以Redis來寫一個(gè)簡單的分布式鎖:~

編寫一個(gè)最簡單的Redis分布式鎖: 采用SpringDataRedis的RedisTemplate

setIfAbsent:setNx+expire

2.1加鎖操作/解鎖操作

// 加鎖
public Boolean tryLock(String key, String value, long timeout, TimeUnit unit) {return redisTemplate.opsForValue().setIfAbsent(key, value, timeout, unit);
}
// 解鎖，防止刪錯(cuò)別人的鎖，以uuid為value校驗(yàn)是否自己的鎖
public void unlock(String lockName, String uuid) {if(uuid.equals(redisTemplate.opsForValue().get(lockName)){        redisTemplate.opsForValue().del(lockName);    }
}
// 結(jié)構(gòu)
if(tryLock){// todo  
}finally{unlock;
}       

上面的代碼完成鎖操作,但是我們會發(fā)現(xiàn)無法保證原子性,一旦并發(fā)量高了,就要好家伙了!~

咋解決呢?咋保證原子性啊?我咋知道啊!(哈哈!我們往下讀,自會柳暗花明)我們引入Redis內(nèi)置的語言:lua,

2.2采用lua腳本

我們使用lua的 ,將操作封裝成一個(gè)lua腳本,通過Redis的eval/evalsha

lua腳本代碼:

腳本名稱: lockDel.lua

if redis.call('get', KEYS[1]) == ARGV[1] then -- 執(zhí)行刪除操作return redis.call('del', KEYS[1]) else -- 不成功，返回0return 0 
end

刪除java代碼:

// 解鎖腳本
DefaultRedisScript<Object> unlockScript = new DefaultRedisScript();
unlockScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("lockDel.lua")));// 執(zhí)行l(wèi)ua腳本解鎖
redisTemplate.execute(unlockScript, Collections.singletonList(keyName), value);

寫到這兒,我們也添加了lua,也能保證原子性了吧!但是你看看是不是少了啥?(哈哈,你心里是不是在說,我看出來個(gè)錘子啊!_沒毛病啊)

嗯嗯,那我提個(gè)醒—按照上面的思路:如果一個(gè)線程多次拿鎖呢?好家伙_{是不是來問題了?怎么解決呢?(想沒想到?都往下看吧!}哈哈)

2.3可重入鎖

怎么保證可重入?

首先: 我們應(yīng)該明白可重入的核心:同一個(gè)線程多次獲取同一把鎖是許可的,不會產(chǎn)生死鎖

嗯~道理沒怎么懂,我們用Synchronized的偏向鎖來理解其實(shí)現(xiàn)思路

1. synchronized實(shí)現(xiàn)重入是在JVM層面,java對象頭MARK word 中含有線程ID和計(jì)數(shù)器對線程做重入判斷,從而避免每次的CAS
2. 當(dāng)一個(gè)線程訪問同步塊并獲取鎖時(shí),會在對象頭和棧幀的鎖記錄里存儲偏向的線程ID
3. 這樣后面此線程ID再進(jìn)入和退出同步塊時(shí),就不用CAS來進(jìn)行加鎖和解鎖啦
4. 只需要檢測MARK WORD對象頭里面是否存儲指向當(dāng)前線程的偏向鎖
   • 測試成功:線程獲得鎖
   • 測試失敗:再測試下MARK WORD偏向鎖標(biāo)志是否設(shè)置為1
     • 0: 沒有CAS競爭
     • 1:CAS將對象頭偏向鎖指向當(dāng)前線程
5. 還有一個(gè)計(jì)數(shù)器:同個(gè)線程進(jìn)入則自增1,離開再減1,直到為0釋放

根據(jù)上面的思路:(我們對要實(shí)現(xiàn)的lua腳本進(jìn)行改造)

1.準(zhǔn)備形參(執(zhí)行的條件)

需要存儲鎖名稱lockName
該鎖線程ID
對應(yīng)線程進(jìn)入的次數(shù): count

2.加鎖(每次線程獲取鎖時(shí),判斷是否已經(jīng)存在該鎖)

不存在:設(shè)置hash的Key為線程ID,value=1設(shè)置expireTime(過期時(shí)間)返回獲取鎖,成功為true
存在:繼續(xù)判斷是否存在當(dāng)前線程ID的hash key存在:線程key的value+1,重入次數(shù)加1(count++),設(shè)置expireTime不存在:返回加鎖失敗

3.解鎖 —>每次線程來解鎖時(shí),判斷是否已經(jīng)存在該鎖

存在:是否有該線程的ID的hash key,有則減1,沒有返回解鎖失敗減1后:判斷剩余count為不為0=0: 不再需要這把鎖,執(zhí)行del命令刪除

1.存儲結(jié)構(gòu):

Redis采用HASH結(jié)構(gòu)

鎖名稱: lock_name1`

key: thread_id (唯一鍵,線程ID)

value:count` (計(jì)數(shù)器)

hset  lock_name1 thread_id  1hget  lock_name1

2.計(jì)數(shù)器加減

當(dāng)同一個(gè)線程獲取同一把鎖時(shí),我們需要對應(yīng)線程的計(jì)數(shù)器count做加減

怎么判斷一個(gè)Redis的key是否存在,

我們可以用exists/hexists判斷一個(gè)hash的key是否存在,

hset  lock_name1  thread_id  1 
exists/exists  lock_name1

hash的自增命令:

hincrby   lock_name1  thread_id  1

3.解鎖的判斷

當(dāng)一把鎖不再被需要了,每解鎖一次,count減1,直到為0,執(zhí)行刪除

最終的LUA代碼

加鎖lock.lua

local key = KEYS[1];
local threadId = ARGV[1];
local releaseTime = ARGV[2];-- lockname不存在
if(redis.call('exists', key) == 0) thenredis.call('hset', key, threadId, '1');redis.call('expire', key, releaseTime);return 1;
end;-- 當(dāng)前線程已id存在
if(redis.call('hexists', key, threadId) == 1) thenredis.call('hincrby', key, threadId, '1');redis.call('expire', key, releaseTime);return 1;
end;
return 0;

解鎖unlock.lua

local key = KEYS[1];
local threadId = ARGV[1];-- lockname、threadId不存在
if (redis.call('hexists', key, threadId) == 0) thenreturn nil;
end;-- 計(jì)數(shù)器-1
local count = redis.call('hincrby', key, threadId, -1);-- 刪除lock
if (count == 0) thenredis.call('del', key);return nil;
end;

RedisLock.java: 實(shí)現(xiàn)分布式鎖

實(shí)現(xiàn)功能: 互斥,可重入,防死鎖

/*** @description 原生redis實(shí)現(xiàn)分布式鎖**/
@Getter
@Setter
public class RedisLock {private RedisTemplate redisTemplate;private DefaultRedisScript<Long> lockScript;private DefaultRedisScript<Object> unlockScript;public RedisLock(RedisTemplate redisTemplate) {this.redisTemplate = redisTemplate;// 加載加鎖的腳本lockScript = new DefaultRedisScript<>();this.lockScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("lock.lua")));this.lockScript.setResultType(Long.class);// 加載釋放鎖的腳本unlockScript = new DefaultRedisScript<>();this.unlockScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("unlock.lua")));}/*** 獲取鎖*/public String tryLock(String lockName, long releaseTime) {// 存入的線程信息的前綴String key = UUID.randomUUID().toString();// 執(zhí)行腳本Long result = (Long) redisTemplate.execute(lockScript,Collections.singletonList(lockName),key + Thread.currentThread().getId(),releaseTime);if (result != null && result.intValue() == 1) {return key;} else {return null;}}/*** 解鎖* @param lockName* @param key*/public void unlock(String lockName, String key) {redisTemplate.execute(unlockScript,Collections.singletonList(lockName),key + Thread.currentThread().getId());}
}

上面代碼,已經(jīng)實(shí)現(xiàn)了大部分的功能,對于一般的場景都是可以從容應(yīng)對了,但是(哈哈哈,最怕但是了是不?)

針對特殊場景:

1.若A進(jìn)程在獲取到鎖的時(shí)候,因?yàn)闃I(yè)務(wù)操作時(shí)間太長,鎖釋放了,但是業(yè)務(wù)還是在執(zhí)行,此刻B 進(jìn)程有可以正常拿到鎖進(jìn)行業(yè)務(wù)操作,這時(shí)候就會出現(xiàn)A,B進(jìn)程共享資源的問題

2.若負(fù)責(zé)存儲這個(gè)分布式鎖的Redis宕機(jī),兒此時(shí)這個(gè)鎖正好處于鎖住狀態(tài),這時(shí)就會造成鎖死的狀態(tài)

那對于這種情況時(shí),我們需要采用鎖續(xù)約

鎖續(xù)約:延長鎖的ReleaseTime直到完成業(yè)務(wù)期望結(jié)果,本質(zhì)就是不斷延長鎖過期時(shí)間

但是,對于當(dāng)中的處理續(xù)約,性能比如鎖的最大等待時(shí)間,無效的鎖申請,與以及失敗重試機(jī)制等等我們寫到猴年馬月啊!~立即推=—>放棄哈哈哈!!

別別別~來這時(shí)候我們就要介紹我們今天的主角:Reidsson!!(解決上面所有問題)

4.Redisson分布式鎖

兵馬未動,糧草先行

我們看看咋使用啊~是不是

4.1引入依賴

<dependency><groupId>org.redisson</groupId><artifactId>redisson</artifactId><version>3.13.6</version>
</dependency><!-- 另一種Spring集成starter，本章未使用 -->
<dependency><groupId>org.redisson</groupId><artifactId>redisson-spring-boot-starter</artifactId><version>3.13.6</version>
</dependency>

4.2配置

@Configuration
public class RedissionConfig {@Value("${spring.redis.host}")private String redisHost;@Value("${spring.redis.password}")private String password;private int port = 6379;@Beanpublic RedissonClient getRedisson() {Config config = new Config();config.useSingleServer().setAddress("redis://" + redisHost + ":" + port).setPassword(password);config.setCodec(new JsonJacksonCodec());return Redisson.create(config);}
}

4.3啟用分布式鎖

簡潔明了，只需要一個(gè)RLock,如下面的代碼所示

@Resource
private RedissonClient redissonClient;RLock rLock = redissonClient.getLock(lockName);
try {boolean isLocked = rLock.tryLock(expireTime, TimeUnit.MILLISECONDS);if (isLocked) {// TODO}} catch (Exception e) {rLock.unlock();}

4.4RLock

還在更…