| **好看請贊，養成習慣** > - 你有一個思想，我有一個思想，我們交換後，一個人就有兩個思想 > > - If you can NOT explain it simply, you do NOT understand it well enough 現陸續將Demo程式碼和技術文章整理在一起 [Github實踐精選](https://github.com/FraserYu/learnings) ，方便大家閱讀檢視，本文同樣收錄在此，覺得不錯，還請Star ---  ## 寫在前面 進入原始碼階段了，寫了十幾篇的 [併發系列](https://dayarch.top/categories/Coding/Java-Concurrency/) 知識鋪墊終於要派上用場了。相信很多人已經忘了其中的一些理論知識，別擔心，我會在原始碼環節帶入相應的理論知識點幫助大家回憶，做到理論與實踐相結合，另外這是超長圖文，建議收藏，如果對你有用還請點贊讓更多人看到  ## Java SDK 為什麼要設計 Lock 曾幾何時幻想過，如果 Java 併發控制只有 synchronized 多好，只有下面三種使用方式，簡單方便 ```java public class ThreeSync { private static final Object object = new Object(); public synchronized void normalSyncMethod(){ //臨界區 } public static synchronized void staticSyncMethod(){ //臨界區 } public void syncBlockMethod(){ synchronized (object){ //臨界區 } } } ``` 如果在 Java 1.5之前，確實是這樣，自從 1.5 版本 Doug Lea 大師就重新造了一個輪子 Lock  我們常說：“避免重複造輪子”，如果有了輪子還是要堅持再造個輪子，那麼肯定傳統的輪子在某些應用場景中不能很好的解決問題  不知你是否還記得 [Coffman 總結的四個可以發生死鎖的情形](https://dayarch.top/p/java-concurrency-dead-lock.html#Coffman-條件) ，其中【不可剝奪條件】是指： > 執行緒已經獲得資源，在未使用完之前，不能被剝奪，只能在使用完時自己釋放 要想破壞這個條件，**就需要具有申請不到進一步資源就釋放已有資源的能力** 很顯然，這個能力是 synchronized 不具備的，使用 synchronized ，如果執行緒申請不到資源就會進入阻塞狀態，我們做什麼也改變不了它的狀態，這是 synchronized 輪子的致命弱點，這就強有力的給了重造輪子 Lock 的理由 ## 顯式鎖 Lock 舊輪子有弱點，新輪子就要解決這些問題，所以要具備不會阻塞的功能，下面的三個方案都是解決這個問題的好辦法（看下面表格描述你就明白三個方案的含義了） | 特性 | 描述 | API | | ---------------- | ------------------------------------------------------------ | ---------------------------- | | 能響應中斷 | 如果不能自己釋放，那可以響應中斷也是很好的。[Java多執行緒中斷機制](https://dayarch.top/p/java-concurrency-interrupt-mechnism.html) 專門描述了中斷過程，目的是通過中斷訊號來跳出某種狀態，比如阻塞 | lockInterruptbly() | | 非阻塞式的獲取鎖 | 嘗試獲取，獲取不到不會阻塞，直接返回 | tryLock() | | 支援超時 | 給定一個時間限制，如果一段時間內沒獲取到，不是進入阻塞狀態，同樣直接返回 | tryLock(long time, timeUnit) | 好的方案有了，但魚和熊掌不可兼得，Lock 多了 synchronized 不具備的特性，自然不會像 synchronized 那樣一個關鍵字三個玩法走遍全天下，在使用上也相對複雜了一丟丟  ### Lock 使用正規化 synchronized 有標準用法，這樣的優良傳統咱 Lock 也得有，相信很多人都知道使用 Lock 的一個**正規化** ```java Lock lock = new ReentrantLock(); lock.lock(); try{ ... }finally{ lock.unlock(); } ``` 既然是正規化（沒事不要挑戰更改寫法的那種），肯定有其理由，我們來看一下 #### 標準1—finally 中釋放鎖 這個大家應該都會明白，在 finally 中釋放鎖，目的是保證在獲取到鎖之後，**最終**能被釋放 #### 標準2—在 try{} 外面獲取鎖 不知道你有沒有想過，為什麼會有標準 2 的存在，我們通常是“喜歡” try 住所有內容，生怕發生異常不能捕獲的 在 `try{}` 外獲取鎖主要考慮兩個方面： 1. 如果沒有獲取到鎖就丟擲異常，最終釋放鎖肯定是有問題的，因為還未曾擁有鎖談何釋放鎖呢 2. 如果在獲取鎖時丟擲了異常，也就是當前執行緒並未獲取到鎖，但執行到 finally 程式碼時，如果恰巧別的執行緒獲取到了鎖，則會被釋放掉（無故釋放） > 不同鎖的實現方式略有不同，正規化的存在就是要避免一切問題的出現，所以大家儘量遵守正規化  ### Lock 是怎樣起到鎖的作用呢？ 如果你熟悉 synchronized，你知道程式編譯成 CPU 指令後，在臨界區會有 `moniterenter` 和 `moniterexit` 指令的出現，可以理解成進出臨界區的標識 從正規化上來看： - `lock.lock()` 獲取鎖，“等同於” synchronized 的 moniterenter指令 - `lock.unlock()` 釋放鎖，“等同於” synchronized 的 moniterexit 指令 那 Lock 是怎麼做到的呢？ 這裡先簡單說明一下，這樣一會到原始碼分析時，你可以遠觀設計輪廓，近觀實現細節，會變得越發輕鬆  > 其實很簡單，比如在 ReentrantLock 內部維護了一個 volatile 修飾的變數 state，通過 CAS 來進行讀寫（最底層還是交給硬體來保證原子性和可見性），如果CAS更改成功，即獲取到鎖，執行緒進入到 try 程式碼塊繼續執行；如果沒有更改成功，執行緒會被【掛起】，不會向下執行 但 Lock 是一個介面，裡面根本沒有 state 這個變數的存在：  它怎麼處理這個 state 呢？很顯然需要一點設計的加成了，介面定義行為，具體都是需要實現類的 > Lock 介面的實現類基本都是通過【聚合】了一個【佇列同步器】的子類完成執行緒訪問控制的 那什麼是佇列同步器呢？ （這應該是你見過的最強標題黨，聊了半個世紀才入正題，評論區留言罵我） ## 佇列同步器 AQS 佇列同步器 （AbstractQueuedSynchronizer），簡稱同步器或AQS，就是我們今天的主人公 > **問：**為什麼你分析 JUC 原始碼，要從 AQS 說起呢？ > > **答：**看下圖  相信看到這個截圖你就明白一二了，你聽過的，面試常被問起的，工作中常用的 - `ReentrantLock` - `ReentrantReadWriteLock` - `Semaphore(訊號量)` - `CountDownLatch` - `公平鎖` - `非公平鎖` - `ThreadPoolExecutor` (關於執行緒池的理解，可以檢視 [為什麼要使用執行緒池?](https://dayarch.top/p/why-we-need-to-use-threadpool.html) ) 都和 AQS 有直接關係，所以瞭解 AQS 的抽象實現，在此基礎上再稍稍檢視上述各類的實現細節，很快就可以全部搞定，不至於檢視原始碼時一頭霧水，丟失主線 上面提到，在鎖的實現類中會聚合同步器，然後利同步器實現鎖的語義，那麼問題來了： > 為什麼要用聚合模式，怎麼進一步理解鎖和同步器的關係呢？  我們絕大多數都是在使用鎖，實現鎖之後，其核心就是要使用方便  從 AQS 的類名稱和修飾上來看，這是一個抽象類，所以從設計模式的角度來看同步器一定是基於【模版模式】來設計的，使用者需要繼承同步器，實現自定義同步器，並重寫指定方法，隨後將同步器組合在自定義的同步元件中，並呼叫同步器的模版方法，而這些模版方法又回撥用使用者重寫的方法 我不想將上面的解釋說的這麼抽象，其實想理解上面這句話，我們只需要知道下面兩個問題就好了 1. 哪些是自定義同步器可重寫的方法？ 2. 哪些是抽象同步器提供的模版方法？ ### 同步器可重寫的方法 同步器提供的可重寫方法只有5個，這大大方便了鎖的使用者：  按理說，需要重寫的方法也應該有 abstract 來修飾的，為什麼這裡沒有？原因其實很簡單，上面的方法我已經用顏色區分成了兩類： - `獨佔式` - `共享式` 自定義的同步元件或者鎖不可能既是獨佔式又是共享式，為了避免強制重寫不相干方法，所以就沒有 abstract 來修飾了，但要丟擲異常告知不能直接使用該方法： ```java protected boolean tryAcquire(int arg) { throw new UnsupportedOperationException(); } ``` 暖暖的很貼心（如果你有類似的需求也可以仿照這樣的設計） 表格方法描述中所說的`同步狀態`就是上文提到的有 volatile 修飾的 state，所以我們在`重寫`上面幾個方法時，還要通過同步器提供的下面三個方法（AQS 提供的）來獲取或修改同步狀態：  而獨佔式和共享式操作 state 變數的區別也就很簡單了  所以你看到的 `ReentrantLock` `ReentrantReadWriteLock` `Semaphore(訊號量)` `CountDownLatch` 這幾個類其實僅僅是在實現以上幾個方法上略有差別，其他的實現都是通過同步器的模版方法來實現的，到這裡是不是心情放鬆了許多呢？我們來看一看模版方法： ### 同步器提供的模版方法 上面我們將同步器的實現方法分為獨佔式和共享式兩類，模版方法其實除了提供以上兩類模版方法之外，只是多了`響應中斷`和`超時限制` 的模版方法供 Lock 使用，來看一下  先不用記上述方法的功能，目前你只需要瞭解個大概功能就好。另外，相信你也注意到了： > 上面的方法都有 final 關鍵字修飾，說明子類不能重寫這個方法 看到這你也許有點亂了，我們稍微歸納一下：  程式設計師還是看程式碼心裡踏實一點，我們再來用程式碼說明一下上面的關係（注意程式碼中的註釋，以下的程式碼並不是很嚴謹，只是為了簡單說明上圖的程式碼實現）： ```java package top.dayarch.myjuc; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.AbstractQueuedSynchronizer; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; /** * 自定義互斥鎖 * * @author tanrgyb * @date 2020/5/23 9:33 PM */ public class MyMutex implements Lock { // 靜態內部類-自定義同步器 private static class MySync extends AbstractQueuedSynchronizer{ @Override protected boolean tryAcquire(int arg) { // 呼叫AQS提供的方法，通過CAS保證原子性 if (compareAndSetState(0, arg)){ // 我們實現的是互斥鎖，所以標記獲取到同步狀態（更新state成功）的執行緒， // 主要為了判斷是否可重入（一會兒會說明） setExclusiveOwnerThread(Thread.currentThread()); //獲取同步狀態成功，返回 true return true; } // 獲取同步狀態失敗，返回 false return false; } @Override protected boolean tryRelease(int arg) { // 未擁有鎖卻讓釋放，會丟擲IMSE if (getState() == 0){ throw new IllegalMonitorStateException(); } // 可以釋放，清空排它執行緒標記 setExclusiveOwnerThread(null); // 設定同步狀態為0，表示釋放鎖 setState(0); return true; } // 是否獨佔式持有 @Override protected boolean isHeldExclusively() { return getState() == 1; } // 後續會用到，主要用於等待/通知機制，每個condition都有一個與之對應的條件等待佇列，在鎖模型中說明過 Condition newCondition() { return new ConditionObject(); } } // 聚合自定義同步器 private final MySync sync = new MySync(); @Override public void lock() { // 阻塞式的獲取鎖，呼叫同步器模版方法獨佔式，獲取同步狀態 sync.acquire(1); } @Override public void lockInterruptibly() throws InterruptedException { // 呼叫同步器模版方法可中斷式獲取同步狀態 sync.acquireInterruptibly(1); } @Override public boolean tryLock() { // 呼叫自己重寫的方法，非阻塞式的獲取同步狀態 return sync.tryAcquire(1); } @Override public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { // 呼叫同步器模版方法，可響應中斷和超時時間限制 return sync.tryAcquireNanos(1, unit.toNanos(time)); } @Override public void unlock() { // 釋放鎖 sync.release(1); } @Override public Condition newCondition() { // 使用自定義的條件 return sync.newCondition(); } } ``` 如果你現在開啟 IDE， 你會發現上文提到的 `ReentrantLock` `ReentrantReadWriteLock` `Semaphore(訊號量)` `CountDownLatch` 都是按照這個結構實現，所以我們就來看一看 AQS 的模版方法到底是怎麼實現鎖 ## AQS實現分析 從上面的程式碼中，你應該理解了`lock.tryLock()` 非阻塞式獲取鎖就是呼叫自定義同步器重寫的 `tryAcquire()` 方法，通過 CAS 設定state 狀態，不管成功與否都會馬上返回；那麼 lock.lock() 這種阻塞式的鎖是如何實現的呢？ 有阻塞就需要排隊，實現排隊必然需要佇列 > CLH：Craig、Landin and Hagersten 佇列，是一個單向連結串列，AQS中的佇列是CLH變體的虛擬雙向佇列（FIFO）——概念瞭解就好，不要記 佇列中每個排隊的個體就是一個 Node，所以我們來看一下 Node 的結構 ### Node 節點 AQS 內部維護了一個同步佇列，用於管理同步狀態。 - 當執行緒獲取同步狀態失敗時，就會將當前執行緒以及等待狀態等資訊構造成一個 Node 節點，將其加入到同步佇列中尾部，阻塞該執行緒 - 當同步狀態被釋放時，會喚醒同步佇列中“首節點”的執行緒獲取同步狀態 為了將上述步驟弄清楚，我們需要來看一看 Node 結構 （如果你能開啟 IDE 一起看那是極好的）  乍一看有點雜亂，我們還是將其歸類說明一下：  上面這幾個狀態說明有個印象就好，有了Node 的結構說明鋪墊，你也就能想象同步佇列的接本結構了：  前置知識基本鋪墊完畢，我們來看一看獨佔式獲取同步狀態的整個過程 ### 獨佔式獲取同步狀態 故事要從正規化lock.lock() 開始 ```java public void lock() { // 阻塞式的獲取鎖，呼叫同步器模版方法，獲取同步狀態 sync.acquire(1); } ``` 進入AQS的模版方法 acquire() ```java public final void acquire(int arg) { // 呼叫自定義同步器重寫的 tryAcquire 方法 if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); } ``` 首先，也會嘗試非阻塞的獲取同步狀態，如果獲取失敗（tryAcquire返回false），則會呼叫 `addWaiter` 方法構造 Node 節點（Node.EXCLUSIVE 獨佔式）並安全的（CAS）加入到同步佇列【尾部】 ```java private Node addWaiter(Node mode) { // 構造Node節點，包含當前執行緒資訊以及節點模式【獨佔/共享】 Node node = new Node(Thread.currentThread(), mode); // 新建變數 pred 將指標指向tail指向的節點 Node pred = tail; // 如果尾節點不為空 if (pred != null) { // 新加入的節點前驅節點指向尾節點 node.prev = pred; // 因為如果多個執行緒同時獲取同步狀態失敗都會執行這段程式碼 // 所以，通過 CAS 方式確保安全的設定當前節點為最新的尾節點 if (compareAndSetTail(pred, node)) { // 曾經的尾節點的後繼節點指向當前節點 pred.next = node; // 返回新構建的節點 return node; } } // 尾節點為空，說明當前節點是第一個被加入到同步佇列中的節點 // 需要一個入隊操作 enq(node); return node; } private Node enq(final Node node) { // 通過“死迴圈”確保節點被正確新增，最終將其設定為尾節點之後才會返回，這裡使用 CAS 的理由和上面一樣 for (;;) { Node t = tail; // 第一次迴圈，如果尾節點為 null if (t == null) { // Must initialize // 構建一個哨兵節點，並將頭部指標指向它 if (compareAndSetHead(new Node())) // 尾部指標同樣指向哨兵節點 tail = head; } else { // 第二次迴圈，將新節點的前驅節點指向t node.prev = t; // 將新節點加入到佇列尾節點 if (compareAndSetTail(t, node)) { // 前驅節點的後繼節點指向當前新節點，完成雙向佇列 t.next = node; return t; } } } } ``` 你可能比較迷惑 enq() 的處理方式，進入該方法就是一個“死迴圈”，我們就用圖來描述它是怎樣跳出迴圈的  有些同學可能會有疑問，為什麼會有哨兵節點？ > 哨兵，顧名思義，是用來解決國家之間邊界問題的，不直接參與生產活動。同樣，電腦科學中提到的哨兵，也用來解決邊界問題，如果沒有邊界，指定環節，按照同樣演算法可能會在邊界處發生異常，比如要繼續向下分析的 `acquireQueued()` 方法 ```java final boolean acquireQueued(final Node node, int arg) { boolean failed = true; try { boolean interrupted = false; // "死迴圈"，嘗試獲取鎖，或者掛起 for (;;) { // 獲取當前節點的前驅節點 final Node p = node.predecessor(); // 只有當前節點的前驅節點是頭節點，才會嘗試獲取鎖 // 看到這你應該理解新增哨兵節點的含義了吧 if (p == head && tryAcquire(arg)) { // 獲取同步狀態成功，將自己設定為頭 setHead(node); // 將哨兵節點的後繼節點置為空，方便GC p.next = null; // help GC failed = false; // 返回中斷標識 return interrupted; } // 當前節點的前驅節點不是頭節點 //【或者】當前節點的前驅節點是頭節點但獲取同步狀態失敗 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } } ``` 獲取同步狀態成功會返回可以理解了，但是如果失敗就會一直陷入到“死迴圈”中浪費資源嗎？很顯然不是，`shouldParkAfterFailedAcquire(p, node)` 和 `parkAndCheckInterrupt()` 就會將執行緒獲取同步狀態失敗的執行緒掛起，我們繼續向下看 ```java private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { // 獲取前驅節點的狀態 int ws = pred.waitStatus; // 如果是 SIGNAL 狀態，即等待被佔用的資源釋放，直接返回 true // 準備繼續呼叫 parkAndCheckInterrupt 方法 if (ws == Node.SIGNAL) return true; // ws 大於0說明是CANCELLED狀態， if (ws > 0) { // 迴圈判斷前驅節點的前驅節點是否也為CANCELLED狀態，忽略該狀態的節點，重新連線佇列 do { node.prev = pred = pred.prev; } while (pred.waitStatus > 0); pred.next = node; } else { // 將當前節點的前驅節點設定為設定為 SIGNAL 狀態，用於後續喚醒操作 // 程式第一次執行到這返回為false，還會進行外層第二次迴圈，最終從程式碼第7行返回 compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } return false; } ``` 到這裡你也許有個問題： > 這個地方設定前驅節點為 SIGNAL 狀態到底有什麼作用？ 保留這個問題，我們陸續揭曉 如果前驅節點的 waitStatus 是 SIGNAL狀態，即 shouldParkAfterFailedAcquire 方法會返回 true ，程式會繼續向下執行 `parkAndCheckInterrupt` 方法，用於將當前執行緒掛起 ```java private final boolean parkAndCheckInterrupt() { // 執行緒掛起，程式不會繼續向下執行 LockSupport.park(this); // 根據 park 方法 API描述，程式在下述三種情況會繼續向下執行 // 1. 被 unpark // 2. 被中斷(interrupt) // 3. 其他不合邏輯的返回才會繼續向下執行 // 因上述三種情況程式執行至此，返回當前執行緒的中斷狀態，並清空中斷狀態 // 如果由於被中斷，該方法會返回 true return Thread.interrupted(); } ``` 被喚醒的程式會繼續執行 `acquireQueued` 方法裡的迴圈，如果獲取同步狀態成功，則會返回 `interrupted = true` 的結果 程式繼續向呼叫棧上層返回，最終回到 AQS 的模版方法 `acquire` ```java public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); } ``` 你也許會有疑惑: > 程式已經成功獲取到同步狀態並返回了，怎麼會有個自我中斷呢？ ```java static void selfInterrupt() { Thread.currentThread().interrupt(); } ```  如果你不能理解中斷，強烈建議你回看 [Java多執行緒中斷機制](https://dayarch.top/p/java-concurrency-interrupt-mechnism.html) 到這裡關於獲取同步狀態我們還遺漏了一條線，acquireQueued 的 finally 程式碼塊如果你仔細看你也許馬上就會有疑惑: > 到底什麼情況才會執行 if(failed) 裡面的程式碼 ？ ```java if (failed) cancelAcquire(node); ``` 這段程式碼被執行的條件是 failed 為 true，正常情況下，如果跳出迴圈，failed 的值為false，如果不能跳出迴圈貌似怎麼也不能執行到這裡，所以只有不正常的情況才會執行到這裡，也就是會發生異常，才會執行到此處 檢視 try 程式碼塊，只有兩個方法會丟擲異常： - `node.processor()` 方法 - 自己重寫的 `tryAcquire()` 方法 先看前者：  很顯然，這裡丟擲的異常不是重點，那就以 ReentrantLock 重寫的 tryAcquire() 方法為例  另外，上面分析 `shouldParkAfterFailedAcquire` 方法還對 CANCELLED 的狀態進行了判斷，那麼 > 什麼時候會生成取消狀態的節點呢？ 答案就在 `cancelAcquire` 方法中， 我們來看看 cancelAcquire到底怎麼設定/處理 CANNELLED 的 ```java private void cancelAcquire(Node node) { // 忽略無效節點 if (node == null) return; // 將關聯的執行緒資訊清空 node.thread = null; // 跳過同樣是取消狀態的前驅節點 Node pred = node.prev; while (pred.waitStatus > 0) node.prev = pred = pred.prev; // 跳出上面迴圈後找到前驅有效節點，並獲取該有效節點的後繼節點 Node predNext = pred.next; // 將當前節點的狀態置為 CANCELLED node.waitStatus = Node.CANCELLED; // 如果當前節點處在尾節點，直接從佇列中刪除自己就好 if (node == tail && compareAndSetTail(node, pred)) { compareAndSetNext(pred, predNext, null); } else { int ws; // 1. 如果當前節點的有效前驅節點不是頭節點，也就是說當前節點不是頭節點的後繼節點 if (pred != head && // 2. 判斷當前節點有效前驅節點的狀態是否為 SIGNAL ((ws = pred.waitStatus) == Node.SIGNAL || // 3. 如果不是，嘗試將前驅節點的狀態置為 SIGNAL (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) && // 判斷當前節點有效前驅節點的執行緒資訊是否為空 pred.thread != null) { // 上述條件滿足 Node next = node.next; // 將當前節點有效前驅節點的後繼節點指標指向當前節點的後繼節點 if (next != null && next.waitStatus <= 0) compareAndSetNext(pred, predNext, next); } else { // 如果當前節點的前驅節點是頭節點，或者上述其他條件不滿足，就喚醒當前節點的後繼節點 unparkSuccessor(node); } node.next = node; // help GC } ``` 看到這個註釋你可能有些亂了，其核心目的就是從等待佇列中移除 CANCELLED 的節點，並重新拼接整個佇列，總結來看，其實設定 CANCELLED 狀態節點只是有三種情況，我們通過畫圖來分析一下：  ---  ---  至此，獲取同步狀態的過程就結束了，我們簡單的用流程圖說明一下整個過程  獲取鎖的過程就這樣的結束了，先暫停幾分鐘整理一下自己的思路。我們上面還沒有說明 SIGNAL 的作用， SIGNAL 狀態訊號到底是幹什麼用的？這就涉及到鎖的釋放了，我們來繼續瞭解，整體思路和鎖的獲取是一樣的， 但是釋放過程就相對簡單很多了 ### 獨佔式釋放同步狀態 故事要從 unlock() 方法說起 ```java public void unlock() { // 釋放鎖 sync.release(1); } ``` 呼叫 AQS 模版方法 release，進入該方法 ```java public final boolean release(int arg) { // 呼叫自定義同步器重寫的 tryRelease 方法嘗試釋放同步狀態 if (tryRelease(arg)) { // 釋放成功，獲取頭節點 Node h = head; // 存在頭節點，並且waitStatus不是初始狀態 // 通過獲取的過程我們已經分析了，在獲取的過程中會將 waitStatus的值從初始狀態更新成 SIGNAL 狀態 if (h != null && h.waitStatus != 0) // 解除執行緒掛起狀態 unparkSuccessor(h); return true; } return false; } ``` 檢視 unparkSuccessor 方法，實際是要喚醒頭節點的後繼節點 ```java private void unparkSuccessor(Node node) { // 獲取頭節點的waitStatus int ws = node.waitStatus; if (ws < 0) // 清空頭節點的waitStatus值，即置為0 compareAndSetWaitStatus(node, ws, 0); // 獲取頭節點的後繼節點 Node s = node.next; // 判斷當前節點的後繼節點是否是取消狀態，如果是，需要移除，重新連線佇列 if (s == null || s.waitStatus > 0) { s = null; // 從尾節點向前查詢，找到佇列第一個waitStatus狀態小於0的節點 for (Node t = tail; t != null && t != node; t = t.prev) // 如果是獨佔式，這裡小於0，其實就是 SIGNAL if (t.waitStatus <= 0) s = t; } if (s != null) // 解除執行緒掛起狀態 LockSupport.unpark(s.thread); } ``` 有同學可能有疑問： > 為什麼這個地方是從佇列尾部向前查詢不是 CANCELLED 的節點？ 原因有兩個： 第一，先回看節點加入佇列的情景： ```java private Node addWaiter(Node mode) { Node node = new Node(Thread.currentThread(), mode); // Try the fast path of enq; backup to full enq on failure Node pred = tail; if (pred != null) { node.prev = pred; if (compareAndSetTail(pred, node)) { pred.next = node; return node; } } enq(node); return node; } ``` 節點入隊並不是原子操作，程式碼第6、7行 ```java node.prev = pred; compareAndSetTail(pred, node) ``` 這兩個地方可以看作是尾節點入隊的原子操作，如果此時程式碼還沒執行到 pred.next = node; 這時又恰巧執行了unparkSuccessor方法，就沒辦法從前往後找了，因為後繼指標還沒有連線起來，所以需要從後往前找 第二點原因，在上面圖解產生 CANCELLED 狀態節點的時候，先斷開的是 Next 指標，Prev指標並未斷開，因此這也是必須要從後往前遍歷才能夠遍歷完全部的Node 同步狀態至此就已經成功釋放了，之前獲取同步狀態被掛起的執行緒就會被喚醒，繼續從下面程式碼第 3 行返回執行： ```java private final boolean parkAndCheckInterrupt() { LockSupport.park(this); return Thread.interrupted(); } ``` 繼續返回上層呼叫棧, 從下面程式碼15行開始執行，重新執行迴圈，再次嘗試獲取同步狀態 ```java final boolean acquireQueued(final Node node, int arg) { boolean failed = true; try { boolean interrupted = false; for (;;) { final Node p = node.predecessor(); if (p == head && tryAcquire(arg)) { setHead(node); p.next = null; // help GC failed = false; return interrupted; } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } } ``` 到這裡，關於獨佔式獲取/釋放鎖的流程已經閉環了，但是關於 AQS 的另外兩個模版方法還沒有介紹 - `響應中斷` - `超時限制`  ### 獨佔式響應中斷獲取同步狀態 故事要從lock.lockInterruptibly() 方法說起 ```java public void lockInterruptibly() throws InterruptedException { // 呼叫同步器模版方法可中斷式獲取同步狀態 sync.acquireInterruptibly(1); } ``` 有了前面的理解，理解獨佔式可響應中斷的獲取同步狀態方式，真是一眼就能明白了： ```java public final void acquireInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); // 嘗試非阻塞式獲取同步狀態失敗，如果沒有獲取到同步狀態，執行程式碼7行 if (!tryAcquire(arg)) doAcquireInterruptibly(arg); } ``` 繼續檢視 `doAcquireInterruptibly` 方法： ```java private void doAcquireInterruptibly(int arg) throws InterruptedException { final Node node = addWaiter(Node.EXCLUSIVE); boolean failed = true; try { for (;;) { final Node p = node.predecessor(); if (p == head && tryAcquire(arg)) { setHead(node); p.next = null; // help GC failed = false; return; } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) // 獲取中斷訊號後，不再返回 interrupted = true 的值，而是直接丟擲 InterruptedException throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } } ``` 沒想到 JDK 內部也有如此相近的程式碼，可響應中斷獲取鎖沒什麼深奧的，就是被中斷丟擲 InterruptedException 異常（程式碼第17行），這樣就逐層返回上層呼叫棧捕獲該異常進行下一步操作了 趁熱打鐵，來看看另外一個模版方法： ### 獨佔式超時限制獲取同步狀態 這個很好理解，就是給定一個時限，在該時間段內獲取到同步狀態，就返回 true， 否則，返回 false。好比執行緒給自己定了一個鬧鐘，鬧鈴一響，執行緒就自己返回了，這就不會使自己是阻塞狀態了 既然涉及到超時限制，其核心邏輯肯定是計算時間間隔，因為在超時時間內，肯定是多次嘗試獲取鎖的，每次獲取鎖肯定有時間消耗，所以計算時間間隔的邏輯就像我們在程式列印程式耗時 log 那麼簡單 > nanosTimeout = deadline - System.nanoTime() 故事要從 `lock.tryLock(time, unit)` 方法說起 ```java public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { // 呼叫同步器模版方法，可響應中斷和超時時間限制 return sync.tryAcquireNanos(1, unit.toNanos(time)); } ``` 來看 tryAcquireNanos 方法 ```java public final boolean tryAcquireNanos(int arg, long nanosTimeout) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); return tryAcquire(arg) || doAcquireNanos(arg, nanosTimeout); } ``` 是不是和上面 `acquireInterruptibly` 方法長相很詳細了，繼續檢視來 doAcquireNanos 方法，看程式, 該方法也是 throws InterruptedException，我們在中斷文章中說過，方法標記上有 `throws InterruptedException` 說明該方法也是可以響應中斷的，所以你可以理解超時限制是 `acquireInterruptibly` 方法的加強版，具有超時和非阻塞控制的雙保險 ```java private boolean doAcquireNanos(int arg, long nanosTimeout) throws InterruptedException { // 超時時間內，為獲取到同步狀態，直接返回false if (nanosTimeout <= 0L) return false; // 計算超時截止時間 final long deadline = System.nanoTime() + nanosTimeout; // 以獨佔方式加入到同步佇列中 final Node node = addWaiter(Node.EXCLUSIVE); boolean failed = true; try { for (;;) { final Node p = node.predecessor(); if (p == head && tryAcquire(arg)) { setHead(node); p.next = null; // help GC failed = false; return true; } // 計算新的超時時間 nanosTimeout = deadline - System.nanoTime(); // 如果超時，直接返回 false if (nanosTimeout <= 0L) return false; if (shouldParkAfterFailedAcquire(p, node) && // 判斷是最新超時時間是否大於閾值 1000 nanosTimeout > spinForTimeoutThreshold) // 掛起執行緒 nanosTimeout 長時間，時間到，自動返回 LockSupport.parkNanos(this, nanosTimeout); if (Thread.interrupted()) throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } } ``` 上面的方法應該不是很難懂，但是又同學可能在第 27 行上有所困惑 > 為什麼 nanosTimeout 和 自旋超時閾值1000進行比較？ ``` /** * The number of nanoseconds for which it is faster to spin * rather than to use timed park. A rough estimate suffices * to improve responsiveness with very short timeouts. */ static final long spinForTimeoutThreshold = 1000L; ``` 其實 doc 說的很清楚，說白了，1000 nanoseconds 時間已經非常非常短暫了，沒必要再執行掛起和喚醒操作了，不如直接當前執行緒直接進入下一次迴圈 到這裡，我們自定義的 MyMutex 只差 Condition 沒有說明了，不知道你累了嗎？我還在堅持  ### Condition 如果你看過之前寫的 [併發程式設計之等待通知機制](https://dayarch.top/p/waiting-notification-mechanism.html) ，你應該對下面這個圖是有印象的：  如果當時你理解了這個模型，再看 Condition 的實現，根本就不是問題了，首先 Condition 還是一個介面，肯定也是需要有實現類的  那故事就從 `lock.newnewCondition` 說起吧 ```java public Condition newCondition() { // 使用自定義的條件 return sync.newCondition(); } ``` 自定義同步器重封裝了該方法： ```java Condition newCondition() { return new ConditionObject(); } ``` ConditionObject 就是 Condition 的實現類，該類就定義在了 AQS 中，只有兩個成員變數： ```java /** First node of condition queue. */ private transient Node firstWaiter; /** Last node of condition queue. */ private transient Node lastWaiter; ``` 所以，我們只需要來看一下 ConditionObject 實現的 await / signal 方法來使用這兩個成員變數就可以了 ```java public final void await() throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); // 同樣構建 Node 節點，並加入到等待佇列中 Node node = addConditionWaiter(); // 釋放同步狀態 int savedState = fullyRelease(node); int interruptMode = 0; while (!isOnSyncQueue(node)) { // 掛起當前執行緒 LockSupport.park(this); if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) break; } if (acquireQueued(node, savedState) && interruptMode != THROW_IE) interruptMode = REINTERRUPT; if (node.nextWaiter != null) // clean up if cancelled unlinkCancelledWaiters(); if (interruptMode != 0) reportInterruptAfterWait(interruptMode); } ``` 這裡注意用詞，在介紹獲取同步狀態時，addWaiter 是加入到【同步佇列】，就是上圖說的入口等待佇列，這裡說的是【等待佇列】，所以 addConditionWaiter 肯定是構建了一個自己的佇列： ```java private Node addConditionWaiter() { Node t = lastWaiter; if (t != null && t.waitStatus != Node.CONDITION) { unlinkCancelledWaiters(); t = lastWaiter; } // 新構建的節點的 waitStatus 是 CONDITION，注意不是 0 或 SIGNAL 了 Node node = new Node(Thread.currentThread(), Node.CONDITION); // 構建單向同步佇列 if (t == null) firstWaiter = node; else t.nextWaiter = node; lastWaiter = node; return node; } ``` 這裡有朋友可能會有疑問： > 為什麼這裡是單向佇列，也沒有使用CAS 來保證加入佇列的安全性呢？ 因為 await 是 Lock 正規化 try 中使用的，說明已經獲取到鎖了，所以就沒必要使用 CAS 了，至於是單向，因為這裡還不涉及到競爭鎖，只是做一個條件等待佇列 在 Lock 中可以定義多個條件，每個條件都會對應一個 條件等待佇列，所以將上圖豐富說明一下就變成了這個樣子：  執行緒已經按相應的條件加入到了條件等待佇列中，那如何再嘗試獲取鎖呢？signal / signalAll 方法就已經排上用場了 ```java public final void signal() { if (!isHeldExclusively()) throw new IllegalMonitorStateException(); Node first = firstWaiter; if (first != null) doSignal(first); } ``` Signal 方法通過呼叫 doSignal 方法，只喚醒條件等待佇列中的第一個節點 ```java private void doSignal(Node first) { do { if ( (firstWaiter = first.nextWaiter) == null) lastWaiter = null; first.nextWaiter = null; // 呼叫該方法，將條件等待佇列的執行緒節點移動到同步佇列中 } while (!transferForSignal(first) && (first = firstWaiter) != null); } ``` 繼續看 `transferForSignal` 方法 ```java final boolean transferForSignal(Node node) { if (!compareAndSetWaitStatus(node, Node.CONDITION, 0)) return false; // 重新進行入隊操作 Node p = enq(node); int ws = p.waitStatus; if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)) // 喚醒同步佇列中該執行緒 LockSupport.unpark(node.thread); return true; } ``` 所以我們再用圖解一下喚醒的整個過程  到這裡，理解 signalAll 就非常簡單了，只不過迴圈判斷是否還有 nextWaiter，如果有就像 signal 操作一樣，將其從條件等待佇列中移到同步佇列中 ```java private void doSignalAll(Node first) { lastWaiter = firstWaiter = null; do { Node next = first.nextWaiter; first.nextWaiter = null; transferForSignal(first); first = next; } while (first != null); } ``` 不知你還是否記得，我在[併發程式設計之等待通知機制](https://dayarch.top/p/waiting-notification-mechanism.html) 中還說過一句話 > 沒有特殊原因儘量用 signalAll 方法 什麼時候可以用 signal 方法也在其中做了說明，請大家自行檢視吧  這裡我還要多說一個細節，從條件等待佇列移到同步佇列是有時間差的，所以使用 await() 方法也是正規化的， 同樣在該文章中做了解釋  有時間差，就會有公平和不公平的問題，想要全面瞭解這個問題，我們就要走近 ReentrantLock 中來看了，除了瞭解公平/不公平問題，檢視 ReentrantLock 的應用還是要反過來驗證它使用的AQS的，我們繼續吧 ## ReentrantLock 是如何應用的AQS 獨佔式的典型應用就是 ReentrantLock 了，我們來看看它是如何重寫這個方法的  乍一看挺奇怪的，怎麼裡面自定義了三個同步器：其實 NonfairSync，FairSync 只是對 Sync 做了進一步劃分：  從名稱上你應該也知道了，這就是你聽到過的 `公平鎖/非公平鎖`了 ### 何為公平鎖/非公平鎖？ 生活中，排隊講求先來後到視為公平。程式中的公平性也是符合請求鎖的絕對時間的，其實就是 FIFO，否則視為不公平 我們來對比一下 ReentrantLock 是如何實現公平鎖和非公平鎖的  其實沒什麼大不了，公平鎖就是判斷同步佇列是否還有先驅節點的存在，只有沒有先驅節點才能獲取鎖；而非公平鎖是不管這個事的，能獲取到同步狀態就可以，就這麼簡單，那問題來了： ### 為什麼會有公平鎖/非公平鎖的設計？ 考慮這個問題，我們需重新回憶上面的鎖獲取實現圖了，其實上面我已經透露了一點  主要有兩點原因： #### 原因一： 恢復掛起的執行緒到真正鎖的獲取還是有時間差的，從人類的角度來看這個時間微乎其微，但是從CPU的角度來看，這個時間差存在的還是很明顯的。所以非公平鎖能更充分的利用 CPU 的時間片，儘量減少 CPU 空閒狀態時間 #### 原因二： 不知你是否還記得我在 [面試問，建立多少個執行緒合適？](https://dayarch.top/p/how-many-threads-should-be-created.html) 文章中反覆提到過，使用多執行緒很重要的考量點是執行緒切換的開銷，想象一下，如果採用非公平鎖，當一個執行緒請求鎖獲取同步狀態，然後釋放同步狀態，因為不需要考慮是否還有前驅節點，所以剛釋放鎖的執行緒在此刻再次獲取同步狀態的機率就變得非常大，所以就減少了執行緒的開銷  相信到這裡，你也就明白了，為什麼 ReentrantLock 預設構造器用的是非公平鎖同步器 ```java public ReentrantLock() { sync = new NonfairSync(); } ``` 看到這裡，感覺非公平鎖 perfect，非也，有得必有失 > 使用公平鎖會有什麼問題？ 公平鎖保證了排隊的公平性，非公平鎖霸氣的忽視這個規則，所以就有可能導致排隊的長時間在排隊，也沒有機會獲取到鎖，這就是傳說中的 **“飢餓”** ### 如何選擇公平鎖/非公平鎖？ 相信到這裡，答案已經在你心中了，如果為了更高的吞吐量，很顯然非公平鎖是比較合適的，因為節省很多執行緒切換時間，吞吐量自然就上去了，否則那就用公平鎖還大家一個公平 我們還差最後一個環節，真的要挺住 ### 可重入鎖 到這裡，我們還沒分析 ReentrantLock 的名字，JDK 起名這麼有講究，肯定有其含義，直譯過來【可重入鎖】 > 為什麼要支援鎖的重入？ 試想，如果是一個有 synchronized 修飾的遞迴呼叫方法，程式第二次進入被自己阻塞了豈不是很大的笑話，所以 synchronized 是支援鎖的重入的 Lock 是新輪子，自然也要支援這個功能，其實現也很簡單，請檢視公平鎖和非公平鎖對比圖，其中有一段程式碼： ```java // 判斷當前執行緒是否和已佔用鎖的執行緒是同一個 else if (current == getExclusiveOwnerThread()) ``` 仔細看程式碼， 你也許發現，我前面的一個說明是錯誤的，我要重新解釋一下  重入的執行緒會一直將 state + 1， 釋放鎖會 state - 1直至等於0，上面這樣寫也是想幫助大家快速的區分 ## 總結 本文是一個長文，說明了為什麼要造 Lock 新輪子，如何標準的使用 Lock，AQS 是什麼，是如何實現鎖的，結合 ReentrantLock 反推 AQS 中的一些應用以及其獨有的一些特性 獨佔式獲取鎖就這樣介紹完了，我們還差 AQS 共享式 `xxxShared` 沒有分析，結合共享式，接下來我們來閱讀一下 Semaphore，ReentrantReadWriteLock 和 CountLatch 等 最後，也歡迎大家的留言，如有錯誤之處還請指出。我的手痠了，眼睛幹了，我去準備擼下一篇.....  ## 靈魂追問 1. 為什麼更改 state 有 setState() , compareAndSetState() 兩種方式，感覺後者更安全，但是鎖的視線中有好多地方都使用了 setState()，安全嗎？ 2. 下面程式碼是一個轉賬程式，是否存在死鎖或者鎖的其他問題呢？ ```java class Account { private int balance; private final Lock lock = new ReentrantLock(); // 轉賬 void transfer(Account tar, int amt){ while (true) { if(this.lock.tryLock()) { try { if (tar.lock.tryLock()) { try { this.balance -= amt; tar.balance += amt; } finally { tar.lock.unlock(); } }//if } finally { this.lock.unlock(); } }//if }//while }//transfer } ``` ## 參考 1. Java 併發實戰 2. Java 併發程式設計的藝術 3. https://tech.meituan.com/2019/12/05/aqs-theory-and-app