硬核乾貨：5W字17張高清圖理解同步器框架AbstractQueuedSynchronizer
阿新 • • 發佈：2020-07-24
## 前提
併發程式設計大師**Doug Lea**在編寫`JUC`（`java.util.concurrent`）包的時候引入了`java.util.concurrent.locks.AbstractQueuedSynchronizer`，其實是`Abstract Queued Synchronizer`，也就是"基於佇列實現的抽象同步器"，一般我們稱之為`AQS`。其實`Doug Lea`大神編寫`AQS`是有嚴謹的理論基礎的，他的個人部落格上有一篇論文[《The java.util.concurrent Synchronizer Framewor》](http://gee.cs.oswego.edu/dl/papers/aqs.pdf)，可以在網際網路找到相應的譯文《JUC同步器框架》，如果想要深入研究`AQS`必須要理解一下該論文的內容，然後結合論文內容詳細分析一下`AQS`的原始碼實現。本文在閱讀`AQS`原始碼的時候選用的`JDK`版本是`JDK11`。
> 出於寫作習慣，下文會把AbstractQueuedSynchronizer稱為AQS、JUC同步器框或者同步器框架。
## AQS的主要功能
`AQS`是`JUC`包中用於構建鎖或者其他同步元件（訊號量、事件等）的基礎框架類。`AQS`從它的實現上看主要提供了下面的功能：
- 同步狀態的原子性管理。
- 執行緒的阻塞和解除阻塞。
- 提供阻塞執行緒的儲存佇列。
基於這三大功能，衍生出下面的附加功能：
- 通過中斷實現的任務取消，此功能基於執行緒中斷實現。
- 可選的超時設定，也就是呼叫者可以選擇放棄等待任務執行完畢直接返回。
- 定義了`Condition介面`，用於支援管程形式的`await/signal/signalAll`操作，代替了`Object`類基於`JNI`提供的`wait/notify/notifyAll`。
`AQS`還根據同步狀態的不同管理方式區分為兩種不同的實現：**獨佔狀態的同步器**和**共享狀態的同步器**。
## 同步器框架基本原理
[《The java.util.concurrent Synchronizer Framework》](http://gee.cs.oswego.edu/dl/papers/aqs.pdf)一文中其實有提及到同步器框架的虛擬碼：
```java
// acquire操作如下：
while (synchronization state does not allow acquire) {
enqueue current thread if not already queued;
possibly block current thread;
}
dequeue current thread if it was queued;
//release操作如下：
update synchronization state;
if (state may permit a blocked thread to acquire){
unblock one or more queued threads;
}
```
撇腳翻譯一下：
```java
// acquire操作如下：
while(同步狀態申請獲取失敗){
if(當前執行緒未進入等待佇列){
當前執行緒放入等待佇列;
}
嘗試阻塞當前執行緒;
}
當前執行緒移出等待佇列
//release操作如下：
更新同步狀態
if(同步狀態足夠允許一個阻塞的執行緒申請獲取){
解除一個或者多個等待佇列中的執行緒的阻塞狀態;
}
```
為了實現上述操作，需要下面三個基本環節的相互協作：
- 同步狀態的原子性管理。
- 等待佇列的管理。
- 執行緒的阻塞與解除阻塞。
其實基本原理很簡單，但是為了應對複雜的併發場景和併發場景下程式執行的正確性，同步器框架在上面的`acquire`操作和`release`操作中使用了**大量的死迴圈和`CAS`等操作**，再加上`Doug Lea`喜歡使用單行復雜的條件判斷程式碼，如**一個`if`條件語句會包含大量操作**，`AQS`很多時候會讓人感覺實現邏輯過於複雜。
### 同步狀態管理
`AQS`內部內部定義了一個`32`位整型的`state`變數用於儲存同步狀態：
```java
/**
* The synchronization state.（同步狀態值）
*/
private volatile int state;
// 獲取state
protected final int getState() {
return state;
}
// 直接覆蓋設定state
protected final void setState(int newState) {
state = newState;
}
// CAS設定state
protected final boolean compareAndSetState(int expect, int update) {
return STATE.compareAndSet(this, expect, update);
}
```
同步狀態`state`在不同的實現中可以有不同的作用或者表示意義，這裡其實不能單純把它理解為中文意義上的"狀態"，它可以代表資源數、鎖狀態等等，下文遇到具體的場景我們再分析它表示的意義。
### CLH佇列與變體
`CLH`鎖即`Craig, Landin, and Hagersten (CLH) locks`，因為它底層是基於佇列實現，一般也稱為`CLH`佇列鎖。`CLH`鎖也是一種基於連結串列的可擴充套件、高效能、公平的自旋鎖，申請執行緒僅僅在本地變數上自旋，它不斷輪詢前驅的狀態，假設發現前驅釋放了鎖就結束自旋。從實現上看，`CLH`鎖是一種自旋鎖，能確保無飢餓性，提供先來先服務的公平性。先看簡單的`CLH`鎖的一個簡單實現：
```java
public class CLHLock implements Lock {
AtomicReference tail = new AtomicReference<>(new QueueNode());
ThreadLocal pred;
ThreadLocal current;
public CLHLock() {
current = ThreadLocal.withInitial(QueueNode::new);
pred = ThreadLocal.withInitial(() -> null);
}
@Override
public void lock() {
QueueNode node = current.get();
node.locked = true;
QueueNode pred = tail.getAndSet(node);
this.pred.set(pred);
while (pred.locked) {
}
}
@Override
public void unlock() {
QueueNode node = current.get();
node.locked = false;
current.set(this.pred.get());
}
static class QueueNode {
boolean locked;
}
// 忽略其他介面方法的實現
}
```
上面是一個簡單的`CLH`佇列鎖的實現，內部類`QueueNode`只使用了一個簡單的布林值`locked`屬性記錄了每個執行緒的狀態，如果該屬性為`true`，則相應的執行緒要麼已經獲取到鎖，要麼正在等待鎖，如果該屬性為`false`，則相應的執行緒已經釋放了鎖。新來的想要獲取鎖的執行緒必須對`tail`屬性呼叫`getAndSet()`方法，使得自身成為佇列的尾部，同時得到一個指向前驅節點的引用`pred`，最後執行緒所在節點在其前驅節點的`locked`屬性上自旋，直到前驅節點釋放鎖。上面的實現是無法執行的，因為一旦自旋就會進入死迴圈導致`CPU`飆升，可以嘗試使用下文將要提到的`LockSupport`進行改造。
`CLH`佇列鎖本質是使用佇列（實際上是單向連結串列）存放等待獲取鎖的執行緒，等待的執行緒總是在其所在節點的前驅節點的狀態上自旋，直到前驅節點釋放資源。**從實際來看，過度自旋帶來的CPU效能損耗比較大，並不是理想的執行緒等待佇列的實現**。
基於原始的`CLH`佇列鎖中提供的等待佇列的基本原理，**`AQS`實現一種了CLH鎖佇列的變體（Variant）**。`AQS`類的`protected`修飾的建構函式裡面有一大段註釋用於說明`AQS`實現的等待佇列的細節事項，這裡列舉幾點重要的：
- `AQS`實現的等待佇列沒有直接使用`CLH`鎖佇列，但是參考了其設計思路，等待節點會儲存前驅節點中執行緒的資訊，內部也會維護一個控制執行緒阻塞的狀態值。
- 每個節點都設計為一個持有單獨的等待執行緒並且"帶有具體的通知方式"的監視器，這裡所謂通知方式就是自定義喚醒阻塞執行緒的方式而已。
- 一個執行緒是等待佇列中的第一個等待節點的持有執行緒會嘗試獲取鎖，但是並不意味著它一定能夠獲取鎖成功（這裡的意思是存在公平和非公平的實現），獲取失敗就要重新等待。
- 等待佇列中的節點通過`prev`屬性連線前驅節點，通過`next`屬性連線後繼節點，簡單來說，就是**雙向連結串列的設計**。
- `CLH`佇列本應該需要一個虛擬的頭節點，但是在`AQS`中沒有直接提供虛擬的頭節點，而是延遲到第一次競爭出現的時候懶建立虛擬的頭節點（其實也會建立尾節點，初始化時頭尾節點是同一個節點）。
- `Condition`（條件）等待佇列中的阻塞執行緒使用的是相同的`Node`結構，但是提供了另一個連結串列用來存放，`Condition`等待佇列的實現比非`Condition`等待佇列複雜。
### 執行緒阻塞與喚醒
執行緒的阻塞和喚醒在`JDK1.5`之前，一般只能依賴於`Object`類提供的`wait()`、`notify()`和`notifyAll()`方法，它們都是`JNI`方法，由`JVM`提供實現，並且它們必須執行在獲取監視器鎖的程式碼塊內（`synchronized`程式碼塊中），這個侷限性先不談效能上的問題，程式碼的簡潔性和靈活性是比較低的。`JDK1.5`引入了`LockSupport`類，底層是基於`Unsafe`類的`park()`和`unpark()`方法，提供了執行緒阻塞和喚醒的功能，它的機制有點像只有一個允許使用資源的訊號量`java.util.concurrent.Semaphore`，也就是一個執行緒只能通過`park()`方法阻塞一次，只能呼叫`unpark()`方法解除呼叫阻塞一次，執行緒就會喚醒（多次呼叫`unpark()`方法也只會喚醒一次），可以想象是內部維護了一個0-1的計數器。
`LockSupport`類如果使用得好，可以提供更靈活的編碼方式，這裡舉個簡單的使用例子：
```java
public class LockSupportMain implements Runnable {
private static final DateTimeFormatter FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");
private Thread thread;
private void setThread(Thread thread) {
this.thread = thread;
}
public static void main(String[] args) throws Exception {
LockSupportMain main = new LockSupportMain();
Thread thread = new Thread(main, "LockSupportMain");
main.setThread(thread);
thread.start();
Thread.sleep(2000);
main.unpark();
Thread.sleep(2000);
}
@Override
public void run() {
System.out.println(String.format("%s-步入run方法,執行緒名稱:%s", FORMATTER.format(LocalDateTime.now()),
Thread.currentThread().getName()));
LockSupport.park();
System.out.println(String.format("%s-解除阻塞,執行緒繼續執行,執行緒名稱:%s", FORMATTER.format(LocalDateTime.now()),
Thread.currentThread().getName()));
}
private void unpark() {
LockSupport.unpark(thread);
}
}
// 某個時刻的執行結果如下：
2019-02-25 00:39:57.780-步入run方法,執行緒名稱:LockSupportMain
2019-02-25 00:39:59.767-解除阻塞,執行緒繼續執行,執行緒名稱:LockSupportMain
```
`LockSupport`類`park()`方法也有帶超時的變體版本方法，遇到帶超時期限阻塞等待場景下不妨可以使用`LockSupport#parkNanos()`。
## 獨佔執行緒的儲存
`AbstractOwnableSynchronizer`是`AQS`的父類，一個同步器框架有可能在一個時刻被某一個執行緒獨佔，`AbstractOwnableSynchronizer`就是為所有的同步器實現和鎖相關實現提供了基礎的儲存、獲取和設定獨佔執行緒的功能，這個類的原始碼很簡單：
```java
public abstract class AbstractOwnableSynchronizer
implements java.io.Serializable {
private static final long serialVersionUID = 3737899427754241961L;
protected AbstractOwnableSynchronizer() { }
// 當前獨佔執行緒的瞬時例項 - 提供Getter和Setter方法
private transient Thread exclusiveOwnerThread;
protected final void setExclusiveOwnerThread(Thread thread) {
exclusiveOwnerThread = thread;
}
protected final Thread getExclusiveOwnerThread() {
return exclusiveOwnerThread;
}
}
```
它就提供了一個儲存獨佔執行緒的變數對應的`Setter`和`Getter`方法，方法都是`final`修飾的，子類只能使用不能覆蓋。
## CLH佇列變體的實現
這裡先重點分析一下`AQS`中等待佇列的節點`AQS`的靜態內部類`Node`的原始碼：
```java
static final class Node {
// 標記一個節點處於共享模式下的等待
static final Node SHARED = new Node();
// 標記一個節點處於獨佔模式下的等待
static final Node EXCLUSIVE = null;
// 取消狀態
static final int CANCELLED = 1;
// 喚醒狀態
static final int SIGNAL = -1;
// 條件等待狀態
static final int CONDITION = -2;
// 傳播狀態
static final int PROPAGATE = -3;
// 等待狀態，初始值為0，其他可選值是上面的4個值
volatile int waitStatus;
// 當前節點前驅節點的引用
volatile Node prev;
// 當前節點後繼節點的引用
volatile Node next;
// 當前節點持有的執行緒，可能是阻塞中等待喚醒的執行緒
volatile Thread thread;
// 下一個等待節點
Node nextWaiter;
// 當前操作的節點是否處於共享模式
final boolean isShared() {
return nextWaiter == SHARED;
}
// 獲取當前節點的前驅節點，確保前驅節點必須存在，否則丟擲NPE
final Node predecessor() {
Node p = prev;
if (p == null)
throw new NullPointerException();
else
return p;
}
// 空節點，主要是首次建立佇列的時候建立的頭和尾節點使用
Node() {}
// 設定下一個等待節點，設定持有執行緒為當前執行緒
Node(Node nextWaiter) {
this.nextWaiter = nextWaiter;
THREAD.set(this, Thread.currentThread());
}
// 設定waitStatus，設定持有執行緒為當前執行緒
Node(int waitStatus) {
WAITSTATUS.set(this, waitStatus);
THREAD.set(this, Thread.currentThread());
}
// CAS更新waitStatus
final boolean compareAndSetWaitStatus(int expect, int update) {
return WAITSTATUS.compareAndSet(this, expect, update);
}
// CAS設定後繼節點
final boolean compareAndSetNext(Node expect, Node update) {
return NEXT.compareAndSet(this, expect, update);
}
// 設定前驅節點
final void setPrevRelaxed(Node p) {
PREV.set(this, p);
}
// 下面是變數控制代碼的實現，在VarHandle出現之前使用的是Unsafe，其實底層還是照樣使用Unsafe
private static final VarHandle NEXT;
private static final VarHandle PREV;
private static final VarHandle THREAD;
private static final VarHandle WAITSTATUS;
static {
try {
MethodHandles.Lookup l = MethodHandles.lookup();
NEXT = l.findVarHandle(Node.class, "next", Node.class);
PREV = l.findVarHandle(Node.class, "prev", Node.class);
THREAD = l.findVarHandle(Node.class, "thread", Thread.class);
WAITSTATUS = l.findVarHandle(Node.class, "waitStatus", int.class);
} catch (ReflectiveOperationException e) {
throw new ExceptionInInitializerError(e);
}
}
}
```
其中，變數控制代碼（`VarHandle`）是`JDK9`引入的新特性，其實底層依賴的還是`Unsafe`的方法，筆者認為可以簡單理解它為`Unsafe`的門面類，而定義的方法基本都是面向變數屬性的操作。這裡需要關注一下`Node`裡面的幾個屬性：
- `waitStatus`：當前`Node`例項的等待狀態，可選值有5個。
1. 初始值整數0：當前節點如果不指定初始化狀態值，預設值就是0，側面說明節點正在等待佇列中處於等待狀態。
2. `Node#CANCELLED`整數值1：表示當前節點例項因為超時或者執行緒中斷而被取消，等待中的節點永遠不會處於此狀態，被取消的節點中的執行緒例項不會阻塞。
3. `Node#SIGNAL`整數值-1：表示當前節點的後繼節點是（或即將是）阻塞的（通過`LockSupport#park()`），當它釋放或取消時，當前節點必須`LockSupport#unpark()`它的後繼節點。
4. `Node#CONDITION`整數值-2：表示當前節點是條件佇列中的一個節點，當它轉換為同步佇列中的節點的時候，狀態會被重新設定為0。
5. `Node#PROPAGATE`整數值-3：此狀態值通常只設置到呼叫了`doReleaseShared()`方法的頭節點，確保`releaseShared()`方法的呼叫可以傳播到其他的所有節點，簡單理解就是共享模式下節點釋放的傳遞標記。
- `prev`、`next`：當前`Node`例項的前驅節點引用和後繼節點引用。
- `thread`：當前`Node`例項持有的執行緒例項引用。
- `nextWaiter`：這個值是一個比較容易令人生疑的值，雖然表面上它稱為"下一個等待的節點"，但是實際上它有三種取值的情況。
1. 值為靜態例項`Node.EXCLUSIVE`（也就是null），代表當前的`Node`例項是獨佔模式。
2. 值為靜態例項`Node.SHARED`，代表當前的`Node`例項是共享模式。
3. 值為非`Node.EXCLUSIVE`和`Node.SHARED`的其他節點例項，**代表Condition等待佇列中當前節點的下一個等待節點**。
`Node`類的等待狀態`waitStatus`理解起來是十分費勁的，下面分析`AQS`其他原始碼段的時候會**標識此狀態變化的時機**。
其實上面的`Node`類可以直接拷貝出來當成一個新建的類，然後嘗試構建一個雙向連結串列自行除錯，這樣子就能深刻它的資料結構。例如：
```java
public class AqsNode {
static final AqsNode SHARED = new AqsNode();
static final AqsNode EXCLUSIVE = null;
static final int CANCELLED = 1;
static final int SIGNAL = -1;
static final int CONDITION = -2;
static final int PROPAGATE = -3;
volatile int waitStatus;
volatile AqsNode prev;
volatile AqsNode next;
volatile Thread thread;
AqsNode nextWaiter;
final boolean isShared() {
return nextWaiter == SHARED;
}
final AqsNode predecessor() {
AqsNode p = prev;
if (p == null)
throw new NullPointerException();
else
return p;
}
AqsNode() {
}
AqsNode(AqsNode nextWaiter) {
this.nextWaiter = nextWaiter;
THREAD.set(this, Thread.currentThread());
}
AqsNode(int waitStatus) {
WAITSTATUS.set(this, waitStatus);
THREAD.set(this, Thread.currentThread());
}
final boolean compareAndSetWaitStatus(int expect, int update) {
return WAITSTATUS.compareAndSet(this, expect, update);
}
final boolean compareAndSetNext(AqsNode expect, AqsNode update) {
return NEXT.compareAndSet(this, expect, update);
}
final void setPrevRelaxed(AqsNode p) {
PREV.set(this, p);
}
private static final VarHandle NEXT;
private static final VarHandle PREV;
private static final VarHandle THREAD;
private static final VarHandle WAITSTATUS;
static {
try {
MethodHandles.Lookup l = MethodHandles.lookup();
NEXT = l.findVarHandle(AqsNode.class, "next", AqsNode.class);
PREV = l.findVarHandle(AqsNode.class, "prev", AqsNode.class);
THREAD = l.findVarHandle(AqsNode.class, "thread", Thread.class);
WAITSTATUS = l.findVarHandle(AqsNode.class, "waitStatus", int.class);
} catch (ReflectiveOperationException e) {
throw new ExceptionInInitializerError(e);
}
}
public static void main(String[] args) throws Exception {
AqsNode head = new AqsNode();
AqsNode next = new AqsNode(AqsNode.EXCLUSIVE);
head.next = next;
next.prev = head;
AqsNode tail = new AqsNode(AqsNode.EXCLUSIVE);
next.next = tail;
tail.prev = next;
List threads = new ArrayList<>();
for (AqsNode node = head; node != null; node = node.next) {
threads.add(node.thread);
}
System.out.println(threads);
}
}
// 某次執行的輸出：
[null, Thread[main,5,main], Thread[main,5,main]]
```
實際上，`AQS`中一共存在兩種等待佇列，其中一種是普通的同步等待佇列，這裡命名為`Sync Queue`，另一種是基於`Sync Queue`實現的條件等待佇列，這裡命名為`Condition Queue`。
### 理解同步等待佇列
前面已經介紹完`AQS`的同步等待佇列節點類，下面重點分析一下同步等待佇列的相關原始碼，**下文的Sync佇列、Sync Queue、同步佇列和同步等待佇列是同一個東西**。首先，我們通過分析`Node`節點得知`Sync`佇列一定是雙向連結串列，`AQS`中有兩個瞬時成員變數用來存放頭節點和尾節點：
```java
// 頭節點引用（注意由transient volatile修飾，不會序列化，並且寫操作會馬上重新整理到主記憶體）
private transient volatile Node head;
// 尾節點引用（注意由transient volatile修飾，不會序列化，並且寫操作會馬上重新整理到主記憶體）
private transient volatile Node tail;
// 變數控制代碼相關，用於CAS操作頭尾節點
private static final VarHandle STATE;
private static final VarHandle HEAD;
private static final VarHandle TAIL;
static {
try {
MethodHandles.Lookup l = MethodHandles.lookup();
STATE = l.findVarHandle(AbstractQueuedSynchronizer.class, "state", int.class);
HEAD = l.findVarHandle(AbstractQueuedSynchronizer.class, "head", Node.class);
TAIL = l.findVarHandle(AbstractQueuedSynchronizer.class, "tail", Node.class);
} catch (ReflectiveOperationException e) {
throw new ExceptionInInitializerError(e);
}
// 確保LockSupport類已經初始化 - 這裡應該是為了修復之前一個因為LockSupport未初始化導致的BUG
Class ensureLoaded = LockSupport.class;
}
// 初始化同步佇列，注意初始化同步佇列的時候，頭尾節點都是指向同一個新的Node例項
private final void initializeSyncQueue() {
Node h;
if (HEAD.compareAndSet(this, null, (h = new Node())))
tail = h;
}
// CAS設定同步佇列的尾節點
private final boolean compareAndSetTail(Node expect, Node update) {
return TAIL.compareAndSet(this, expect, update);
}
// 設定頭節點，重點注意這裡：傳入的節點設定成頭節點之後，前驅節點和持有的執行緒會置為null，這是因為：
// 1.頭節點一定沒有前驅節點。
// 2.當節點被設定為頭節點，它所在的執行緒一定是已經解除了阻塞。
private void setHead(Node node) {
head = node;
node.thread = null;
node.prev = null;
}
```
當前執行緒加入同步等待佇列和同步等待佇列的初始化是同一個方法，前文提到過：同步等待佇列的初始化會延遲到第一次可能出現競爭的情況，這是為了避免無謂的資源浪費，具體方法是`addWaiter(Node mode)`：
```java
// 新增等待節點到同步等待佇列，實際上初始化佇列也是這個方法完成的
private Node addWaiter(Node mode) {
// 基於當前執行緒建立一個新節點，節點的模式由呼叫者決定
Node node = new Node(mode);
for (;;) {
Node oldTail = tail;
// 尾節點不為空說明佇列已經初始化過，則把新節點加入到連結串列中，作為新的尾節點，建立和前驅節點的關聯關係
if (oldTail != null) {
node.setPrevRelaxed(oldTail);
if (compareAndSetTail(oldTail, node)) {
oldTail.next = node;
return node;
}
} else {
// 尾節點為空說明佇列尚未初始化過，進行一次初始化操作
initializeSyncQueue();
}
}
}
```
在首次呼叫`addWaiter()`方法，死迴圈至少執行兩輪再跳出，因為同步佇列必須初始化完成後（第一輪迴圈），然後再把當前執行緒所在的新節點例項新增到等待佇列中再返（第二輪迴圈）當前的節點，**這裡需要注意的是新加入同步等待佇列的節點一定是新增到佇列的尾部並且會更新`AQS`中的tail屬性為最新入隊的節點例項**。
假設我們使用`Node.EXCLUSIVE`模式把新增的等待執行緒加入佇列，例如有三個執行緒分別是`thread-1`、`thread-2`和`thread-3`，執行緒入隊的時候都處於阻塞狀態，模擬一下依次呼叫上面的入隊方法的同步佇列的整個連結串列的狀態。
先是執行緒`thread-1`加入等待佇列：
接著是執行緒`thread-2`加入等待佇列：
最後是執行緒`thread-3`加入等待佇列：
如果仔細研究會發現，如果所有的入隊執行緒都處於阻塞狀態的話，新入隊的執行緒總是新增到佇列的`tail`節點，**阻塞的執行緒總是"爭搶"著成為`head`節點，這一點和`CLH`佇列鎖的阻塞執行緒總是基於前驅節點自旋以獲取鎖的思路是一致的**。下面將會分析的**獨佔模式與共享模式，執行緒加入等待佇列都是通過`addWaiter()`方法**。
### 理解條件等待佇列
前面已經相對詳細地介紹過同步等待佇列，在`AQS`中還存在另外一種相對特殊和複雜的等待佇列-**條件等待佇列**。介紹條件等待佇列之前，要先介紹`java.util.concurrent.locks.Condition`介面。
```java
public interface Condition {
// 當前執行緒進入等待狀態直到被喚醒或者中斷
void await() throws InterruptedException;
// 當前執行緒進入等待狀態，不響應中斷，阻塞直到被喚醒
void awaitUninterruptibly();
// 當前執行緒進入等待狀態直到被喚醒或者中斷，阻塞帶時間限制
long awaitNanos(long nanosTimeout) throws InterruptedException;
// 當前執行緒進入等待狀態直到被喚醒或者中斷，阻塞帶時間限制
boolean await(long time, TimeUnit unit) throws InterruptedException;
// 當前執行緒進入等待狀態直到被喚醒或者中斷，阻塞帶時間限制
boolean awaitUntil(Date deadline) throws InterruptedException;
// 喚醒單個阻塞執行緒
void signal();
// 喚醒所有阻塞執行緒
void signalAll();
}
```
`Condition`可以理解為`Object`中的`wait()`、`notify()`和`notifyAll()`的替代品，因為`Object`中的相應方法是`JNI`（`Native`）方法，由`JVM`實現，對使用者而言並不是十分友好（有可能伴隨`JVM`版本變更而受到影響），而`Condition`是基於資料結構和相應演算法實現對應的功能，我們可以從原始碼上分析其實現。
`Condition`的實現類是`AQS`的公有內部類`ConditionObject`。`ConditionObject`提供的入佇列方法如下：
```java
public class ConditionObject implements Condition, java.io.Serializable {
private static final long serialVersionUID = 1173984872572414699L;
/** First node of condition queue. */ - 條件佇列的第一個節點
private transient Node firstWaiter;
/** Last node of condition queue. */ - 條件佇列的最後一個節點
private transient Node lastWaiter;
// 公有建構函式
public ConditionObject() { }
// 新增條件等待節點
private Node addConditionWaiter() {
// 這裡做一次判斷，當前執行緒必須步入此同步器例項
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
// 臨時節點t賦值為lastWaiter引用
Node t = lastWaiter;
// If lastWaiter is cancelled, clean out.
// 最後一個節點不為條件等待狀態，則是取消狀態
if (t != null && t.waitStatus != Node.CONDITION) {
// 解除所有取消等待的節點的連線
unlinkCancelledWaiters();
t = lastWaiter;
}
// 基於當前執行緒新建立一個條件等待型別的節點
Node node = new Node(Node.CONDITION);
// 首次建立Condition的時候，最後一個節點臨時引用t為null，則把第一個節點置為新建的節點
if (t == null)
firstWaiter = node;
else
// 已經存在第一個節點，則通過nextWaiter連線新的節點
t.nextWaiter = node;
// 最後一個節點的引用更新為新節點的引用
lastWaiter = node;
return node;
}
// 從條件等待佇列解除所有取消等待的節點的連線，其實就是所有取消節點移除的操作，涉及到雙向連結串列的斷鏈操作、第一個和最後一個節點的引用更新
private void unlinkCancelledWaiters() {
Node t = firstWaiter;
Node trail = null;
while (t != null) {
Node next = t.nextWaiter;
// 注意這裡等待狀態的判斷
if (t.waitStatus != Node.CONDITION) {
t.nextWaiter = null;
if (trail == null)
firstWaiter = next;
else
trail.nextWaiter = next;
if (next == null)
lastWaiter = trail;
}
else
trail = t;
t = next;
}
}
// 當前同步器例項持有的執行緒是否當前執行緒(currentThread())
protected boolean isHeldExclusively() {
throw new UnsupportedOperationException();
}
// 暫時不分析其他方法
}
```
實際上，`Condition`的所有`await()`方法變體都呼叫`addConditionWaiter()`新增阻塞執行緒到條件佇列中。我們按照分析同步等待佇列的情況，分析一下條件等待佇列。正常情況下，假設有2個執行緒`thread-1`和`thread-2`進入條件等待佇列，都處於阻塞狀態。
先是`thread-1`進入條件佇列：
然後是`thread-2`進入條件佇列：
條件等待佇列看起來也並不複雜，但是它並不是單獨存在和使用的，一般依賴於同步等待佇列，下面的一節分析`Condition`的實現的時候再詳細分析。
## 獨佔模式與共享模式
前文提及到，同步器涉及到獨佔模型和共享模式。下面就針對這兩種模式詳細分析一下`AQS`的具體實現原始碼。
### 獨佔模式
`AQS`同步器如果使用獨佔（`EXCLUSIVE`）模式，那麼意味著同一個時刻，只有唯一的一個節點所線上程獲取（`acuqire`）原子狀態`status`成功，此時該執行緒可以從阻塞狀態解除繼續執行，而同步等待佇列中的其他節點持有的執行緒依然處於阻塞狀態。獨佔模式同步器的功能主要由下面的四個方法提供：
- `acquire(int arg)`：申請獲取`arg`個原子狀態`status`（申請成功可以簡單理解為`status = status - arg`）。
- `acquireInterruptibly(int arg)`：申請獲取`arg`個原子狀態`status`，響應執行緒中斷。
- `tryAcquireNanos(int arg, long nanosTimeout)`：申請獲取`arg`個原子狀態`status`，帶超時的版本。
- `release(int arg)`：釋放`arg`個原子狀態`status`（釋放成功可以簡單理解為`status = status + arg`）。
獨佔模式下，`AQS`同步器例項初始化時候傳入的`status`值，可以簡單理解為"允許申請的資源數量的上限值"，下面的`acquire`型別的方法暫時稱為"獲取資源"，而`release`方法暫時稱為"釋放資源"。接著我們分析前面提到的四個方法的原始碼，先看`acquire(int arg)`：
```java
public final void acquire(int arg) {
// 獲取資源成功或者新增一個獨佔型別節點到同步等待佇列成功則直接返回，否則中斷當前執行緒
if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
// 此方法必須又子類覆蓋，用於決定是否獲取資源成功
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}
// 中斷當前執行緒
static void selfInterrupt() {
Thread.currentThread().interrupt();
}
// 不可中斷的獨佔模式下，同步等待佇列中的執行緒獲取資源的方法
final boolean acquireQueued(final Node node, int arg) {
boolean interrupted = false;
try {
for (;;) {
// 獲取新入隊節點的前驅節點
final Node p = node.predecessor();
// 前驅節點為頭節點並且嘗試獲取資源成功，也就是每一輪迴圈都會呼叫tryAcquire嘗試獲取資源，除非阻塞或者跳出迴圈
if (p == head && tryAcquire(arg)) {
// 設定新入隊節點為頭節點，原來的節點會從佇列中斷開
setHead(node);
p.next = null; // help GC
return interrupted; // <== 注意，這個位置是跳出死迴圈的唯一位置
}
// 判斷是否需要阻塞當前獲取資源失敗的節點中持有的執行緒
if (shouldParkAfterFailedAcquire(p, node))
// 阻塞當前執行緒，如果被喚醒則返回並清空執行緒的中斷標記
interrupted |= parkAndCheckInterrupt();
}
} catch (Throwable t) {
cancelAcquire(node);
if (interrupted)
selfInterrupt();
throw t;
}
}
/**
* 檢查並且更新獲取資源失敗的節點的狀態，返回值決定執行緒是否需要被阻塞。
* 這個方法是所有迴圈獲取資源方法中訊號控制的主要方法
*/
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
// 這裡記住ws是當前處理節點的前驅節點的等待狀態
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
// 前驅節點狀態設定成Node.SIGNAL成功，等待被release呼叫釋放，後繼節點可以安全地進入阻塞狀態
return true;
if (ws >