一、Redis實(shí)現(xiàn)分布式鎖原理

為什么需要分布式鎖

在聊分布式鎖之前，有必要先解釋一下，為什么需要分布式鎖。

與分布式鎖相對(duì)就的是單機(jī)鎖，我們?cè)趯?xiě)多線程程序時(shí)，避免同時(shí)操作一個(gè)共享變量產(chǎn)生數(shù)據(jù)問(wèn)題，通常會(huì)使用一把鎖來(lái)互斥以保證共享變量的正確性，其使用范圍是在同一個(gè)進(jìn)程中。如果換做是多個(gè)進(jìn)程，需要同時(shí)操作一個(gè)共享資源，如何互斥呢？現(xiàn)在的業(yè)務(wù)應(yīng)用通常是微服務(wù)架構(gòu)，這也意味著一個(gè)應(yīng)用會(huì)部署多個(gè)進(jìn)程，多個(gè)進(jìn)程如果需要修改MySQL中的同一行記錄，為了避免操作亂序?qū)е屡K數(shù)據(jù)，此時(shí)就需要引入分布式鎖了。

想要實(shí)現(xiàn)分布式鎖，必須借助一個(gè)外部系統(tǒng)，所有進(jìn)程都去這個(gè)系統(tǒng)上申請(qǐng)加鎖。這個(gè)外部系統(tǒng)必須具有互斥能力，也就是說(shuō)，如果兩個(gè)請(qǐng)求同時(shí)到達(dá)，系統(tǒng)只會(huì)成功地為一個(gè)進(jìn)程加鎖，而另一個(gè)進(jìn)程會(huì)失敗。這個(gè)外部系統(tǒng)可以是數(shù)據(jù)庫(kù)，也可以是Redis或Zookeeper，但為了追求性能，我們通常會(huì)選擇使用Redis或Zookeeper來(lái)做。

Redis可以作為一個(gè)共享存儲(chǔ)系統(tǒng)，多個(gè)客戶端可以共享訪問(wèn)，因此可以被用來(lái)保存分布式鎖。而且 Redis 的讀寫(xiě)性能高，可以應(yīng)對(duì)高并發(fā)的鎖操作場(chǎng)景。這篇文章的重點(diǎn)在于介紹如何使用Redis實(shí)現(xiàn)分布式鎖，并探討在實(shí)現(xiàn)過(guò)程中可能會(huì)遇到的問(wèn)題。

分布式鎖如何實(shí)現(xiàn)

作為分布式鎖實(shí)現(xiàn)過(guò)程中的共享存儲(chǔ)系統(tǒng)，Redis可以使用鍵值對(duì)來(lái)保存鎖變量，在接收和處理不同客戶端發(fā)送的加鎖和釋放鎖的操作請(qǐng)求。那么，鍵值對(duì)的鍵和值具體是怎么定的呢？我們要賦予鎖變量一個(gè)變量名，把這個(gè)變量名作為鍵值對(duì)的鍵，而鎖變量的值，則是鍵值對(duì)的值，這樣一來(lái)，Redis就能保存鎖變量了，客戶端也就可以通過(guò)Redis的命令操作來(lái)實(shí)現(xiàn)鎖操作。

想要實(shí)現(xiàn)分布式鎖，必須要求Redis有互斥的能力?？梢允褂肧ETNX命令，其含義是SET IF NOT EXIST，即如果key不存在，才會(huì)設(shè)置它的值，否則什么也不做。實(shí)現(xiàn)一種分布式鎖的方法是，兩個(gè)客戶端進(jìn)程互斥地執(zhí)行該命令。

以下展示了Redis使用key/value對(duì)保存鎖變量，以及兩個(gè)客戶端同時(shí)請(qǐng)求加鎖的操作過(guò)程。

