原文連結：[https://www.changxuan.top/?p=1230](https://www.changxuan.top/?p=1230) 在單體架構向分散式叢集架構演進的過程中，專案中必不可少的一個功能元件就是分散式鎖。在開發團隊有技術積累的情況下，做為團隊的一個“工具人”往往有限的時間都投入到了業(C)務(U)開(R)發(D)上，並不會去深究工具類中的分散式鎖到底是如何實現的。大家只需要清楚如何使用某個同事寫好的 Redis 工具類就可以了。所以，今天就帶大家從零開始實現一個基於Redis的可以在專案中直接使用的分散式鎖。 首先，需要搞清楚一個問題，**我們為什麼需要分散式鎖或者說為什麼需要鎖**？下面我們通過一張圖來說明這個問題，  在上面這張圖中，同時有兩個執行緒 Thread A 和 Thread B 要做同一件事，你可以理解它們為要同時執行一個程式碼塊。但是，執行這個程式碼塊有個特殊的要求：不能有多個執行緒同時執行，不然系統資料就會出亂子！所有需要執行這段程式碼的執行緒，需要挨個排隊來執行。所以，此時就需要有一個所有執行緒都能訪問到的一個變數，根據這個變數的狀態來判斷此時是否有其它執行緒正在執行，來決定當前執行緒是否能夠執行，那麼這個變數就是“鎖”。假如設變數為 `static volitate int lock = 0;` ，當 `lock` 的值為 0 時，表明此時沒有執行緒在執行這段有特殊要求的程式碼，當 `lock` 的值為 1 時，表明此時有其它執行緒在執行這段有特殊要求的程式碼。當某個執行緒獲取到 `lock` 值為 0，且將 `lock` 值改為 1 的過程，稱為成功**獲取鎖**（*注：此過程需要是原子性的*）；當該執行緒執行完這段程式碼後，將 `lock` 的值改為 0 的操作稱為**釋放鎖**。 在分散式系統中，由於子系統需要支援水平擴充套件所以就不能把記憶體變數的狀態做為“一把鎖”了。不過我們可以把變數放到 Redis 中，這樣所有節點的執行緒都能夠訪問和操作了。 ### 一、 一把簡單的“鎖” “一口吃不成個大胖子”，我們先實現一個最簡單 Redis 鎖。我們通過逐漸發現問題並解決的過程，來加深理解。 根據前文所描述的鎖的基本原理，我首先寫了兩個方法，一個是獲取鎖 `boolean lock(String key)` ，一個是釋放鎖 `void unlock(String key)`。 **@RedisUtil.java** ```java /** * 加鎖 * @param key 鎖名稱 * @return lock 是否獲取鎖， true：獲取, false：未獲取 */ public boolean lock(String key) { boolean lock; try { // setIfAbsent 等價於 Redis 的 setnx 命令，具有原子性 lock = Boolean.TRUE.equals(redisTemplate.opsForValue().setIfAbsent(key, "LOCK")); } catch (Exception e) { log.error("獲取鎖：{}，出現異常{}", key, e); return false; } return lock; } ``` ```java /** * 釋放鎖 * @param key 鎖名稱 */ public void unlock(String key) { if (Boolean.TRUE.equals(redisTemplate.hasKey(key))) { redisTemplate.delete(key); } } ``` 下面我們就寫程式碼來測試一下， ``` 測試思路： 建立十個執行緒呼叫 work() 方法 加鎖：方法獲取到鎖的執行緒對變數 count 執行1000次自增操作，未獲取到鎖的執行緒則不執行自增操作。 不加鎖：由於自增操作是非原子性的，所以最終 count 的結果會小於 10000 大於 1000 。 ``` **@WorkService.java** ```java public static int count = 0; private static final int TIME = 1000; public void work() { String key = "TEST_KEY"; if (redisUtil.lock(key)) { log.info("執行緒：{},已經獲取鎖", Thread.currentThread().getName()); try { for (int i = 0; i < TIME; i++) { WorkService.count++; } }catch (Exception e) { log.error("發生錯誤",e); }finally { redisUtil.unlock(key); log.info("執行緒：{},已經釋放鎖", Thread.currentThread().getName()); } } else { log.info("執行緒：{},未獲取到鎖", Thread.currentThread().getName()); } } public void notLockWork() { for (int i = 0; i < TIME; i++) { WorkService.count++; } } ``` **@WorkServiceTest.java** ```java @Test void work() throws Exception { CountDownLatch downLatch = new CountDownLatch(10); LinkedList threads = new LinkedList<>(); for (int i = 0; i < 10; ++i) { Thread thread = new Thread(() -> { // 加鎖測試 workService.work(); // 未加鎖測試 // workService.notLockWork() downLatch.countDown(); }); threads.add(thread); } for (Thread thread : threads) { thread.start(); } downLatch.await(); System.out.println("count = " + WorkService.count); } ``` **加鎖測試結果（控制檯輸出）** ```shell 2021-01-17 16:52:35.256 INFO 8648 --- [Thread-150] com.cxcoder.services.WorkService : 執行緒：Thread-150,已經獲取鎖 2021-01-17 16:52:35.277 INFO 8648 --- [Thread-152] com.cxcoder.services.WorkService : 執行緒：Thread-152,未獲取到鎖 2021-01-17 16:52:35.277 INFO 8648 --- [Thread-149] com.cxcoder.services.WorkService : 執行緒：Thread-149,未獲取到鎖 2021-01-17 16:52:35.277 INFO 8648 --- [Thread-154] com.cxcoder.services.WorkService : 執行緒：Thread-154,未獲取到鎖 2021-01-17 16:52:35.277 INFO 8648 --- [Thread-148] com.cxcoder.services.WorkService : 執行緒：Thread-148,未獲取到鎖 2021-01-17 16:52:35.277 INFO 8648 --- [Thread-153] com.cxcoder.services.WorkService : 執行緒：Thread-153,未獲取到鎖 2021-01-17 16:52:35.277 INFO 8648 --- [Thread-151] com.cxcoder.services.WorkService : 執行緒：Thread-151,未獲取到鎖 2021-01-17 16:52:35.277 INFO 8648 --- [Thread-146] com.cxcoder.services.WorkService : 執行緒：Thread-146,未獲取到鎖 2021-01-17 16:52:35.277 INFO 8648 --- [Thread-147] com.cxcoder.services.WorkService : 執行緒：Thread-147,未獲取到鎖 2021-01-17 16:52:35.277 INFO 8648 --- [Thread-155] com.cxcoder.services.WorkService : 執行緒：Thread-155,未獲取到鎖 2021-01-17 16:52:35.320 INFO 8648 --- [Thread-150] com.cxcoder.services.WorkService : 執行緒：Thread-150,已經釋放鎖 count = 1000 ``` **未加鎖測試結果（控制檯輸出）** ```shell count = 3287 ``` 從控制檯輸出結果來看，目前的鎖已經可以用來限制某塊程式碼在某一時刻只能有一個執行緒在執行。但是也能夠發現，該鎖的實現機制還存在一些問題和不能滿足的需求。 例如，如果在專案裡中出現”惡意程式碼“或者不規範程式碼的情況下則會出現預料之外的結果。看下面的例子， **@WorkService.java** ```java public void work(){ String key = "TEST_KEY"; try { if (redisUtil.lock(key)){ log.info("執行緒：{}，獲取鎖", Thread.currentThread().getName()); ... ... //（A） } }catch (Exception e) { log.error("發生錯誤",e); }finally { redisUtil.unlock(key); log.info("執行緒：{}，釋放鎖", Thread.currentThread().getName()); } } ``` 如果 `void work()` 是上面的這種寫法，會出現什麼問題呢？當有`執行緒X`獲取到鎖後，正在執行 A 處的程式碼時。這時 `執行緒B` 來到後獲取鎖失敗，卻執行了 `finally` 裡的程式碼將鎖給釋放了。此時 `執行緒A` 還在執行的過程中，又來了 `執行緒C` 獲取到鎖後也開始執行。所以，這個 Redis 鎖的實現機制存在一個比較嚴重的問題是**某個執行緒所持有的鎖可以被其它執行緒隨意給釋放掉**。另外，從控制檯輸出的結果中可以看出在某個執行緒持有鎖的時間段內，其它執行緒是未被阻塞的。目前這個鎖應該被稱為存在問題的**基於Redis的非阻塞的分散式