文章目錄

• Pre
• 引
• 分布式鎖演進 V1
• 分布式鎖演進 V2
• 分布式鎖演進 V3
• 分布式鎖演進 V4
• 分布式鎖演進 V5
• 終極版-分布式鎖演進（Redisson ） V6
• • Code
  • Redisson分布式鎖實現原理
  • 源碼分析

---

Pre

Redis Version : 5.0.3

Redis進階-核心數據結構進階實戰 中我們講 strings 數據結構的時候，舉了一個例子

事實上，要實現一把相對完善的分布式鎖，需要注意的細節還是蠻多的，這里我們好好的梳理一把。

---

引

我們先來看段代碼

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

redis中提前存儲了一個key stock , value為 100

上述代碼有問題嗎？

是不是我們熟悉的超賣問題？

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

那怎么辦呢？ 沒有分布式經驗的童鞋，可能會說 加把鎖啊 云云

加鎖后 變成了啥呢？

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

那 這樣的代碼還有問題嗎？

性能問題

更為重要的是，如果你的應用是集群模式，好比 你有N個tomcat， 用戶通過NG地址訪問，你想想你的這個JVM級別的鎖 ，還有啥用，一樣會超賣…

這個時候你需要一把分布式鎖，這里我們討論的是如何使用redis實現分布式鎖

---

分布式鎖演進 V1

來， 上代碼

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("扣減成功，剩余庫存:" + realStock + "");}stringRedisTemplate.delete(key);return "扣減成功";

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

---

分布式鎖演進 V2

那這個代碼嚴謹嗎？ ---------> 有的同學說，你這個中間要是出異常了，沒有執行 stringRedisTemplate.delete(key);，那豈不是這把鎖釋放不了了，死鎖了呀？ 要不try catch finally ?

那代碼變成如下

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

---

分布式鎖演進 V3

那假設在運行的過程中，還沒有執行到finally ， 這個時候tomcat掛了，但是鎖已經set到redis里了 咋辦？ --------》 有的同學說， 簡單啊 加個超時時間唄。

那還有問題嗎？ ----------》如果宕機時間發生在

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

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

繼續聊

---

分布式鎖演進 V4

本質上： 要把set key和 設置過期時間 搞成一個原子命令 .

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

最關鍵的一行代碼

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

代碼就變成了

對于一般的應用，并發不是很高，這個也足夠用了，因為簡單啊

但是如果在高并發下，那還有問題嗎？ 這樣就滿足所有場景了嗎 ？

我們在設置key的時候，給key設置的過期時間是 10秒 ，也就說 10秒后，這個key會被redis給刪除掉， 假設你的這個業務執行了15秒才執行完。當前業務還未執行結束，第二個線程的請求已經過來了，它也能加鎖成功。 第二個線程繼續執行，執行了5秒，你的第一個線程也執行完了，最后一步 刪除key , 那第一個線程就把第二個線程加的鎖給刪掉了啊。。。。。

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

---

分布式鎖演進 V5

加鎖和解鎖必須是同一個線程 . 實現的話也簡單，value 不寫死，寫成一個線程ID或者隨機數等等 都行，刪除key的時候，比較下，相等的話才刪除

根據V4存在的問題，我們來看下代碼

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

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

這里我們不展開了。

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

關于到底設置多長的過期時間合適， 這個不好講了， 1秒中是長是短 ，1分鐘呢？ 要權衡一下。 那有沒有更好的辦法呢？

---

終極版-分布式鎖演進（Redisson ） V6

針對v5中存在的問題， 雖然解決了 加鎖和解鎖都是同一個線程， 但是還是有點小bug , 比如 你給key設置了過期時間為10秒， 但你的方法執行了15秒，方法還沒執行完，鎖已經被redis干掉了。。。另外一個線程就可以拿到鎖，繼續干活了。 多個線程同時執行，還是有潛在的bug出現。

超時的問題，你設置多長時間都不合適…

真的要徹底解決，咋弄呢？ -------》 可不可以給鎖續命？ 沒執行完就給鎖延期唄。 說起來簡單，實現起來有點復雜了。。。

簡單來說，后臺弄個定時任務，檢測這個鎖是否存在，存在的話延長時間，不存在的話就是被刪掉了，不考慮即可。

好在Redisson提供了這個牛逼的功能。

Code

@Beanpublic Redisson redisson() {// 此為單機模式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 {// 加鎖，實現鎖續命功能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("扣減成功，剩余庫存:" + realStock + "");}}finally {// 釋放鎖redissonLock.unlock();}return "扣減成功";}

總結一下 三部曲

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

第二步： 加鎖，實現鎖續命功能 redissonLock.lock();

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

---

Redisson分布式鎖實現原理

---

源碼分析

Redis進階- Redisson分布式鎖實現原理及源碼解析

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。