上篇講解了如何用 Redis 實現分布式鎖的五種方案，但我們還是有更優的王者方案，就是用 Redisson。
緩存系列文章：
緩存實戰（一）：20 圖 ｜6 千字｜緩存實戰（上篇）
緩存實戰（二）：Redis 分布式鎖｜從青銅到鑽石的五種演進方案
我們先來看下 Redis 官網怎么說，
而 Java 版的 分布式鎖的框架就是 Redisson。本篇實戰內容將會基於我的開源項目 PassJava 來整合 Redisson。
我把后端、前端、小程序都上傳到同一個倉庫里面了，大家可以通過 Github 或 碼雲訪問。地址如下：
Github: https://github.com/Jackson0714/PassJava-Platform
碼雲：https://gitee.com/jayh2018/PassJava-Platform
配套教程：www.passjava.cn
在實戰之前，我們先來看下使用 Redisson 的原理。
一、Redisson 是什么？
如果你之前是在用 Redis 的話，那使用 Redisson 的話將會事半功倍，Redisson 提供了使用 Redis的最簡單和最便捷的方法。
Redisson的宗旨是促進使用者對 Redis 的關注分離（Separation of Concern），從而讓使用者能夠將精力更集中地放在處理業務邏輯上。
Redisson 是一個在 Redis 的基礎上實現的 Java 駐內存數據網格（In-Memory Data Grid）。
-
Netty 框架：Redisson采用了基於NIO的Netty框架，不僅能作為Redis底層驅動客戶端，具備提供對Redis各種組態形式的連接功能，對Redis命令能以同步發送、異步形式發送、異步流形式發送或管道形式發送的功能，LUA腳本執行處理，以及處理返回結果的功能
-
基礎數據結構：將原生的Redis Hash，List，Set，String，Geo，HyperLogLog等數據結構封裝為Java里大家最熟悉的映射（Map），列表（List），集（Set），通用對象桶（Object Bucket），地理空間對象桶（Geospatial Bucket），基數估計算法（HyperLogLog）等結構，
-
分布式數據結構：這基礎上還提供了分布式的多值映射（Multimap），本地緩存映射（LocalCachedMap），有序集（SortedSet），計分排序集（ScoredSortedSet），字典排序集（LexSortedSet），列隊（Queue），阻塞隊列（Blocking Queue），有界阻塞列隊（Bounded Blocking Queue），雙端隊列（Deque），阻塞雙端列隊（Blocking Deque），阻塞公平列隊（Blocking Fair Queue），延遲列隊（Delayed Queue），布隆過濾器（Bloom Filter），原子整長形（AtomicLong），原子雙精度浮點數（AtomicDouble），BitSet等Redis原本沒有的分布式數據結構。
-
分布式鎖：Redisson還實現了Redis文檔中提到像分布式鎖Lock這樣的更高階應用場景。事實上Redisson並沒有不止步於此，在分布式鎖的基礎上還提供了聯鎖（MultiLock），讀寫鎖（ReadWriteLock），公平鎖（Fair Lock），紅鎖（RedLock），信號量（Semaphore），可過期性信號量（PermitExpirableSemaphore）和閉鎖（CountDownLatch）這些實際當中對多線程高並發應用至關重要的基本部件。正是通過實現基於Redis的高階應用方案，使Redisson成為構建分布式系統的重要工具。
二、整合 Redisson
Spring Boot 整合 Redisson 有兩種方案：
- 程序化配置。
- 文件方式配置。
本篇介紹如何用程序化的方式整合 Redisson。
2.1 引入 Maven 依賴
在 passjava-question 微服務的 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 自定義配置類
下面的代碼是單節點 Redis 的配置。
@Configuration
public class MyRedissonConfig {
/**
* 對 Redisson 的使用都是通過 RedissonClient 對象
* @return
* @throws IOException
*/
@Bean(destroyMethod="shutdown") // 服務停止后調用 shutdown 方法。
public RedissonClient redisson() throws IOException {
// 1.創建配置
Config config = new Config();
// 集群模式
// config.useClusterServers().addNodeAddress("127.0.0.1:7004", "127.0.0.1:7001");
// 2.根據 Config 創建出 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 對象實現了java.util.concurrent.locks.Lock接口。同時還提供了異步（Async）、反射式（Reactive）和RxJava2標准的接口。
RLock lock = redisson.getLock("anyLock");
// 最常見的使用方法
lock.lock();
我們用 passjava 這個開源項目測試下可重入鎖的兩個點：
- （1）多個線程搶占鎖，后面鎖需要等待嗎？
- （2）如果搶占到鎖的線程所在的服務停了，鎖會不會被釋放？
3.1.1 驗證一：可重入鎖是阻塞的嗎？
為了驗證以上兩點，我寫了個 demo 程序：代碼的流程就是設置WuKong-lock鎖，然后加鎖，打印線程 ID，等待 10 秒后釋放鎖，最后返回響應：“test lock ok”。
@ResponseBody
@GetMapping("test-lock")
public String TestLock() {
// 1.獲取鎖，只要鎖的名字一樣，獲取到的鎖就是同一把鎖。
RLock lock = redisson.getLock("WuKong-lock");
// 2.加鎖
lock.lock();
try {
System.out.println("加鎖成功，執行后續代碼。線程 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
第一個線程對應的線程 ID 為 86，10秒后，釋放鎖。在這期間，第二個線程需要等待鎖釋放。
第一個線程釋放鎖之后，第二個線程獲取到了鎖，10 秒后，釋放鎖。
畫了一個流程圖，幫助大家理解。如下圖所示：
- 第一步：線程 A 在 0 秒時，搶占到鎖，0.1 秒后，開始執行等待 10 s。
- 第二步：線程 B 在 0.1 秒嘗試搶占鎖，未能搶到鎖（被 A 搶占了）。
- 第三步：線程 A 在 10.1 秒后，釋放鎖。
- 第四步：線程 B 在 10.1 秒后搶占到鎖，然后等待 10 秒后釋放鎖。
由此可以得出結論，Redisson 的可重入鎖（lock）是阻塞其他線程的，需要等待其他線程釋放的。
3.1.2 驗證二：服務停了，鎖會釋放嗎？
如果線程 A 在等待的過程中，服務突然停了，那么鎖會釋放嗎？如果不釋放的話，就會成為死鎖，阻塞了其他線程獲取鎖。
我們先來看下線程 A 的獲取鎖后的，Redis 客戶端查詢到的結果，如下圖所示：
WuKong-lock 有值，而且大家可以看到 TTL 在不斷變小，說明 WuKong-lock 是自帶過期時間的。
通過觀察，經過 30 秒后，WuKong-lock 過期消失了。說明 Redisson 在停機后，占用的鎖會自動釋放。
那這又是什么原理呢？這里就要提一個概念了，看門狗。
3.2 看門狗原理
如果負責儲存這個分布式鎖的 Redisson 節點宕機以后，而且這個鎖正好處於鎖住的狀態時，這個鎖會出現鎖死的狀態。為了避免這種情況的發生，Redisson內部提供了一個監控鎖的看門狗，它的作用是在Redisson實例被關閉前，不斷的延長鎖的有效期。
默認情況下，看門狗的檢查鎖的超時時間是30秒鍾，也可以通過修改Config.lockWatchdogTimeout來另行指定。
如果我們未制定 lock 的超時時間，就使用 30 秒作為看門狗的默認時間。只要占鎖成功，就會啟動一個定時任務：每隔 10 秒重新給鎖設置過期的時間，過期時間為 30 秒。
如下圖所示：
當服務器宕機后，因為鎖的有效期是 30 秒，所以會在 30 秒內自動解鎖。（30秒等於宕機之前的鎖占用時間+后續鎖占用的時間）。
如下圖所示：
3.3 設置鎖過期時間
我們也可以通過給鎖設置過期時間，讓其自動解鎖。
如下所示，設置鎖 8 秒后自動過期。
lock.lock(8, TimeUnit.SECONDS);
如果業務執行時間超過 8 秒，手動釋放鎖將會報錯，如下圖所示：
所以我們如果設置了鎖的自動過期時間，則執行業務的時間一定要小於鎖的自動過期時間，否則就會報錯。
四、王者方案
上一篇我講解了分布式鎖的五種方案：《從青銅到鑽石的演進方案》，這一篇主要是講解如何用 Redisson 在 Spring Boot 項目中實現分布式鎖的方案。
因為 Redisson 非常強大，實現分布式鎖的方案非常簡潔，所以稱作王者方案。
原理圖如下：
代碼如下所示：
// 1.設置分布式鎖
RLock lock = redisson.getLock("lock");
// 2.占用鎖
lock.lock();
// 3.執行業務
...
// 4.釋放鎖
lock.unlock();
和之前 Redis 的方案相比，簡潔很多。
五、分布式讀寫鎖
基於 Redis 的 Redisson 分布式可重入讀寫鎖RReadWriteLock Java對象實現了java.util.concurrent.locks.ReadWriteLock接口。其中讀鎖和寫鎖都繼承了 RLock接口。
寫鎖是一個拍他鎖（互斥鎖），讀鎖是一個共享鎖。
- 讀鎖 + 讀鎖：相當於沒加鎖，可以並發讀。
- 讀鎖 + 寫鎖：寫鎖需要等待讀鎖釋放鎖。
- 寫鎖 + 寫鎖：互斥，需要等待對方的鎖釋放。
- 寫鎖 + 讀鎖：讀鎖需要等待寫鎖釋放。
示例代碼如下：
RReadWriteLock rwlock = redisson.getReadWriteLock("anyRWLock");
// 最常見的使用方法
rwlock.readLock().lock();
// 或
rwlock.writeLock().lock();
另外Redisson還通過加鎖的方法提供了leaseTime的參數來指定加鎖的時間。超過這個時間后鎖便自動解開了。
// 10秒鍾以后自動解鎖
// 無需調用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標准的接口。
關於信號量的使用大家可以想象一下這個場景，有三個停車位，當三個停車位滿了后，其他車就不停了。可以把車位比作信號，現在有三個信號，停一次車，用掉一個信號，車離開就是釋放一個信號。
我們用 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 發送 park 請求占用一個停車位。
然后在 redis 客戶端查看 park 的值，發現已經改為 2 了。繼續調用兩次，發現 park 的等於 0，當調用第四次的時候，會發現請求一直處於等待中，說明車位不夠了。如果想要不阻塞，可以用 tryAcquire 或 tryAcquireAsync。
我們再調用離開車位的方法，park 的值變為了 1，代表車位剩余 1 個。
注意：多次執行釋放信號量操作，剩余信號量會一直增加，而不是到 3 后就封頂了。
其他分布式鎖：
-
公平鎖（Fair Lock）
-
聯鎖（MultiLock）
-
紅鎖（RedLock）
-
讀寫鎖（ReadWriteLock）
-
可過期性信號量（PermitExpirableSemaphore）
-
閉鎖（CountDownLatch）
還有其他分布式鎖就不在本篇展開了，感興趣的同學可以查看官方文檔。
參考資料：
https://github.com/redisson/redisson
歡迎關注我的公眾號：悟空聊架構