---

theme: nico

“持續創作，加速成長！這是我參與「掘金日新計劃 · 6 月更文挑戰」的第6天，點選檢視活動詳情”

前言

本文主要對 redis 的分散式鎖的原理及實現進行深入講解。

以後，再針對 redis 分散式鎖相關的問題都有據可查。

一、背景

說說我們為什麼需要分散式鎖？

當多個執行緒同時操作同個資源的時候，我們通常會使用例如 synchronized 來保證同一時刻只能有一個執行緒獲取到物件鎖進而處理資源。而在分散式條件下，各個服務獨立部署，此時鎖服務的物件就變為當前應用服務，即其他服務仍然可以執行這塊程式碼，導致服務出現問題，那麼如果我們想讓多個服務獨立部署時，也能控制資源的單獨執行，那麼就需要引入分散式鎖來解決這種場景。

什麼是分散式鎖？

分散式鎖。就是控制分散式系統中不同程序共同訪問同一共享資源的一種鎖的實現。

分散式鎖，可以理解為“總部“ 的概念。

總部來控制鎖的佔有和通知其他服務等待。各個獨立的部署都從總部這裡獲取鎖的訊息，從而避免了各地為政的可能。分散式鎖就是這種思想。

我們可以使用一個第三方元件（例如redis、zookeeper、資料庫）進行全域性鎖的監控，控制鎖的持有和釋放。

二、分散式鎖原理及使用

setnx 是[set if not exists] 的簡寫。

將 key 的值設為 value，當且僅當 key 不存在，若給定的 key 已經存在，則 setnx 不做任何動作。

接下來，我們使用這個命令來看下分散式鎖使用的簡單案例，瞭解其內在原理。

2.1 分散式鎖演進：階段1

程式碼demo 如下：

Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", "111"); if(lock){ // 加鎖成功 Map<String, List<Catelog2VO>> stringListMap = getStringListMap(); redisTemplate.delete("lock");// 刪除鎖 return stringListMap; }else{ // 加鎖失敗: 重試 (自旋鎖) return getCatalogJsonFromRedis(); }

問題：

這是最原始的鎖使用問題，但這裡明顯有一個問題，如果鎖佔用後在執行業務的過程中，程式宕機，此時這個鎖就會永遠在redis中存在。

解決辦法：

• 設定鎖的過期時間，即使沒有刪除，會自動刪除。

2.2 分散式鎖演進：階段2

程式碼 demo 如下：

Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", "111"); if(lock){ redisTemplate.expire("lock",30,TimeUnit.SECONDS); // 加鎖成功 Map<String, List<Catelog2VO>> stringListMap = getStringListMap(); redisTemplate.delete("lock");// 刪除鎖 return stringListMap; }else{ // 加鎖失敗: 重試 (自旋鎖) return getCatalogJsonFromRedis(); }

問題：

• setnx 設定好，又去設定過期時間，中間出現宕機，也會出現死鎖。

解決：

• 設定過期時間和佔位必須是原子性的。redis 支援使用 setnx ex 命令。

2.3 分散式鎖演進：階段3

Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", "111",30,TimeUnit.SECONDS); if(lock){ // 加鎖成功 Map<String, List<Catelog2VO>> stringListMap = getStringListMap(); redisTemplate.delete("lock");// 刪除鎖 return stringListMap; }else{ // 加鎖失敗: 重試 (自旋鎖) return getCatalogJsonFromRedis(); }

可以使用一條命令來完成這個上鎖操作。

問題：

• 鎖是直接刪除的嗎？（如果業務很忙，鎖自己過期了，我們直接刪除，可能會把別人正在持有的鎖刪除）

解決：

• 上鎖的時候，值指定為 uuid，每個人匹配是自己的鎖才刪除。

---

2.4 分散式鎖演進：階段4