加鎖操作完成后，加鎖成功的客戶端，就可以去操作共享資源，例如，修改MySQL的某一行數(shù)據(jù)。操作完成后，還要及時(shí)釋放鎖，給后來(lái)者讓出操作共享資源的機(jī)會(huì)。如何釋放鎖呢？直接使用DEL命令刪除這個(gè)key即可。這個(gè)邏輯非常簡(jiǎn)單，整體的流程寫(xiě)成偽代碼就是下面這樣。

// 加鎖
SETNX lock_key 1
// 業(yè)務(wù)邏輯
DO THINGS
// 釋放鎖
DEL lock_key

但是，以上實(shí)現(xiàn)存在一個(gè)很大的問(wèn)題，當(dāng)客戶端1拿到鎖后，如果發(fā)生下面的場(chǎng)景，就會(huì)造成死鎖。

程序處理業(yè)務(wù)邏輯異常，沒(méi)及時(shí)釋放鎖進(jìn)程掛了，沒(méi)機(jī)會(huì)釋放鎖

以上情況會(huì)導(dǎo)致已經(jīng)獲得鎖的客戶端一直占用鎖，其他客戶端永遠(yuǎn)無(wú)法獲取到鎖。

如何避免死鎖

為了解決以上死鎖問(wèn)題，最容易想到的方案是在申請(qǐng)鎖時(shí)，在Redis中實(shí)現(xiàn)時(shí)，給鎖設(shè)置一個(gè)過(guò)期時(shí)間，假設(shè)操作共享資源的時(shí)間不會(huì)超過(guò)10s，那么加鎖時(shí)，給這個(gè)key設(shè)置10s過(guò)期即可。

但以上操作還是有問(wèn)題，加鎖、設(shè)置過(guò)期時(shí)間是2條命令，有可能只執(zhí)行了第一條，第二條卻執(zhí)行失敗，例如：

1.SETNX執(zhí)行成功，執(zhí)行EXPIRE時(shí)由于網(wǎng)絡(luò)問(wèn)題，執(zhí)行失敗
2.SETNX執(zhí)行成功，Redis異常宕機(jī)，EXPIRE沒(méi)有機(jī)會(huì)執(zhí)行
3.SETNX執(zhí)行成功，客戶端異常崩潰，EXPIRE沒(méi)有機(jī)會(huì)執(zhí)行

總之這兩條命令如果不能保證是原子操作，就有潛在的風(fēng)險(xiǎn)導(dǎo)致過(guò)期時(shí)間設(shè)置失敗，依舊有可能發(fā)生死鎖問(wèn)題。幸好在Redis 2.6.12之后，Redis擴(kuò)展了SET命令的參數(shù)，可以在SET的同時(shí)指定EXPIRE時(shí)間，這條操作是原子的，例如以下命令是設(shè)置鎖的過(guò)期時(shí)間為10秒。

SET lock_key 1 EX 10 NX

至此，解決了死鎖問(wèn)題，但還是有其他問(wèn)題。想像下面這個(gè)這樣一種場(chǎng)景：

1. 客戶端1加鎖成功，開(kāi)始操作共享資源

2. 客戶端1操作共享資源耗時(shí)太久，超過(guò)了鎖的過(guò)期時(shí)間，鎖失效（鎖被自動(dòng)釋放）

3. 客戶端2加鎖成功，開(kāi)始操作共享資源

4. 客戶端1操作共享資源完成，在finally塊中手動(dòng)釋放鎖，但此時(shí)它釋放的是客戶端2的鎖。

這里存在兩個(gè)嚴(yán)重的問(wèn)題：

• 鎖過(guò)期

• 釋放了別人的鎖

第1個(gè)問(wèn)題是評(píng)估操作共享資源的時(shí)間不準(zhǔn)確導(dǎo)致的，如果只是一味增大過(guò)期時(shí)間，只能緩解問(wèn)題降低出現(xiàn)問(wèn)題的概率，依舊無(wú)法徹底解決問(wèn)題。原因在于客戶端在拿到鎖之后，在操作共享資源時(shí)，遇到的場(chǎng)景是很復(fù)雜的，既然是預(yù)估的時(shí)間，也只能是大致的計(jì)算，不可能覆蓋所有導(dǎo)致耗時(shí)變長(zhǎng)的場(chǎng)景。

