我們先來看下 Redis 官網(wǎng)對分布式鎖的說法：

而 Java 版的 分布式鎖的框架就是 Redisson。

本篇實戰(zhàn)內(nèi)容將會基于我的開源項目 PassJava 來整合 Redisson。

我把后端、前端、小程序都上傳到同一個倉庫里面了，大家可以通過?Github?或?碼云訪問。地址如下：

Github: https://github.com/Jackson0714/PassJava-Platform

碼云：https://gitee.com/jayh2018/PassJava-Platform

配套教程：www.passjava.cn

在實戰(zhàn)之前，我們先來看下使用 Redisson 的原理。

一、Redisson 是什么？

如果你之前是在用 Redis 的話，那使用 Redisson 的話將會事半功倍，Redisson 提供了使用 Redis的最簡單和最便捷的方法。

Redisson的宗旨是促進(jìn)使用者對 Redis 的關(guān)注分離（Separation of Concern），從而讓使用者能夠?qū)⒕Ω械胤旁谔幚順I(yè)務(wù)邏輯上。

Redisson 是一個在 Redis 的基礎(chǔ)上實現(xiàn)的 Java 駐內(nèi)存數(shù)據(jù)網(wǎng)格（In-Memory Data Grid）。

• Netty 框架：Redisson采用了基于NIO的Netty框架，不僅能作為Redis底層驅(qū)動客戶端，具備提供對Redis各種組態(tài)形式的連接功能，對Redis命令能以同步發(fā)送、異步形式發(fā)送、異步流形式發(fā)送或管道形式發(fā)送的功能，LUA腳本執(zhí)行處理，以及處理返回結(jié)果的功能

• 基礎(chǔ)數(shù)據(jù)結(jié)構(gòu)：將原生的Redis?Hash，List，Set，String，Geo，HyperLogLog等數(shù)據(jù)結(jié)構(gòu)封裝為Java里大家最熟悉的映射（Map），列表（List），集（Set），通用對象桶（Object Bucket），地理空間對象桶（Geospatial Bucket），基數(shù)估計算法（HyperLogLog）等結(jié)構(gòu)，

• 分布式數(shù)據(jù)結(jié)構(gòu)：這基礎(chǔ)上還提供了分布式的多值映射（Multimap），本地緩存映射（LocalCachedMap），有序集（SortedSet），計分排序集（ScoredSortedSet），字典排序集（LexSortedSet），列隊（Queue），阻塞隊列（Blocking Queue），有界阻塞列隊（Bounded Blocking Queue），雙端隊列（Deque），阻塞雙端列隊（Blocking Deque），阻塞公平列隊（Blocking Fair Queue），延遲列隊（Delayed Queue），布隆過濾器（Bloom Filter），原子整長形（AtomicLong），原子雙精度浮點數(shù)（AtomicDouble），BitSet等Redis原本沒有的分布式數(shù)據(jù)結(jié)構(gòu)。

• 分布式鎖：Redisson還實現(xiàn)了Redis文檔中提到像分布式鎖Lock這樣的更高階應(yīng)用場景。事實上Redisson并沒有不止步于此，在分布式鎖的基礎(chǔ)上還提供了聯(lián)鎖（MultiLock），讀寫鎖（ReadWriteLock），公平鎖（Fair Lock），紅鎖（RedLock），信號量（Semaphore），可過期性信號量（PermitExpirableSemaphore）和閉鎖（CountDownLatch）這些實際當(dāng)中對多線程高并發(fā)應(yīng)用至關(guān)重要的基本部件。正是通過實現(xiàn)基于Redis的高階應(yīng)用方案，使Redisson成為構(gòu)建分布式系統(tǒng)的重要工具。

• 節(jié)點：Redisson作為獨立節(jié)點可以用于獨立執(zhí)行其他節(jié)點發(fā)布到分布式執(zhí)行服務(wù)和分布式調(diào)度服務(wù)里的遠(yuǎn)程任務(wù)。

二、整合 Redisson

Spring Boot 整合 Redisson 有兩種方案：

