目標(biāo)：

我們了解分布式鎖先要理解幾個(gè)問(wèn)題：

1.任何時(shí)候只有一個(gè)線程持有鎖

(相關(guān)資料圖)

2.要防止一個(gè)線程長(zhǎng)期持有鎖甚至是死鎖的情況

3.加鎖和解鎖必須是同一個(gè)進(jìn)程

4.鎖延續(xù)

Redis分布式鎖：

常見(jiàn)的分布式鎖有redis分布式鎖，zookeeper分布式鎖，本文將為大家闡述redis分布式鎖。

首先，redis分布式鎖的本質(zhì)就是在redis占一個(gè)坑位，利用的setnx命令，然后處理完其余的業(yè)務(wù)后再del。再setnx后如果有其它的線程進(jìn)來(lái)再setnx那么是set不進(jìn)去的。這就是占坑的原理。

此時(shí)第一個(gè)問(wèn)題就出現(xiàn)了：在del之前 我的業(yè)務(wù)如果出現(xiàn)了錯(cuò)誤，那么就不會(huì)去執(zhí)行del，就會(huì)出現(xiàn)死鎖的情況。

這種情況的解決方案很簡(jiǎn)單 我們只需要增加一個(gè)超時(shí)時(shí)間即可。比如設(shè)置超時(shí)時(shí)間10s鎖將會(huì)自動(dòng)釋放。在redis2.8之后 setnx和expire是原子操作 我們不用考慮setnx后因?yàn)楦鞣N問(wèn)題沒(méi)有expire的情況。

那么現(xiàn)在就會(huì)有第二個(gè)問(wèn)題：鎖超時(shí)問(wèn)題。

Redisson分布式鎖這邊我們使用redisson的分布式鎖來(lái)解決這個(gè)問(wèn)題。

先看一段lua腳本：

if (redis.call("exists", KEYS[1]) == 0) then " +   "redis.call("hincrby", KEYS[1], ARGV[2], 1); " +   "redis.call("pexpire", KEYS[1], ARGV[1]); " +   "return nil; " +   "end; " +

和大家解釋一下這一段lua腳本的意思：

exsist 先判斷有沒(méi)有這個(gè)key，來(lái)看鎖是否存在。

存在的話用hincrby設(shè)置一個(gè)hsah結(jié)構(gòu)，然后再pexpire設(shè)置過(guò)期時(shí)間

我們?cè)倏匆幌聄edisson的一個(gè)加鎖解鎖流程圖：

我們可以看到redisson使用了 watchdog來(lái)做鎖延遲操作。

在我們r(jià)edisson.trylock的時(shí)候有一個(gè)參數(shù)是releasedTime，這個(gè)參數(shù)的含義就是釋放鎖的時(shí)間。我們這個(gè)參數(shù)如果傳了，那么看門(mén)狗就會(huì)不生效，沒(méi)傳的話看門(mén)狗生效，這一點(diǎn)很重要。

redisson 看門(mén)狗會(huì)默認(rèn)10s執(zhí)行一次，如果沒(méi)有鎖釋放，那么自動(dòng)鎖延續(xù)。

大家看這張圖可以看到，redisson還采用了redis的消息訂閱與發(fā)布，如果一個(gè)線程設(shè)置了waitTime，他就會(huì)去在這個(gè)時(shí)間里去等待，訂閱了一個(gè)channel，當(dāng)占鎖線程一旦釋放了鎖，占鎖線程就回去發(fā)布一條消息，等待的線程訂閱到了 就可以去重試再占鎖。[外鏈圖片轉(zhuǎn)存失敗,源站可能有防盜鏈機(jī)制,建議將圖片保存下來(lái)直接上傳(img-C0EVK9Y0-1678841063259)(redis分布式鎖流程.png)]

流程分析：

1.客戶端1嘗試獲取鎖，返回null則加鎖成功，如果有設(shè)置釋放時(shí)間則直接通過(guò)lua腳本去操作redis，如果沒(méi)有設(shè)置則開(kāi)啟看門(mén)狗機(jī)制。當(dāng)沒(méi)有設(shè)置釋放時(shí)間，默認(rèn)釋放時(shí)間為30s，看門(mén)狗機(jī)制會(huì)10s進(jìn)行一次所延續(xù)。

2.當(dāng)客戶端2獲取鎖失敗，則通過(guò)redis的channel訂閱鎖釋放的時(shí)間。當(dāng)超過(guò)最大等待時(shí)間，則鎖失效。如果等待到了鎖釋放時(shí)間的通知，則開(kāi)始重新進(jìn)入循環(huán)開(kāi)始重試加鎖。

3.循環(huán)中每次都先試著獲取鎖，并得到已存在鎖的剩余時(shí)間。如果拿到了鎖，直接返回。如果鎖還存在，那么等待釋放鎖的消息，這里采用了信號(hào)量來(lái)阻塞線程，當(dāng)鎖釋放并發(fā)布釋放鎖的消息后，信號(hào)量的release方法被調(diào)用，此時(shí)被信號(hào)量阻塞的隊(duì)列中的第一個(gè)線程就可以繼續(xù)嘗試獲取鎖了。

我們?cè)倏匆幌箩尫沛i的代碼

// 判斷鎖 key 是否存在            "if (redis.call("hexists", KEYS[1], ARGV[3]) == 0) then " +                "return nil;" +            "end; " +            // 將該客戶端對(duì)應(yīng)的鎖的 hash 結(jié)構(gòu)的 value 值遞減為 0 后再進(jìn)行刪除            // 然后再向通道名為 redisson_lock__channel publish 一條 UNLOCK_MESSAGE 信息            "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.