第二個(gè)問(wèn)題在于解鎖操作是不夠嚴(yán)謹(jǐn)?shù)?，因?yàn)樗且环N不加區(qū)分地釋放鎖的操作，沒(méi)有對(duì)鎖的所有權(quán)進(jìn)行檢查。如何解決呢？

鎖被別人給釋放了

解決辦法是，客戶端在加鎖時(shí)，設(shè)置一個(gè)只有自己知道的唯一標(biāo)識(shí)進(jìn)去，例如可以是自己的線程ID，如果是redis實(shí)現(xiàn)，就是SET key unique_value EX 10 NX。之后在釋放鎖時(shí)，要先判斷這把鎖是否歸自己持有，只有是自己的才能釋放它。

//釋放鎖 比較unique_value是否相等，避免誤釋放
if redis.get("key") == unique_value then
    return redis.del("key")

這里釋放鎖使用的是GET + DEL兩條命令，這時(shí)又會(huì)遇到原子性問(wèn)題了。

1. 客戶端1執(zhí)行GET，判斷鎖是自己的

2. 客戶端2執(zhí)行了SET命令，強(qiáng)制獲取到鎖（雖然發(fā)生概念很低，但要嚴(yán)謹(jǐn)考慮鎖的安全性）

3. 客戶端1執(zhí)行DEL，卻釋放了客戶端2的鎖

由此可見(jiàn)，以上GET + DEL兩個(gè)命令還是必須原子的執(zhí)行才行。怎樣原子執(zhí)行兩條命令呢？答案是Lua腳本，可以把以上邏輯寫(xiě)成Lua腳本，讓Redis執(zhí)行。因?yàn)镽edis處理每個(gè)請(qǐng)求是單線程執(zhí)行的，在執(zhí)行一個(gè)Lua腳本時(shí)其它請(qǐng)求必須等待，直到這個(gè)Lua腳本處理完成，這樣一來(lái)GET+DEL之間就不會(huì)有其他命令執(zhí)行了。

以下是使用Lua腳本（unlock.script）實(shí)現(xiàn)的釋放鎖操作的偽代碼，其中，KEYS[1]表示lock_key，ARGV[1]是當(dāng)前客戶端的唯一標(biāo)識(shí)，這兩個(gè)值都是我們?cè)趫?zhí)行 Lua腳本時(shí)作為參數(shù)傳入的。

//Lua腳本語(yǔ)言，釋放鎖 比較unique_value是否相等，避免誤釋放
if redis.call("get",KEYS[1]) == ARGV[1] then
    return redis.call("del",KEYS[1])
else
    return 0
end

最后我們執(zhí)行以下命令，即可

redis-cli  --eval  unlock.script lock_key , unique_value

這樣一路優(yōu)先下來(lái)，整個(gè)加鎖、解鎖流程就更嚴(yán)謹(jǐn)了，先小結(jié)一下，基于Redis實(shí)現(xiàn)的分布式鎖，一個(gè)嚴(yán)謹(jǐn)?shù)牧鞒倘缦拢?/p>

1. 加鎖時(shí)要設(shè)置過(guò)期時(shí)間SET lock_key unique_value EX expire_time NX

2. 操作共享資源

3. 釋放鎖：Lua腳本，先GET判斷鎖是否歸屬自己，再DEL釋放鎖

有了這個(gè)嚴(yán)謹(jǐn)?shù)逆i模型，我們還需要重新思考之前的那個(gè)問(wèn)題，鎖的過(guò)期時(shí)間不好評(píng)估怎么辦。

如何確定鎖的過(guò)期時(shí)間