• 程序化配置。

• 文件方式配置。

本篇介紹如何用程序化的方式整合 Redisson。

2.1 引入 Maven 依賴

在 passjava-question 微服務(wù)的 pom.xml 引入 redisson的 maven 依賴。

<!--?https://mvnrepository.com/artifact/org.redisson/redisson?--> <dependency><groupId>org.redisson</groupId><artifactId>redisson</artifactId><version>3.15.5</version> </dependency>

2.2 自定義配置類

下面的代碼是單節(jié)點 Redis 的配置。

@Configuration public?class?MyRedissonConfig?{/***?對?Redisson?的使用都是通過?RedissonClient?對象*?@return*?@throws?IOException*/@Bean(destroyMethod="shutdown")?//?服務(wù)停止后調(diào)用 shutdown 方法。public?RedissonClient?redisson()?throws?IOException?{//?1.創(chuàng)建配置Config?config?=?new?Config();//?集群模式//?config.useClusterServers().addNodeAddress("127.0.0.1:7004",?"127.0.0.1:7001");// 2.根據(jù) Config 創(chuàng)建出 RedissonClient 示例。config.useSingleServer().setAddress("redis://127.0.0.1:6379");return?Redisson.create(config);} }

2.3 測試配置類

新建一個單元測試方法。

@Autowired RedissonClient?redissonClient;@Test public?void?TestRedisson()?{System.out.println(redissonClient); }

我們運行這個測試方法，打印出 redissonClient

org.redisson.Redisson@77f66138

三、分布式可重入鎖

3.1 可重入鎖測試

基于Redis的Redisson分布式可重入鎖RLockJava 對象實現(xiàn)了java.util.concurrent.locks.Lock接口。同時還提供了異步（Async）、反射式（Reactive）和RxJava2標(biāo)準(zhǔn)的接口。

RLock?lock?=?redisson.getLock("anyLock"); //?最常見的使用方法 lock.lock();

我們用 passjava 這個開源項目測試下可重入鎖的兩個點：

• （1）多個線程搶占鎖，后面鎖需要等待嗎？

• （2）如果搶占到鎖的線程所在的服務(wù)停了，鎖會不會被釋放？

3.1.1 驗證一：可重入鎖是阻塞的嗎？

為了驗證以上兩點，我寫了個 demo 程序：代碼的流程就是設(shè)置WuKong-lock鎖，然后加鎖，打印線程 ID，等待 10 秒后釋放鎖，最后返回響應(yīng)：“test lock ok”。

@ResponseBody @GetMapping("test-lock") public?String?TestLock()?{// 1.獲取鎖，只要鎖的名字一樣，獲取到的鎖就是同一把鎖。RLock?lock?=?redisson.getLock("WuKong-lock");//?2.加鎖lock.lock();try?{System.out.println("加鎖成功，執(zhí)行后續(xù)代碼。線程 ID："?+?Thread.currentThread().getId());Thread.sleep(10000);}?catch?(Exception?e)?{//TODO}?finally?{lock.unlock();//?3.解鎖System.out.println("Finally，釋放鎖成功。線程 ID："?+?Thread.currentThread().getId());}return?"test?lock?ok"; }

先驗證第一個點，用兩個 http 請求來測試搶占鎖。

請求的 URL：

http://localhost:11000/question/v1/redisson/test/test-lock

第一個線程對應(yīng)的線程 ID 為 86，10秒后，釋放鎖。在這期間，第二個線程需要等待鎖釋放。

第一個線程釋放鎖之后，第二個線程獲取到了鎖，10 秒后，釋放鎖。

畫了一個流程圖，幫助大家理解。如下圖所示：

• 第一步：線程 A 在 0 秒時，搶占到鎖，0.1 秒后，開始執(zhí)行等待 10 s。

• 第二步：線程 B 在 0.1 秒嘗試搶占鎖，未能搶到鎖（被 A 搶占了）。

• 第三步：線程 A 在 10.1 秒后，釋放鎖。

• 第四步：線程 B 在 10.1 秒后搶占到鎖，然后等待 10 秒后釋放鎖。

由此可以得出結(jié)論，Redisson 的可重入鎖（lock）是阻塞其他線程的，需要等待其他線程釋放的。

3.1.2 驗證二：服務(wù)停了，鎖會釋放嗎？

如果線程 A 在等待的過程中，服務(wù)突然停了，那么鎖會釋放嗎？如果不釋放的話，就會成為死鎖，阻塞了其他線程獲取鎖。

我們先來看下線程 A 的獲取鎖后的，Redis 客戶端查詢到的結(jié)果，如下圖所示：

WuKong-lock 有值，而且大家可以看到 TTL 在不斷變小，說明 WuKong-lock 是自帶過期時間的。

通過觀察，經(jīng)過 30 秒后，WuKong-lock 過期消失了。說明 Redisson 在停機(jī)后，占用的鎖會自動釋放。

那這又是什么原理呢？這里就要提一個概念了，看門狗。

3.2 看門狗原理

如果負(fù)責(zé)儲存這個分布式鎖的 Redisson 節(jié)點宕機(jī)以后，而且這個鎖正好處于鎖住的狀態(tài)時，這個鎖會出現(xiàn)鎖死的狀態(tài)。為了避免這種情況的發(fā)生，Redisson內(nèi)部提供了一個監(jiān)控鎖的看門狗，它的作用是在Redisson實例被關(guān)閉前，不斷的延長鎖的有效期。

默認(rèn)情況下，看門狗的檢查鎖的超時時間是30秒鐘，也可以通過修改Config.lockWatchdogTimeout來另行指定。

如果我們未制定 lock 的超時時間，就使用 30 秒作為看門狗的默認(rèn)時間。只要占鎖成功，就會啟動一個定時任務(wù)：每隔 10 秒重新給鎖設(shè)置過期的時間，過期時間為 30 秒。

如下圖所示：

看門狗原理圖-1

當(dāng)服務(wù)器宕機(jī)后，因為鎖的有效期是 30 秒，所以會在 30 秒內(nèi)自動解鎖。（30秒等于宕機(jī)之前的鎖占用時間+后續(xù)鎖占用的時間）。

如下圖所示：

看門狗原理圖-2

3.3 設(shè)置鎖過期時間

我們也可以通過給鎖設(shè)置過期時間，讓其自動解鎖。

如下所示，設(shè)置鎖 8 秒后自動過期。

lock.lock(8,?TimeUnit.SECONDS);

如果業(yè)務(wù)執(zhí)行時間超過 8 秒，手動釋放鎖將會報錯，如下圖所示：

所以我們?nèi)绻O(shè)置了鎖的自動過期時間，則執(zhí)行業(yè)務(wù)的時間一定要小于鎖的自動過期時間，否則就會報錯。

四、王者方案

上一篇我講解了分布式鎖的五種方案：Redis 分布式鎖｜從青銅到鉆石的五種演進(jìn)方案，這一篇主要是講解如何用 Redisson 在 Spring Boot 項目中實現(xiàn)分布式鎖的方案。

因為 Redisson 非常強(qiáng)大，實現(xiàn)分布式鎖的方案非常簡潔，所以稱作王者方案。

原理圖如下：

代碼如下所示：

//?1.設(shè)置分布式鎖 RLock?lock?=?redisson.getLock("lock"); //?2.占用鎖 lock.lock(); //?3.執(zhí)行業(yè)務(wù) ... //?4.釋放鎖 lock.unlock();

和之前 Redis 的方案相比，簡潔很多。

下面講解下 Redisson 的其他幾種分布式鎖，相信大家在以后的項目中也會用到。

五、分布式讀寫鎖

基于 Redis 的 Redisson 分布式可重入讀寫鎖RReadWriteLock?Java對象實現(xiàn)了java.util.concurrent.locks.ReadWriteLock接口。其中讀鎖和寫鎖都繼承了?RLock接口。

寫鎖是一個排他鎖（互斥鎖），讀鎖是一個共享鎖。

• 讀鎖 + 讀鎖：相當(dāng)于沒加鎖，可以并發(fā)讀。

• 讀鎖 + 寫鎖：寫鎖需要等待讀鎖釋放鎖。

• 寫鎖 + 寫鎖：互斥，需要等待對方的鎖釋放。

• 寫鎖 + 讀鎖：讀鎖需要等待寫鎖釋放。

示例代碼如下：

RReadWriteLock?rwlock?=?redisson.getReadWriteLock("anyRWLock"); //?最常見的使用方法 rwlock.readLock().lock(); //?或 rwlock.writeLock().lock();

另外Redisson還通過加鎖的方法提供了leaseTime的參數(shù)來指定加鎖的時間。超過這個時間后鎖便自動解開了。

//?10秒鐘以后自動解鎖 //?無需調(diào)用unlock方法手動解鎖 rwlock.readLock().lock(10,?TimeUnit.SECONDS); //?或 rwlock.writeLock().lock(10,?TimeUnit.SECONDS);//?嘗試加鎖，最多等待100秒，上鎖以后10秒自動解鎖 boolean?res?=?rwlock.readLock().tryLock(100,?10,?TimeUnit.SECONDS); //?或 boolean?res?=?rwlock.writeLock().tryLock(100,?10,?TimeUnit.SECONDS); ... lock.unlock();

六、分布式信號量

基于Redis的Redisson的分布式信號量（Semaphore）Java對象RSemaphore采用了與java.util.concurrent.Semaphore相似的接口和用法。同時還提供了異步（Async）、反射式（Reactive）和RxJava2標(biāo)準(zhǔn)的接口。

關(guān)于信號量的使用大家可以想象一下這個場景，有三個停車位，當(dāng)三個停車位滿了后，其他車就不停了。可以把車位比作信號，現(xiàn)在有三個信號，停一次車，用掉一個信號，車離開就是釋放一個信號。

我們用 Redisson 來演示上述停車位的場景。

先定義一個占用停車位的方法：

/** *?停車，占用停車位 *?總共?3?個車位 */ @ResponseBody @RequestMapping("park") public?String?park()?throws?InterruptedException?{//?獲取信號量（停車場）RSemaphore?park?=?redisson.getSemaphore("park");//?獲取一個信號（停車位）park.acquire();return?"OK"; }

再定義一個離開車位的方法：

/***?釋放車位*?總共?3?個車位*/ @ResponseBody @RequestMapping("leave") public?String?leave()?throws?InterruptedException?{//?獲取信號量（停車場）RSemaphore?park?=?redisson.getSemaphore("park");//?釋放一個信號（停車位）park.release();return?"OK"; }

為了簡便，我用 Redis 客戶端添加了一個 key：“park”，值等于 3，代表信號量為 park，總共有三個值。

然后用 postman 發(fā)送 park 請求占用一個停車位。

然后在 redis 客戶端查看 park 的值，發(fā)現(xiàn)已經(jīng)改為 2 了。繼續(xù)調(diào)用兩次，發(fā)現(xiàn) park 的等于 0，當(dāng)調(diào)用第四次的時候，會發(fā)現(xiàn)請求一直處于等待中，說明車位不夠了。如果想要不阻塞，可以用 tryAcquire 或 tryAcquireAsync。

我們再調(diào)用離開車位的方法，park 的值變?yōu)榱?1，代表車位剩余 1 個。

注意：多次執(zhí)行釋放信號量操作，剩余信號量會一直增加，而不是到 3 后就封頂了。

其他分布式鎖：

• 公平鎖（Fair Lock）

• 聯(lián)鎖（MultiLock）

• 紅鎖（RedLock）

• 讀寫鎖（ReadWriteLock）

• 可過期性信號量（PermitExpirableSemaphore）

• 閉鎖（CountDownLatch）

參考資料：

https://github.com/redisson/redisson

總結(jié)

以上是生活随笔為你收集整理的分布式锁中的王者方案：Redisson的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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