難以理解的AQS（上）
在一篇部落格中，我們看了下CopyOnWriteArrayList的原始碼，不是很難，裡面用到了一個可重入的排他鎖： ReentrantLock，這東西看上去和Synchronized差不多，但是和Synchronized是完全不同的東西。
Synchronized鎖的特性是JVM保證的，ReentrantLock鎖的特性是上層的Java程式碼控制的。而ReentrantLock的基礎就是AQS，事實上，很多併發容器都用了ReentrantLock，也就間接的用到了AQS，還有併發框架，如CountDownLatch，CyclicBarrier，Semaphore也都用到了AQS，可見AQS的重要性。
但是要想稍微深入一點理解AQS實屬不易，牽扯到不少東西，所以本篇部落格將會拆分成兩篇，第一篇將會介紹AQS的前置知識：LockSupport，AQS的核心概念，以及獨佔、共享模式下，AQS的核心原始碼解析等，第二篇將會介紹AQS對條件變數的支援，以及AQS的應用等。
要深入一些學習AQS，首先要掌握一個前置知識：LockSupport。
LockSupport
LockSupport是一個工具類，它的主要作用是掛起和喚醒執行緒，它的底層是呼叫的native方法，這個我們不去深入探究，主要看下LockSupport的應用。
park，unpark
如果呼叫park方法的執行緒已經拿到了與LockSupport關聯的許可證，呼叫park後，會立即返回，否則該執行緒會被阻塞，直到拿到了許可證。
如果一個執行緒呼叫了unpark方法，就會獲得與LockSupport關聯的許可證，如果該執行緒之前呼叫了park而被阻塞，那麼會被喚醒，如果該執行緒之前沒有呼叫park方法，那麼呼叫park方法後，會立刻返回。
public static void main(String[] args) {
System.out.println("Hello，LockSupport");
LockSupport.park();
System.out.println("Bye，LockSupport");
}
執行結果：
image.png
執行緒打印出第一句話，就被阻塞了，因為執行緒沒有獲得與LockSupport關聯的許可證。
public static void main(String[] args) {
System.out.println("Hello，LockSupport");
LockSupport.unpark(Thread.currentThread());
LockSupport.park();
System.out.println("Bye，LockSupport");
}
執行結果：
image.png
先呼叫unpark方法，傳入了當前執行緒，當前執行緒獲得了與LockSupport關聯的許可證，隨後呼叫park方法，因為該執行緒已經有了許可證，所以立即返回，打印出了第二句話。
public static void main(String[] args) {
Thread thread=new Thread(()->{
System.out.println("Hello，LockSupport");
LockSupport.park();
System.out.println("Bye，LockSupport");
});
thread.start();
LockSupport.unpark(thread);
}
執行結果：
image.png
首先建立了一個Thread ，內部呼叫了park方法，隨之啟動執行緒，在主執行緒中，呼叫了unpark方法，傳入了子執行緒。
此方法有兩種情況：
- 主執行緒先呼叫了unpark方法，子執行緒中的park方法隨後呼叫。
- 子執行緒的park方法先呼叫，主執行緒的unpark方法隨後呼叫。
但是不管是哪種情況，最後的結果都是一樣的。只是過程有些區別，第一種情況是
主執行緒呼叫了unpark方法後，讓子執行緒拿到了許可證，子執行緒內部呼叫park後立即返回，第二種情況是子執行緒的park方法先呼叫到，因為目前還沒有拿到許可證，所以被阻塞，隨後主執行緒呼叫了unpark，讓子執行緒拿到了許可證，子執行緒被返回。
parkNanos(long nanos)
和park方法類似，不同之處在於多了個超時時間，如果呼叫parkNanos，執行緒被阻塞了，超過了nanos後，不管有沒有獲得許可，都會被返回。
public static void main(String[] args) {
System.out.println("Hello，LockSupport");
LockSupport.parkNanos(Integer.MAX_VALUE);
System.out.println("Bye，LockSupport");
}
執行結果：
image.png
為了可以看到比較明顯的效果，所以我把時間設定成了Integer.MAX_VALUE，可以看到雖然沒有呼叫unpark方法拿到許可證，但是一定的時間後，該方法還是被返回了。
park(Object blocker)
此方法是比較推薦使用的，因為使用它，可以通過jstack命令檢視有關阻塞物件的資訊。
public class Main {
public void test() {
LockSupport.park(this);
}
public static void main(String[] args) {
Main main = new Main();
main.test();
}
}
使用jstack pid命令：
image.png
還有幾個方法，就不一一介紹了。
有了上面的基礎，我們終於可以進入今天的正題了：AQS。
什麼是AQS
AQS的全稱是AbstractQueuedSynchronizer，翻譯是中文是抽象同步佇列。剛接觸AQS的時候，第一感覺這個東西和抽象有關係，因為Abstract。。。後來發現，這個東西和抽象沒有半毛錢關係，慢慢的，又有新的理解，這個東西和抽象還真的有點關係，因為它把實現同步佇列的一些方法給抽象出來了，供其他上層元件重寫或者複用。重點來了，其他上層元件需要重寫其中的方法！再說的詳細點，就是其他元件需要繼承AbstractQueuedSynchronizer，對其中的部分方法進行重寫。
我們先來看下AQS的UML圖：
image.png
AQS核心概念
我們先要對AQS進行一個大概的介紹，瞭解下AQS中比較核心的東西。
AQS維護了一個FIFO的雙向佇列，什麼是FIFO？就是先進先出的意思，雙向佇列就是上一個節點指向下一個節點的同時，下一個節點也指向上一個節點，我們從AbstractQueuedSynchronizer關聯的Node類中就可以看出來這一點：prev儲存的是當前節點上一個node，next儲存的是當前節點的下一個節點，有一個專業的名詞，分別是前驅節點，後繼節點，同時AbstractQueuedSynchronizer類有兩個欄位，一個是head，一個是tail，顧名思義，head儲存了頭節點，tail儲存了尾節點。
Node類中的SHARED是用來標記該執行緒是獲取共享資源時被放入等待佇列的，EXCLUSIVE用來標記該執行緒是獲取獨佔資源時被放入等待佇列的，從這句話，我們可以看出Node類其實就是儲存了放入等待佇列的執行緒，而有的執行緒是因為獲取共享資源失敗放入等待佇列的，而有的執行緒是因為獲取獨佔資源失敗而被放入等待佇列的，所以這裡需要有一個標記去區分。
再囉嗦一句，FIFO雙向佇列其實就是AQS中的等待佇列。
在Node類中，還有一個欄位：waitStatus，它有五個取值，分別是：
- SIGNAL：值為-1，當前節點在入隊後、進入休眠狀態前，應確保將其prev節點型別改為SIGNAL，以便後者取消或釋放時將當前節點喚醒。也就是說當前節點的waitStatus為SIGNAL的時候，被釋放的時候，才會喚醒後繼節點。其實，如果要較真的話，這種理解也有點些問題，就先這麼理解吧。
- CANCELLED：值為1，被取消的，在等待佇列中等待的執行緒超時或被中斷，進入該狀態的節點的將不再變化。
- CONDITION：值為-2，該節點處於條件佇列中，當其他執行緒呼叫了Condition的signal()方法後，節點轉移到AQS的等待佇列中，特別要注意的是，條件佇列和AQS的等待佇列並不是一回事。
- PROPAGATE：值為-3。對於這個狀態到底是做什麼的，網上大多數部落格，包括書籍都是簡單提了下這是共享模式下專用的，和傳播有關，但是沒有更深的解釋。無奈，菜的摳腳的我至今也沒能領悟這個狀態值的含義。
- 0：預設值。
在AbstractQueuedSynchronizer類中，有一個state欄位，被標記為volatile，是為了保證可見性，這個欄位的設計可厲害了。對於ReentrantLock來說，state儲存的是重入次數，對於ReentrantReadWriteLock來說，state儲存的是獲取讀鎖的重入次數和寫鎖的重入次數。
AbstractQueuedSynchronizer類中，還有一個內部類：ConditionObject，用來提供條件變數的支援。
AQS提供了兩種方式來獲取資源，一種是獨佔模式，一種是共享模式。
上面提到過，需要去定義一個類去繼承AbstractQueuedSynchronizer類，重寫其中的方法，一般來說
- 對於獨佔模式，需要重寫tryAcquire(arg) ，tryRelease(int arg)方法。
- 對於共享模式，需要重寫tryAcquireShared(arg) ，tryReleaseShared(int arg)方法。
原始碼解析
獨佔模式
acquire
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
此方法是獨佔模式下獲取資源的頂級方法，如果執行緒呼叫tryAcquire(arg)方法成功了，說明已經獲取到了資源，直接返回，如果不成功，則將當前執行緒封裝成waitStaus為Node.EXCLUSIVE的Node插入到AQS等待佇列的尾部。
我們來看下tryAcquire方法：
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}
納尼，直接報錯了，這是什麼鬼？別忘了，我們需要重寫這個方法。
我們再來看下addWaiter(Node.EXCLUSIVE), arg)方法：
private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode);//封裝成Node，新的Node
// Try the fast path of enq; backup to full enq on failure
Node pred = tail;//把尾節點賦值給pred ，pred也就是尾節點了
if (pred != null) {//如果pred不為NULL
node.prev = pred;//pred賦值給新節點的前驅節點，也就是新節點的前驅節點是尾節點
if (compareAndSetTail(pred, node)) {//CAS，如果pred還是尾節點，則把新節點設定成尾節點，設定成功後，進入if
pred.next = node;//把新節點賦值給pred的後繼節點
return node;//返回新節點
}
}
enq(node);
return node;
}
此方法先把執行緒封裝成一個(Node.EXCLUSIVE的Node，先嚐試把這個Node直接放入隊尾，如果成功的話，直接返回，如果失敗的話，呼叫enq(node)進行入隊操作：
private Node enq(final Node node) {
for (;;) {//自旋
Node t = tail;//把尾節點賦值給t
//如果尾節點為空，則新建一個空的Node，用CAS把空的Node設定成頭節點
//成功後，再把尾部節點也指向空的Node
if (t == null) { // Must initialize
if (compareAndSetHead(new Node()))
tail = head;
} else {
node.prev = t;//把尾節點賦值給傳進來的node的前驅節點
if (compareAndSetTail(t, node)) {//CAS，如果t還是尾部節點，則用傳進來的node替換舊的尾部節點
t.next = node;//設定t的後繼節點為傳進來的node
return t;
}
}
}
}
這個方法概括的來說，就是把獲取資源失敗的node放入AQS等待佇列。
我們再回到頂級方法看下acquireQueued方法：
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();//拿到node的前驅節點，賦值給p
if (p == head && tryAcquire(arg)) {//如果p已經是頭節點了，代表這個時候
//node是第二個節點，再次呼叫tryAcquire獲取資源
setHead(node);//設定頭節點
p.next = null; // help GC
failed = false;
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) &&//判斷此node是否可以被park
parkAndCheckInterrupt())//park
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
又是CAS自旋，首先拿到node的前驅節點，賦值給p，如果p已經是頭節點了，代表這個時候node是第二個節點，再次嘗試呼叫tryAcquire獲取資源，如果成功，設定頭節點為node，返回中斷標記位，如果失敗，先判斷自己是否可以被park，如果可以的話，就park，等待unpark。
再來看下parkAndCheckInterrupt方法：
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;//拿到前驅節點的waitStatus，賦值給ws
if (ws == Node.SIGNAL)//如果是SIGNAL
return true;
if (ws > 0) {//如果是ws>0，則說明前驅節點被取消了，通過while迴圈，
//找到最近的一個沒有取消的節點，排到後面
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);//CAS設定前驅節點的waitStatus為SIGNAL
}
return false;
}
如果前驅節點的waitStatus為SIGNAL，直接返回，如果前驅節點被取消了，則通過while迴圈，找到最近的一個沒有被取消的節點，排到後面去，如果前驅節點處於其他狀態，則通過CAS把前驅節點的waitStatus設定為SIGNAL。
再來看下parkAndCheckInterrupt方法：
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
這方法就比較簡單了，就是park自己，返回當前執行緒是否被中斷。
我們來為acquireQueued方法做一個總結：
找到一個安全點park自己，如果被喚醒了，檢查自己是否是第二個節點，如果是的話，再次嘗試獲取資源，成功的話，就把自己設定為頭節點。
好了，整個頂級的acquire核心內容已經分析完畢了，我們來做一個總結：
- 嘗試快速獲取資源，如果成功，直接返回，失敗，進入下一步。
- 進行入隊操作。
- 在等待佇列中的執行緒獲取資源。
最後，畫個流程圖幫助理解整個流程：
image.png
release
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
此方法是獨佔模式下釋放資源的頂級方法。
首先呼叫tryRelease方法，如果成功了，把頭節點賦值給h，如果h不為null並且waitStatus 不等於0，呼叫unparkSuccessor方法，喚醒下一個node。
tryRelease：
protected boolean tryRelease(int arg) {
throw new UnsupportedOperationException();
}
此方法還是直接報錯，因為我們需要重寫。這裡我們需要尤其注意，此方法是判斷資源是否被完全釋放了，如果鎖是可以重入的，可能多次獲得了鎖，所以必須把最後一個鎖也釋放了，這裡才能返回ture，否則返回false。
unparkSuccessor：
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;//拿到當前節點的waitStatus，賦值給ws
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
Node s = node.next;//當前節點的下一個節點賦值給s
if (s == null || s.waitStatus > 0) {//如果s==null或者已經被取消了，就通過for迴圈找到下一個需要被喚醒的節點
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);//喚醒
}
此方法的核心就是喚醒下一個需要被喚醒的節點。
共享模式
acquireShared
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
該方法是共享模式下獲取資源的頂級方法。
首先呼叫tryAcquireShared，來嘗試獲取資源，成功的話，則呼叫doAcquireShared，進入等待佇列，直到獲取了資源。
tryAcquireShared：
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}
我們需要重寫tryAcquireShared方法。
doAcquireShared：
private void doAcquireShared(int arg) {
final Node node = addWaiter(Node.SHARED);//入隊
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();//拿到當前節點的前驅節點，賦值給p
if (p == head) {//如果p是頭節點
int r = tryAcquireShared(arg);//呼叫tryAcquireShared嘗試獲取資源
if (r >= 0) {
setHeadAndPropagate(node, r);//設定頭節點，如果還有剩餘資源，喚醒下一個節點
p.next = null; // help GC
if (interrupted)
selfInterrupt();
failed = false;
return;
}
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
此方法和獨佔模式下的流程區別不大，最大的不同在於setHeadAndPropagate方法，我們來看看這個方法做了什麼：
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head;
setHead(node);//設定頭節點
//如果還有剩餘資源
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next;//找到後繼節點
if (s == null || s.isShared())
doReleaseShared();//呼叫doReleaseShared方法
}
}
首先設定當前節點為頭節點，如果還有剩餘的資源，就找到後繼節點，呼叫doReleaseShared方法，這個方法我們後面再看，但是從方法名稱來看，我們可以知道它與釋放共享資源有關。
releaseShared
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
此方法是共享模式下釋放資源的頂級方法。
tryReleaseShared方法還是需要我們去重寫的，如果成功了，呼叫doReleaseShared方法：
private void doReleaseShared() {
for (;;) {
Node h = head;//把頭節點賦值給h
if (h != null && h != tail) {
int ws = h.waitStatus;//拿到h的waitStatus賦值給ws
if (ws == Node.SIGNAL) {//如果為SIGNAL
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue;
unparkSuccessor(h);//喚醒後繼節點
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue;
}
if (h == head)
break;
}
}
這個方法在共享模式下獲取共享資源的頂級方法acquireShared中的doAcquireShared中的setHeadAndPropagate也會呼叫。
好了，獨佔模式，共享模式下的獲取資源，釋放資源核心流程已經分析完畢了。
細心的你，一定發現在AQS中還有acquireInterruptibly()/acquireSharedInterruptibly()這兩個方法，這兩個方法從名稱上來看僅僅是多了一個Interruptibly，它們是會對中斷進行響應的，而我們上面介紹的acquire，acquireShared是忽略中斷的。
本篇部落格到這裡就結束了，但是還有一塊東西沒有講到：對條件變數的支援，這部分內容將放到下一篇部落格再詳細介紹。