前面提到過(guò)，過(guò)期時(shí)間如果評(píng)估得不好，這個(gè)鎖就會(huì)有提前過(guò)期的風(fēng)險(xiǎn)，一種妥協(xié)的解決方案是，盡量冗余過(guò)期時(shí)間，降低鎖提前過(guò)期的概率，但這個(gè)方案并不能完美解決問(wèn)題。是否可以設(shè)置這樣的方案，加鎖時(shí)，先設(shè)置一個(gè)預(yù)估的過(guò)期時(shí)間，然后開(kāi)啟一個(gè)守護(hù)線程，定時(shí)去檢測(cè)這個(gè)鎖的失效時(shí)間，如果鎖快要過(guò)期了，操作共享資源還未完成，那么就自動(dòng)對(duì)鎖進(jìn)行續(xù)期，重新設(shè)置過(guò)期時(shí)間。

Redisson是一個(gè)已封裝好這些工作的庫(kù)，可以說(shuō)是一種非常優(yōu)秀的解決方案。Redisson是一個(gè)Java語(yǔ)言實(shí)現(xiàn)的Redis SDK客戶端，在使用分布式鎖時(shí)，它就采用了自動(dòng)續(xù)期的方案來(lái)避免鎖過(guò)期，這個(gè)守護(hù)線程我們一般叫它看門(mén)狗線程。這個(gè)SDK提供的API非常友好，它可以像操作本地鎖一樣操作分布式鎖?？蛻舳艘坏┘渔i成功，就會(huì)啟動(dòng)一個(gè)watch dog看門(mén)狗線程，它是一個(gè)后臺(tái)線程，會(huì)每隔一段時(shí)間（這段時(shí)間的長(zhǎng)度與設(shè)置的鎖的過(guò)期時(shí)間有關(guān)）檢查一下，如果檢查時(shí)客戶端還持有鎖key（也就是說(shuō)還在操作共享資源），那么就會(huì)延長(zhǎng)鎖key的生存時(shí)間。

那如果客戶端在加鎖成功后就宕機(jī)了呢？宕機(jī)了那么看門(mén)狗任務(wù)就不存在了，也就無(wú)法為鎖續(xù)期了，鎖到期自動(dòng)失效。

Redis的部署方式對(duì)鎖的影響

上面討論的情況，都是鎖在單個(gè)Redis 實(shí)例中可能產(chǎn)生的問(wèn)題，并沒(méi)有涉及到Redis的部署架構(gòu)細(xì)節(jié)。

Redis發(fā)展到現(xiàn)在，幾種常見(jiàn)的部署架構(gòu)有：

• 單機(jī)模式；

• 主從模式；

• 哨兵（sentinel）模式；

• 集群模式；

我們使用Redis時(shí)，一般會(huì)采用主從集群+哨兵的模式部署，哨兵的作用就是監(jiān)測(cè)redis節(jié)點(diǎn)的運(yùn)行狀態(tài)。普通的主從模式，當(dāng)master崩潰時(shí)，需要手動(dòng)切換讓slave成為master，使用主從+哨兵結(jié)合的好處在于，當(dāng)master異常宕機(jī)時(shí)，哨兵可以實(shí)現(xiàn)故障自動(dòng)切換，把slave提升為新的master，繼續(xù)提供服務(wù)，以此保證可用性。那么當(dāng)主從發(fā)生切換時(shí)，分布式鎖依舊安全嗎？

想像這樣的場(chǎng)景：

1. 客戶端1在master上執(zhí)行SET命令，加鎖成功

2. 此時(shí)，master異常宕機(jī)，SET命令還未同步到slave上（主從復(fù)制是異步的）

3. 哨兵將slave提升為新的master，但這個(gè)鎖在新的master上丟失了，導(dǎo)致客戶端2來(lái)加鎖成功了，兩個(gè)客戶端共同操作共享資源

