我們前面幾張提到過，JUC 這個包里面的工具類的底層就是使用 CAS 和 volatile 來保證線程安全的，整個 JUC 包里面的類都是基於它們構建的。今天我們介紹一個非常重要的同步器，這個類是 JDK 在 CAS 和 volatile 的基礎上為我們提供的一個同步工具類。
背景
AbstractQueuedSynchronizer，JDK 1.5 引入了 JUC 包，這個包提供了一些列支持並發的組件，這些組件是一些列同步器，他們主要完成以下功能：
- 內部狀態的管理和更新，比如表示一個鎖的狀態是獲取還是釋放。
- 線程同步狀態阻塞。
- 線程同步狀態釋放。
AQS 是一個小框架，基於這個框架我們可以實現很多的同步器，ReentrantLock，CountDownLatch，Semaphore 等都是基於 AQS 實現的。
功能
- 獨占鎖：每次只有一個線程能夠持有鎖，比如前面給大家演示的 ReentrantLock 就是以獨占方式實現的互斥鎖。
- 共享鎖：允許多個線程同時獲取鎖，並發訪問共享資源，比如 ReentrantReadWriteLock。
設計思想
同步器的核心方法是 acquire 和 release 操作。
acquire
while(當前同步器的狀態不允許獲取操作){
如果當前線程不再隊列中，將其加入隊列
阻塞當前線程
}
線程如果位於隊列中，將其移出隊列
release
更新同步器的狀態
if(新的狀態允許某個被阻塞的線程獲取成功)
解除隊列中一個或多個線程的阻塞狀態。
從上面的操作思想中我們可以提出三大關鍵操作：同步器狀態變更，線程阻塞和釋放，插入和移出隊列。由此可以引申出三個基本組件：
- 同步器狀態的原子性管理
- 線程阻塞與解除阻塞
- 隊列的管理
同步狀態
AQS 類使用 int 值來保存同步狀態，並且暴露出 getState，setState 和 compareAndSet 操作來讀取和更新這個同步狀態。線程通過修改（加/減指定的數量）碼是否成功來決定當前線程是否成功獲取到同步狀態。
State 被聲明成了 volatile，保證了可見性和有序性。又通過 CAS 指令來實現 compareAndSet ，使得當且僅當同步狀態擁有一個一致的期望值的時候，才會被原子地設置成新值，這樣就保證了同步狀態的原子性。
阻塞
直到 JSR166，阻塞線程和解除線程阻塞都是基於 Java 的內置管程。
JUC 包使用 LockSupport 類來解決這個問題。LockSupport.park 阻塞當前線程直到有 LockSupport.unpark 方法被調用。
隊列
整個框架的核心就是如何管理線程阻塞隊列，該隊列是嚴格的 FIFO 隊列，因此不支持線程優先級的同步。同步隊列的最佳選擇是自身沒有使用底層鎖來構造的非阻塞數據結構。這里采用了 CLH 鎖。
CLH隊列實際並不那么像隊列，它的入隊和出隊與實際的業務密切相關。它是一個鏈表隊列。用過 AQS 的兩個字段 head(頭節點) 和 tail(尾節點)來存取，這兩個字段初始化的時候都指向了一個空節點。
入隊操作：
CLH 隊列是 FIFO 隊列，所以新的節點來到的時候，是要插入到當前隊列的尾節點之后。當一個線程獲取到同步狀態之后，其他線程無法獲取，轉而被構造成節點加入到同步隊列中，而且這個加入隊列的過程必須要保證線程安全，因此使用了 CAS方法，它需要傳遞當前線程認為的尾節點和當前節點，只有設置成功后，當前節點才正式與之前的尾節點建立關聯。
出隊操作：
因為是 FIFO 隊列，所以能成功獲取到 AQS 同步狀態的必定是首節點，首節點的線程在釋放同步狀態時，會喚醒后續節點，而后續節點會在獲取 AQS 同步狀態成功的時候將自己設置為首屆點。設置首節點是由獲取同步成功的線程來完成的，所以不需要像入隊這樣的 CAS 操作。
條件隊列
上一節是 AQS 的同步隊列，這一節是條件隊列。AQS 只有一個同步隊列，但是可以有多個條件隊列。AQS 框架提供了一個 ConditionObject 類，給維護獨占同步的類以及實現 Lock 接口的類使用。
ConditionObject 類 和 AQS 共用了內部節點，有自己單獨的條件隊列。Singal 操作是通過將節點從條件隊列轉移到同步隊列來實現的。
singal：
await：
方法結構
| 組件 | 數據結構 |
|---|---|
| 同步狀態 | volatile int state |
| 阻塞 | LockSupport類 |
| 隊列 | Node節點 |
| 條件隊列 | ConditionObject |
源代碼
我們通過獨占式同步狀態的釋放和獲取，以及共享式同步狀態的釋放和獲取來看看 AQS 是如何實現的。
獨占式
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
上述代碼主要完成了同步狀態的獲取，節點構造，加入同步隊列以及在同步隊列中自旋等待等相關工作。
- 調用子類實現的 tryAcquire 方法，該方法保證線程安全同時獲取同步狀態。
- 獲取同步狀態失敗，則構造獨占式同步節點。
- 通過 addWriter 將該節點加入到同步隊列的尾部。
- 最后通過 acquireQueued 方法，使得該節點以自選的方式獲取同步狀態。
來看看節點構造和加入隊列的實現：
private Node addWaiter(Node mode) {
// 當前線程構造成Node節點
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
// 嘗試快速在尾節點后新增節點 提升算法效率 先將尾節點指向pred
Node pred = tail;
if (pred != null) {
//尾節點不為空 當前線程節點的前驅節點指向尾節點
node.prev = pred;
//並發處理 尾節點有可能已經不是之前的節點 所以需要CAS更新
if (compareAndSetTail(pred, node)) {
//CAS更新成功 當前線程為尾節點 原先尾節點的后續節點就是當前節點
pred.next = node;
return node;
}
}
//第一個入隊的節點或者是尾節點后續節點新增失敗時進入enq
enq(node);
return node;
}
private Node enq(final Node node) {
for (;;) {
Node t = tail;
if (t == null) { // Must initialize
//尾節點為空 第一次入隊 設置頭尾節點一致 同步隊列的初始化
if (compareAndSetHead(new Node()))
tail = head;
} else {
//所有的線程節點在構造完成第一個節點后 依次加入到同步隊列中
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
節點進入同步隊列后，就進入了一個自旋的過程，每個線程節點都在自旋地觀察，當條件滿足，獲取到了同步狀態，就可以從自旋過程中退出，否則依舊自旋。
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);
}
}
shouldParkAfterFailedAcquire 和 parkAndCheckInterrupt 阻塞線程的過程。
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
//前驅節點的狀態決定后續節點的行為
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
/*前驅節點為-1 后續節點可以被阻塞
* This node has already set status asking a release
* to signal it, so it can safely park.
*/
return true;
if (ws > 0) {
/*
* Predecessor was cancelled. Skip over predecessors and
* indicate retry.
*/
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
/*前驅節點是初始或者共享狀態就設置為-1 使后續節點阻塞
* waitStatus must be 0 or PROPAGATE. Indicate that we
* need a signal, but don't park yet. Caller will need to
* retry to make sure it cannot acquire before parking.
*/
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
private final boolean parkAndCheckInterrupt() {
//阻塞線程
LockSupport.park(this);
return Thread.interrupted();
}
當獲取同步狀態成功之后，對於鎖這種並發組件而言，就意味着當前線程獲取到了鎖。
再看 release 方法：
head節點表示獲取鎖成功的節點，當頭結點在釋放同步狀態時，會喚醒后繼節點，如果后繼節點獲得鎖成功，會把自己設置為頭結點，節點的變化過程如下。修改head節點指向下一個獲得鎖的節點，新的獲得鎖的節點，將prev的指針指向null。
public final boolean release(int arg) {
if (tryRelease(arg)) {//同步狀態釋放成功
Node h = head;
if (h != null && h.waitStatus != 0)
//直接釋放頭節點
unparkSuccessor(h);
return true;
}
return false;
}
private void unparkSuccessor(Node node) {
/*
* If status is negative (i.e., possibly needing signal) try
* to clear in anticipation of signalling. It is OK if this
* fails or if status is changed by waiting thread.
*/
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
/*尋找符合條件的后續節點
* Thread to unpark is held in successor, which is normally
* just the next node. But if cancelled or apparently null,
* traverse backwards from tail to find the actual
* non-cancelled successor.
*/
Node s = node.next;
if (s == null || s.waitStatus > 0) {
s = null;
for (Node t = tail; t != null && t != node; t = t.prev)
if (t.waitStatus <= 0)
s = t;
}
if (s != null)
//喚醒后續節點
LockSupport.unpark(s.thread);
}
總結：在獲取同步狀態時，同步器維護一個同步隊列，獲取狀態失敗的線程都會被加入到隊列中進行自旋。移除的條件是前驅節點是頭節點並且成功獲取了同步狀態。釋放時，會喚醒頭節點的后繼節點。
應用
ReentrantLock：ReentrantLock 類使用 AQS 同步狀態來保存鎖重復持有的次數。當鎖被一個線程獲取時，ReentrantLock 也會記錄下當前獲得鎖的線程表示，以便檢查是否重復獲取。
ReentrantReadWriteLock：ReentrantReadWriteLock 使用 AQS 同步狀態中的 16 為來保存寫鎖的持有次數，剩下的 16 為來保存讀鎖的持有次數。WriteLock 的構建方式和 ReentrantLock 一樣。ReadLock 則通過使用 acquireShared 方法來支持同時允許多個讀線程。
Semaphore：信號量使用 AQS 同步狀態來保存信號量當前計數。它里面定義的 acquireShared 方法會減少計數，當計數為非正值時阻塞線程。tryRelease 會增加技術，在計數為正值時還要解除線程的阻塞。
CountDownLatch：使用 AQS 同步狀態來表示計數。當該計數為 0 時，所有的 acquire 方法才能通過。
FutureTask：使用 AQS 的同步狀態來表示某個異步計算任務的運行狀態（初始化，運行中，被取消和完成）。設置（FutureTask 的 set 方法）或取消（FutureTask 的 cancel 方法）一個 FutureTask 時會調用 AQS 的 release 操作。等待計算結果的線程阻塞解除是通過 AQS 的 acquire 實現的。
SynchronousQueues：SynchronousQueues類使用了內部的等待節點，這些節點可以用於協調生產者和消費者。同時，它使用AQS同步狀態來控制當某個消費者消費當前一項時，允許一個生產者繼續生產，反之亦然。
流程圖
-
多線程並發修改同步狀態，修改成功的線程標記為擁有同步狀態。
-
獲取失敗的線程，加入到同步隊列的隊尾；加入到隊列中后，如果當前節點的前驅節點為頭節點再次嘗試獲取同步狀態（下文代碼：p == head && tryAcquire(arg)）。
-
如果頭節點的下一個節點嘗試獲取同步狀態失敗后，會進入等待狀態；其他節點則繼續自旋。
-
當線程執行完相應邏輯后，需要釋放同步狀態，使后繼節點有機會同步狀態（讓出資源，讓排隊的線程使用）。這時就需要調用release(int arg)方法。調用該方法后，會喚醒后繼節點。
-
后繼節點獲取同步狀態成功，頭節點出隊。需要注意的事，出隊操作是間接的，有節點獲取到同步狀態時，會將當前節點設置為head，而原本的head設置為null。
-
當同步隊列中頭節點喚醒后繼節點時，此時可能有其他線程嘗試獲取同步狀態。
-
假設獲取成功，將會被設置為頭節點。
-
頭節點后續節點獲取同步狀態失敗。
-
共享模式和獨占模式最主要的區別是在支持同一時刻有多個線程同時獲取同步狀態。為了避免帶來額外的負擔，在上文中提到的同步隊列中都是用獨占模式進行講述，其實同步隊列中的節點應該是獨占和共享節點並存的。
-
共享節點嘗試獲取同步狀態。
-
當一個同享節點獲取到同步狀態，並喚醒后面等待的共享狀態的結果如下圖所示：
-
最后，獲取到同步狀態的線程執行完畢，同步隊列中只有一個獨占節點：
總結
- AQS通過一個int同步狀態碼，和一個（先進先出）隊列來控制多個線程訪問資源
- 支持獨占和共享兩種模式獲取同步狀態碼
- 當線程獲取同步狀態失敗會被加入到同步隊列中
- 當線程釋放同步狀態，會喚醒后繼節點來獲取同步狀態
- 共享模式下的節點獲取到同步狀態或者釋放同步狀態時，不僅會喚醒后繼節點，還會向后傳播，喚醒所有同步節點
- 使用volatile關鍵字保證狀態碼在線程間的可見性，CAS操作保證修改狀態碼過程的原子性。