String uuid = UUID.randomUUID().toString(); Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", uuid,30,TimeUnit.SECONDS); if(lock){ // 加鎖成功 Map<String, List<Catelog2VO>> stringListMap = getStringListMap(); String lockValue = redisTemplate.opsForValue().get("lock"); if(uuid.equals(lockValue)){ redisTemplate.delete("lock");// 刪除鎖 } return stringListMap; }else{ // 加鎖失敗: 重試 (自旋鎖) return getCatalogJsonFromRedis(); }

這個還是存在點問題。還是有可能把其他執行緒的值給刪除了。

為什麼呢？

假如我們從redis 中獲取到 鎖的值，並且通過了 equal校驗，進入刪除鎖的邏輯，但是此時，鎖到期了，自動刪除了，且另外的執行緒進入佔用了鎖，那麼就存在問題，此時刪除的鎖就是別人的鎖。原因也跟之前一樣：在鎖對比和值刪除的時候，不是一個原子操作。

解決方法就是使用 lua 指令碼進行刪除。

2.5 分散式鎖演進：階段5

使用lua指令碼刪除鎖。

String uuid = UUID.randomUUID().toString(); Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", uuid,30,TimeUnit.SECONDS); if(lock){ Map<String, List<Catelog2VO>> stringListMap; try{ // 加鎖成功 stringListMap = getStringListMap(); }finally { // 定義lua 指令碼 String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end"; // 原子刪除 Integer lock1 = redisTemplate.execute(new DefaultRedisScript<Integer>(script, Integer.class), Arrays.asList("lock", uuid)); } return stringListMap; }else{ // 加鎖失敗: 重試 (自旋鎖) return getCatalogJsonFromRedis(); }

三、分散式鎖Redission

3.1 Redission 簡介

Redisson是一個在Redis的基礎上實現的Java駐記憶體資料網格（In-Memory Data Grid）。它不僅提供了一系列的分散式的Java常用物件，還提供了許多分散式服務。其中包括(BitSet, Set, Multimap, SortedSet, Map, List, Queue, BlockingQueue, Deque, BlockingDeque, Semaphore, Lock, AtomicLong, CountDownLatch, Publish / Subscribe, Bloom filter, Remote service, Spring cache, Executor service, Live Object service, Scheduler service) Redisson提供了使用Redis的最簡單和最便捷的方法。Redisson的宗旨是促進使用者對Redis的關注分離（Separation of Concern），從而讓使用者能夠將精力更集中地放在處理業務邏輯上。

redission 是redis 官方推薦的客戶端，提供了RedLock 的鎖，RedLock 繼承自 juc 的Lock 介面，提供了中斷、超時、嘗試獲取鎖等操作，支援可重入，互斥等特性。

3.2 使用

匯入依賴

<!-- 整合 redission作為分散式鎖等功能框架 --> <dependency> <groupId>org.redisson</groupId> <artifactId>redisson</artifactId> <version>3.15.2</version> </dependency>

配置：

以下為官網提供的參考配置：

``` // 預設連線地址 127.0.0.1:6379 RedissonClient redisson = Redisson.create();

// 配置 Config config = new Config(); config.useSingleServer().setAddress("redis://127.0.0.1:6379"); RedissonClient redisson = Redisson.create(config); ```

``` Config config = new Config(); config.useClusterServers() .setScanInterval(2000) // 叢集狀態掃描間隔時間，單位是毫秒 //可以用"rediss://"來啟用SSL連線 .addNodeAddress("redis://127.0.0.1:7000", "redis://127.0.0.1:7001") .addNodeAddress("redis://127.0.0.1:7002");

RedissonClient redisson = Redisson.create(config); ```

3.3 測試分散式鎖

``` @Controller public class TestRedissonClient { @Autowired RedissonClient redisson;

@ResponseBody
@GetMapping("/hello")
public String hello(){
    // 1、獲取一把鎖，只要鎖的名字一樣，既是同一把鎖
    RLock lock = redisson.getLock ("my-lock");

    // 2、加鎖
    lock.lock ();// 阻塞式等待

    try {
        System.out.println ("加鎖成功，執行業務..."+Thread.currentThread ().getId () );
        // 模擬超長等待
        Thread.sleep (20000);
    } catch (Exception e) {
        e.printStackTrace ( );
    }finally {
        // 3、解鎖
        System.out.println ("釋放鎖..."+Thread.currentThread ().getId () );
        lock.unlock ();
    }
    return "hello";
}

} ```