可見(jiàn)，當(dāng)引入Redis副本后，分布式鎖還是可能受到影響。即使Redis通過(guò)sentinel保證高可用，如果這個(gè)master節(jié)點(diǎn)由于某些原因發(fā)生了主從切換，那么就會(huì)出現(xiàn)鎖丟失的情況。

集群模式+Redlock實(shí)現(xiàn)高可靠的分布式鎖

為了避免Redis實(shí)例故障而導(dǎo)致的鎖無(wú)法工作的問(wèn)題，Redis的開(kāi)發(fā)者 Antirez提出了分布式鎖算法Redlock。Redlock算法的基本思路，是讓客戶端和多個(gè)獨(dú)立的Redis實(shí)例依次請(qǐng)求加鎖，如果客戶端能夠和半數(shù)以上的實(shí)例成功地完成加鎖操作，那么我們就認(rèn)為，客戶端成功地獲得分布式鎖了，否則加鎖失敗。這樣一來(lái)，即使有單個(gè)Redis實(shí)例發(fā)生故障，因?yàn)殒i變量在其它實(shí)例上也有保存，所以，客戶端仍然可以正常地進(jìn)行鎖操作，鎖變量并不會(huì)丟失。

來(lái)具體看下Redlock算法的執(zhí)行步驟。Redlock算法的實(shí)現(xiàn)要求Redis采用集群部署模式，無(wú)哨兵節(jié)點(diǎn)，需要有N個(gè)獨(dú)立的Redis實(shí)例（官方推薦至少5個(gè)實(shí)例）。接下來(lái)，我們可以分成3步來(lái)完成加鎖操作。

第一步是，客戶端獲取當(dāng)前時(shí)間。

第二步是，客戶端按順序依次向N個(gè)Redis實(shí)例執(zhí)行加鎖操作。

這里的加鎖操作和在單實(shí)例上執(zhí)行的加鎖操作一樣，使用SET命令，帶上NX、EX/PX選項(xiàng)，以及帶上客戶端的唯一標(biāo)識(shí)。當(dāng)然，如果某個(gè)Redis實(shí)例發(fā)生故障了，為了保證在這種情況下，Redlock算法能夠繼續(xù)運(yùn)行，我們需要給加鎖操作設(shè)置一個(gè)超時(shí)時(shí)間。如果客戶端在和一個(gè)Redis實(shí)例請(qǐng)求加鎖時(shí)，一直到超時(shí)都沒(méi)有成功，那么此時(shí)，客戶端會(huì)和下一個(gè)Redis實(shí)例繼續(xù)請(qǐng)求加鎖。一般需要將加鎖操作的超時(shí)時(shí)間設(shè)置為鎖的有效時(shí)間的一小部分，通常約為幾十毫秒。

第三步是，一旦客戶端完成了和所有Redis實(shí)例的加鎖操作，客戶端就要計(jì)算整個(gè)加鎖過(guò)程的總耗時(shí)。

客戶端只有在滿足兩個(gè)條件時(shí)，才能認(rèn)為是加鎖成功，條件一是客戶端從超過(guò)半數(shù)（大于等于 N/2+1）的Redis實(shí)例上成功獲取到了鎖；條件二是客戶端獲取鎖的總耗時(shí)沒(méi)有超過(guò)鎖的有效時(shí)間。

為何只有在大多數(shù)實(shí)例加鎖成功時(shí)才能算操作成功？事實(shí)上，多個(gè)Redis實(shí)例一起使用組成了一個(gè)分布式系統(tǒng)。在分布式系統(tǒng)中總會(huì)出現(xiàn)異常節(jié)點(diǎn)，所以在談?wù)摲植际较到y(tǒng)時(shí)，需要考慮異常節(jié)點(diǎn)達(dá)到多少個(gè)，也依舊不影響整個(gè)系統(tǒng)的正確運(yùn)行。這是一個(gè)分布式系統(tǒng)的容錯(cuò)問(wèn)題，這個(gè)問(wèn)題的結(jié)論是：如果只存在故障節(jié)點(diǎn)，只要大多數(shù)節(jié)點(diǎn)正常，那么整個(gè)系統(tǒng)依舊可以提供正確服務(wù)。

