文章目錄

• Pre
• 引
• 分布式鎖演進(jìn) V1
• 分布式鎖演進(jìn) V2
• 分布式鎖演進(jìn) V3
• 分布式鎖演進(jìn) V4
• 分布式鎖演進(jìn) V5
• 終極版-分布式鎖演進(jìn)（Redisson ） V6
• • Code
  • Redisson分布式鎖實(shí)現(xiàn)原理
  • 源碼分析

---

Pre

Redis Version : 5.0.3

Redis進(jìn)階-核心數(shù)據(jù)結(jié)構(gòu)進(jìn)階實(shí)戰(zhàn) 中我們講 strings 數(shù)據(jù)結(jié)構(gòu)的時(shí)候，舉了一個(gè)例子

事實(shí)上，要實(shí)現(xiàn)一把相對(duì)完善的分布式鎖，需要注意的細(xì)節(jié)還是蠻多的，這里我們好好的梳理一把。

---

引

我們先來(lái)看段代碼

int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));if (stock > 0) {int realStock = stock - 1;stringRedisTemplate.opsForValue().set("stock", realStock + "");}

redis中提前存儲(chǔ)了一個(gè)key stock , value為 100

上述代碼有問(wèn)題嗎？

是不是我們熟悉的超賣問(wèn)題？

為啥會(huì)超賣？ 假設(shè)同時(shí)有兩個(gè)線程都執(zhí)行到了 int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); ， 比如都取到了stock為 100 ， 然后繼續(xù)執(zhí)行后面的業(yè)務(wù)邏輯，到最后將扣減后的值set到redis中，應(yīng)該剩98吧， 事實(shí)上呢？ 你庫(kù)存里的值是 99個(gè)… 賣到最后，是不是賣多了？ 。。。。

那怎么辦呢？ 沒有分布式經(jīng)驗(yàn)的童鞋，可能會(huì)說(shuō) 加把鎖啊 云云

加鎖后 變成了啥呢？

synchronized(this){int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));if (stock > 0) {int realStock = stock - 1;stringRedisTemplate.opsForValue().set("stock", realStock + "");} }

那 這樣的代碼還有問(wèn)題嗎？

性能問(wèn)題

更為重要的是，如果你的應(yīng)用是集群模式，好比 你有N個(gè)tomcat， 用戶通過(guò)NG地址訪問(wèn)，你想想你的這個(gè)JVM級(jí)別的鎖 ，還有啥用，一樣會(huì)超賣…

這個(gè)時(shí)候你需要一把分布式鎖，這里我們討論的是如何使用redis實(shí)現(xiàn)分布式鎖

---

分布式鎖演進(jìn) V1

來(lái)， 上代碼

String key = "STOCK_LOCK";Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(key,"ARTISAN_LOCK");if (!result){ // 如果未獲取到鎖，直接返回return "1001";}int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));if (stock > 0) {int realStock = stock - 1;stringRedisTemplate.opsForValue().set("stock", realStock + "");System.out.println("扣減成功，剩余庫(kù)存:" + realStock + "");}stringRedisTemplate.delete(key);return "扣減成功";

我們來(lái)分析下， stringRedisTemplate.opsForValue().setIfAbsent(key,"ARTISAN_LOCK"); 這行代碼就保證了只有一個(gè)線程能set成功 （redis 的工作線程是單線程的嘛 ）， setIfAbsent 不存在才設(shè)置，如果有一個(gè)線程設(shè)置成功了，在這個(gè)線程未釋放之前，其他線程是無(wú)法set成功的，所以其他線程返回false，直接return了。

---

分布式鎖演進(jìn) V2

那這個(gè)代碼嚴(yán)謹(jǐn)嗎？ ---------> 有的同學(xué)說(shuō)，你這個(gè)中間要是出異常了，沒有執(zhí)行 stringRedisTemplate.delete(key);，那豈不是這把鎖釋放不了了，死鎖了呀？ 要不try catch finally ?

那代碼變成如下

那，這樣就完美了嗎？ 拋出異常的場(chǎng)景我們是處理了，在finally里釋放。

---

分布式鎖演進(jìn) V3

那假設(shè)在運(yùn)行的過(guò)程中，還沒有執(zhí)行到finally ， 這個(gè)時(shí)候tomcat掛了，但是鎖已經(jīng)set到redis里了 咋辦？ --------》 有的同學(xué)說(shuō)， 簡(jiǎn)單啊 加個(gè)超時(shí)時(shí)間唄。

那還有問(wèn)題嗎？ ----------》如果宕機(jī)時(shí)間發(fā)生在

Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(key,"ARTISAN_LOCK");stringRedisTemplate.expire(key,5000,TimeUnit.SECONDS);

這兩行代碼之間，有怎么辦？ … 不會(huì)這么巧吧 …但理論上是存在的

繼續(xù)聊

---

分布式鎖演進(jìn) V4

本質(zhì)上： 要把set key和 設(shè)置過(guò)期時(shí)間 搞成一個(gè)原子命令 .

低版本的Redis，你可能需要lua腳本，但是現(xiàn)在Redis提供了setnx 命令， spring也幫我們封裝好了

最關(guān)鍵的一行代碼

Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(key,"ARTISAN_LOCK",10,TimeUnit.SECONDS);

代碼就變成了

對(duì)于一般的應(yīng)用，并發(fā)不是很高，這個(gè)也足夠用了，因?yàn)楹?jiǎn)單啊

但是如果在高并發(fā)下，那還有問(wèn)題嗎？ 這樣就滿足所有場(chǎng)景了嗎 ？