redission 解決了兩個問題：

• 1、鎖的自動續期，如果業務超長，執行期間自動給鎖上新的 30s，不用擔心鎖的時間過長。鎖自動過期被刪掉。
• 2、加鎖的業務，只要執行完成，就不會給當前鎖續期，即使不手動解鎖，鎖預設在30s以後刪除（當前執行緒銷燬前會呼叫 lock 方法）

如果給 lock 傳了超時時間，就傳送給 redis 執行指令碼，進行佔鎖，預設超時時間就是我們設定的時間

如果未指定超時時間，就是用預設的開門狗時間。（30*1000）

只要佔鎖成功，就會啟動一個定時任務【重新給鎖設定過期時間，新的過期時間就是看門狗的預設時間】，每隔10秒續簽，續簽滿時間。

internaLockLeaseTime【看門狗時間】/3 10s 。

最佳實戰，【推薦寫法】

加個時間，省掉續簽的時間。

// 10s自動解鎖，指定時間一定要大於業務時間（不然會報錯，沒把握就不要用） lock.lock (10, TimeUnit.SECONDS);

3.4 讀寫鎖

一次只有一個執行緒可以佔有寫模式的讀寫鎖，但是可以有多個執行緒同時佔有讀模式的讀寫鎖。

讀寫鎖適合於對資料結構的讀次數比寫次數多的情況，因為，讀模式鎖定時可以共享，以寫模式鎖住意味著獨佔，所以讀寫鎖又叫共享-獨佔鎖。

保證一定能讀到最新資料，修改期間，寫鎖是一個排他鎖。讀鎖是一個共享鎖，

寫鎖沒釋放讀就必須等待

寫 + 讀： 等待寫鎖

寫+讀： 等待寫鎖釋放

寫+寫：阻塞方式

讀+寫 ：有讀鎖。寫也需要等待

只要有寫的存在，都必須等待

3.5 快取一致性問題

快取裡的資料如何與資料庫保持一致。

3.5.1 雙寫模式

資料庫改完後，再改快取。

問題：會有髒資料\ 方案一：加鎖\ 方案二：若是允許延遲（今天更新的資料明天展示，或者幾分鐘幾小時延遲），設定過期時間 （建議）

3.5.2 失效模式

資料庫改完，再將快取刪掉，等待下次主動查詢，再進行更新。

問題：會有讀寫，髒資料\ 方案一：加鎖\ 方案二：如果經常寫，少讀，不如直接資料庫操作，去掉快取層。

3.5.3 方案 （過期時間+讀寫鎖）

無論是雙寫模式還是失效模式，都會導致快取的不一致問題。即多個例項同時更新會出事。怎麼辦?\ （1）如果是使用者緯度資料（訂單資料、使用者資料)，這種併發機率非常小，不用考慮這個問題，快取資料加上過期時間，每隔一段時間觸發讀的主動更新即可\ （2）如果是選單，商品介紹等基礎資料，也可以去使用canal訂閱binlog的方式。\ （3）快取資料+過期時間也足夠解決大部分業務對於快取的要求。\ （4）通過加鎖保證併發讀+寫，寫+寫的時候按順序排好隊。讀讀無所謂。所以適合使用讀寫鎖。(業務不關心臟資料，允許臨時髒資料可忽略）

總結:\ （1）我們能放入快取的資料本就不應該是實時性、一致性要求超高的。所以快取資料的時候加上過期時間，保證每天拿到當前最新資料即可。\ （2）我們不應該過度設計，增加系統的複雜性\ （3）遇到實時性、一致性要求高的資料，就應該查資料庫，即使慢點。

針對快取資料一致性問題，我們可以使用 Cannal 來完成這一場景。一般針對大資料專案