在滿足了這兩個(gè)條件后，我們需要重新計(jì)算這把鎖的有效時(shí)間，計(jì)算的結(jié)果是鎖的最初有效時(shí)間減去客戶端為獲取鎖的總耗時(shí)。如果鎖的有效時(shí)間已經(jīng)來(lái)不及完成共享數(shù)據(jù)的操作了，我們可以釋放鎖，以免出現(xiàn)還沒(méi)完成共享資源操作，鎖就過(guò)期了的情況。

當(dāng)然，如果客戶端在和所有實(shí)例執(zhí)行完加鎖操作后，沒(méi)能同時(shí)滿足這兩個(gè)條件，那么，客戶端就要向所有Redis節(jié)點(diǎn)發(fā)起釋放鎖的操作。為什么釋放鎖，要操作所有的節(jié)點(diǎn)呢，不能只操作那些加鎖成功的節(jié)點(diǎn)嗎？因?yàn)樵谀骋粋€(gè)Redis節(jié)點(diǎn)加鎖時(shí)，可能因?yàn)榫W(wǎng)絡(luò)原因?qū)е录渔i失敗，例如一個(gè)客戶端在一個(gè)Redis實(shí)例上加鎖成功，但在讀取響應(yīng)結(jié)果時(shí)由于網(wǎng)絡(luò)問(wèn)題導(dǎo)致讀取失敗，那這把鎖其實(shí)已經(jīng)在Redis上加鎖成功了。所以釋放鎖時(shí)，不管之前有沒(méi)有加鎖成功，需要釋放所有節(jié)點(diǎn)上的鎖以保證清理節(jié)點(diǎn)上的殘留的鎖。

在Redlock算法中，釋放鎖的操作和在單實(shí)例上釋放鎖的操作一樣，只要執(zhí)行釋放鎖的 Lua腳本就可以了。如果N個(gè)Redis實(shí)例中超過(guò)一半的實(shí)例正常工作，就能確保分布式鎖正常運(yùn)作。為了提高分布式鎖的可靠性，您可以在實(shí)際業(yè)務(wù)應(yīng)用中使用Redlock算法。

二、代碼實(shí)現(xiàn)Redis分布式鎖

1.SpringBoot整合redis用到最多的當(dāng)然屬于我們的老朋友RedisTemplate，pom依賴(lài)如下：

<!-- springboot整合redis -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

2.Redis配置類(lèi):

package com.example.redisdemo.config;

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

/**
 * @description: Redis配置類(lèi)
 * @author Keson
 * @date 21:20 2022/11/14
 * @Param
 * @return
 * @version 1.0
 */
@Configuration
public class RedisConfig {

    @Bean
    public RedisTemplate<String, Object> redisTemplate(LettuceConnectionFactory lettuceConnectionFactory) {
        // 設(shè)置序列化
        Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<Object>(Object.class);
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
        jackson2JsonRedisSerializer.setObjectMapper(om);
        // 配置redisTemplate
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<String, Object>();
        redisTemplate.setConnectionFactory(lettuceConnectionFactory);
        RedisSerializer<?> stringSerializer = new StringRedisSerializer();
        redisTemplate.setKeySerializer(stringSerializer);// key序列化
        redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);// value序列化
        redisTemplate.setHashKeySerializer(stringSerializer);// Hash key序列化
        redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);// Hash value序列化
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }
}

3.Service層面

package com.example.redisdemo.service;

import com.example.redisdemo.entity.CustomerBalance;
import java.util.concurrent.Callable;

/**
 * @author Keson
 * @version 1.0
 * @description: TODO
 * @date 2022/11/14 15:12
 */
public interface RedisService {

    <T> T callWithLock(CustomerBalance customerBalance, Callable<T> callable) throws Exception;
}