我們?cè)谠O(shè)置key的時(shí)候，給key設(shè)置的過(guò)期時(shí)間是 10秒 ，也就說(shuō) 10秒后，這個(gè)key會(huì)被redis給刪除掉， 假設(shè)你的這個(gè)業(yè)務(wù)執(zhí)行了15秒才執(zhí)行完。當(dāng)前業(yè)務(wù)還未執(zhí)行結(jié)束，第二個(gè)線程的請(qǐng)求已經(jīng)過(guò)來(lái)了，它也能加鎖成功。 第二個(gè)線程繼續(xù)執(zhí)行，執(zhí)行了5秒，你的第一個(gè)線程也執(zhí)行完了，最后一步 刪除key , 那第一個(gè)線程就把第二個(gè)線程加的鎖給刪掉了啊。。。。。

刪了別的線程加的鎖，并發(fā)一高，你這個(gè)鎖就沒啥用了哇。。。所以 還有另外一個(gè)原則： 加鎖和解鎖必須是同一個(gè)線程 .

---

分布式鎖演進(jìn) V5

加鎖和解鎖必須是同一個(gè)線程 . 實(shí)現(xiàn)的話也簡(jiǎn)單，value 不寫死，寫成一個(gè)線程ID或者隨機(jī)數(shù)等等 都行，刪除key的時(shí)候，比較下，相等的話才刪除

根據(jù)V4存在的問(wèn)題，我們來(lái)看下代碼

那有的童鞋會(huì)問(wèn)，如果 在finally 中 執(zhí)行到if 掛了。。。并沒有執(zhí)行delete咋辦？ 理論上是有可能發(fā)生的， 其實(shí)也不要緊，我們set key的時(shí)候，設(shè)置了一個(gè)超時(shí)時(shí)間， 那最多鎖10秒嘛 ，不會(huì)死鎖。 也能接受。

如果你非得要想改這個(gè)地方，把查詢和delete弄成一個(gè)原子命令，lua腳本就排上用場(chǎng)了。

這里我們不展開了。

到這里，一把相對(duì)完善的鎖，就OK了。

關(guān)于到底設(shè)置多長(zhǎng)的過(guò)期時(shí)間合適， 這個(gè)不好講了， 1秒中是長(zhǎng)是短 ，1分鐘呢？ 要權(quán)衡一下。 那有沒有更好的辦法呢？

---

終極版-分布式鎖演進(jìn)（Redisson ） V6

針對(duì)v5中存在的問(wèn)題， 雖然解決了 加鎖和解鎖都是同一個(gè)線程， 但是還是有點(diǎn)小bug , 比如 你給key設(shè)置了過(guò)期時(shí)間為10秒， 但你的方法執(zhí)行了15秒，方法還沒執(zhí)行完，鎖已經(jīng)被redis干掉了。。。另外一個(gè)線程就可以拿到鎖，繼續(xù)干活了。 多個(gè)線程同時(shí)執(zhí)行，還是有潛在的bug出現(xiàn)。

超時(shí)的問(wèn)題，你設(shè)置多長(zhǎng)時(shí)間都不合適…

真的要徹底解決，咋弄呢？ -------》 可不可以給鎖續(xù)命？ 沒執(zhí)行完就給鎖延期唄。 說(shuō)起來(lái)簡(jiǎn)單，實(shí)現(xiàn)起來(lái)有點(diǎn)復(fù)雜了。。。

簡(jiǎn)單來(lái)說(shuō)，后臺(tái)弄個(gè)定時(shí)任務(wù)，檢測(cè)這個(gè)鎖是否存在，存在的話延長(zhǎng)時(shí)間，不存在的話就是被刪掉了，不考慮即可。

好在Redisson提供了這個(gè)牛逼的功能。

Code

@Beanpublic Redisson redisson() {// 此為單機(jī)模式Config config = new Config();config.useSingleServer().setAddress("redis://192.168.18.130:6379").setConnectionMinimumIdleSize(10).setDatabase(0);/*config.useClusterServers().addNodeAddress("redis://192.168.0.61:8001").addNodeAddress("redis://192.168.0.62:8002").addNodeAddress("redis://192.168.0.63:8003").addNodeAddress("redis://192.168.0.61:8004").addNodeAddress("redis://192.168.0.62:8005").addNodeAddress("redis://192.168.0.63:8006");*/return (Redisson) Redisson.create(config);} @RequestMapping("/deduct_stock")public String deductStock() throws InterruptedException {String lockKey = "STOCK_LOCK";// 獲取鎖RLock redissonLock = redisson.getLock(lockKey);try {// 加鎖，實(shí)現(xiàn)鎖續(xù)命功能redissonLock.lock();int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")if (stock > 0) {int realStock = stock - 1;stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)System.out.println("扣減成功，剩余庫(kù)存:" + realStock + "");}}finally {// 釋放鎖redissonLock.unlock();}return "扣減成功";}

總結(jié)一下 三部曲

第一步：獲取鎖 RLock redissonLock = redisson.getLock(lockKey);

第二步： 加鎖，實(shí)現(xiàn)鎖續(xù)命功能 redissonLock.lock();

第三步：釋放鎖 redissonLock.unlock();

---

Redisson分布式鎖實(shí)現(xiàn)原理

---

源碼分析

Redis進(jìn)階- Redisson分布式鎖實(shí)現(xiàn)原理及源碼解析

總結(jié)

以上是生活随笔為你收集整理的Redis进阶-细说分布式锁的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。