package com.example.redisdemo.service.impl;

import com.example.redisdemo.entity.CustomerBalance;
import com.example.redisdemo.service.RedisService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.types.Expiration;
import org.springframework.stereotype.Service;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

/**
 * @author Keson
 * @version 1.0
 * @description: TODO Redis實(shí)現(xiàn)分布式鎖
 * @date 2022/11/14 15:13
 */
@Service
@Slf4j
public class RedisServiceImpl implements RedisService {

    //設(shè)置默認(rèn)過(guò)期時(shí)間
    private final static int DEFAULT_LOCK_EXPIRY_TIME = 20;
    //自定義lock key前綴
    private final static String LOCK_PREFIX = "LOCK:CUSTOMER_BALANCE";

    @Autowired
    private RedisTemplate redisTemplate;

    @Override
    public <T> T callWithLock(CustomerBalance customerBalance, Callable<T> callable) throws Exception{
        //自定義lock key
        String lockKey = getLockKey(customerBalance.getCustomerNumber(), customerBalance.getSubAccountNumber(), customerBalance.getCurrencyCode());
        //將UUID當(dāng)做value，確保唯一性
        String lockReference = UUID.randomUUID().toString();

        try {
            if (!lock(lockKey, lockReference, DEFAULT_LOCK_EXPIRY_TIME, TimeUnit.SECONDS)) {
                throw new Exception("lock加鎖失敗");
            }
            return callable.call();
        } finally {
            unlock(lockKey, lockReference);
        }
    }

    //定義lock key
    String getLockKey(String customerNumber, String subAccountNumber, String currencyCode) {
        return String.format("%s:%s:%s:%s", LOCK_PREFIX, customerNumber, subAccountNumber, currencyCode);
    }

    //redis加鎖
    private boolean lock(String key, String value, long timeout, TimeUnit timeUnit) {
        Boolean locked;
        try {
            //SET_IF_ABSENT --> NX: Only set the key if it does not already exist.
            //SET_IF_PRESENT --> XX: Only set the key if it already exist.
            locked = (Boolean) redisTemplate.execute((RedisCallback<Boolean>) connection ->
                    connection.set(key.getBytes(StandardCharsets.UTF_8), value.getBytes(StandardCharsets.UTF_8),
                            Expiration.from(timeout, timeUnit), RedisStringCommands.SetOption.SET_IF_ABSENT));
        } catch (Exception e) {
            log.error("Lock failed for redis key: {}, value: {}", key, value);
            locked = false;
        }
        return locked != null && locked;
    }

    //redis解鎖
    private boolean unlock(String key, String value) {
        try {
            //使用lua腳本保證刪除的原子性，確保解鎖
            String script = "if redis.call(&#39;get&#39;, KEYS[1]) == ARGV[1] " +
                            "then return redis.call(&#39;del&#39;, KEYS[1]) " +
                            "else return 0 end";
            Boolean unlockState = (Boolean) redisTemplate.execute((RedisCallback<Boolean>) connection ->
                    connection.eval(script.getBytes(), ReturnType.BOOLEAN, 1,
                            key.getBytes(StandardCharsets.UTF_8), value.getBytes(StandardCharsets.UTF_8)));
            return unlockState == null || !unlockState;
        } catch (Exception e) {
            log.error("unLock failed for redis key: {}, value: {}", key, value);
            return false;
        }
    }
}

4.業(yè)務(wù)調(diào)用實(shí)現(xiàn)分布式鎖示例：

    @Override
    public int updateById(CustomerBalance customerBalance) throws Exception {
        return redisService.callWithLock(customerBalance, ()-> customerBalanceMapper.updateById(customerBalance));
    }

以上是怎么在SpringBoot中使用Redis實(shí)現(xiàn)分布式鎖的詳細(xì)內(nèi)容。更多信息請(qǐng)關(guān)注PHP中文網(wǎng)其他相關(guān)文章！