難以理解的AQS（下）
在上一篇部落格，簡單的說下了AQS的基本概念，核心原始碼解析，但是還有一部分內容沒有涉及到，就是AQS對條件變數的支援，這篇部落格將著重介紹這方面的內容。
條件變數
基本應用
我們先通過模擬一個消費者/生產者模型來看下條件變數的基本應用：
- 當有資料的時候，生產者停止生產資料，通知消費者消費資料；
- 當沒有資料的時候，消費者停止消費資料，通知生產者生產資料；
public class CommonResource {
private boolean isHaveData = false;
Lock lock = new ReentrantLock();
Condition producer_con = lock.newCondition();
Condition consumer_con = lock.newCondition();
public void product() {
lock.lock();
try {
while (isHaveData) {
try {
System.out.println("還有資料，等待消費資料");
producer_con.await();
} catch (InterruptedException e) {
}
}
System.out.println("生產者生產資料了");
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
isHaveData = true;
consumer_con.signal();
} finally {
lock.unlock();
}
}
public void consume() {
lock.lock();
try {
while (!isHaveData) {
try {
System.out.println("沒有資料了，等待生產者消費資料");
consumer_con.await();
} catch (InterruptedException e) {
}
}
System.out.println("消費者消費資料");
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
isHaveData = false;
producer_con.signal();
} finally {
lock.unlock();
}
}
}
複製程式碼
public class Main {
public static void main(String[] args) {
CommonResource resource = new CommonResource();
new Thread(() -> {
while (true) {
resource.product();
}
}).start();
new Thread(() -> {
while (true) {
resource.consume();
}
}).start();
}
}
複製程式碼
執行結果：
這就是條件變數的應用，第一反應是不是和object中的wait/nofity很像，wait/nofity是配合synchronized工作的，而條件變數的await/signal是配合使用AQS實現的鎖 來完成工作的，當然也要看用AQS實現的鎖是否支援了條件變數。synchronized只能與一個共享變數進行工作，而AQS實現的鎖支援多個條件變數。
我們試著分析下上面的程式碼：
首先建立了兩個條件變數，一個條件變數用來阻塞/喚醒消費者執行緒，一個條件變數用來阻塞/喚醒生產者執行緒。
生產者，首先獲取了獨佔鎖，判斷是否有資料：
- 如果有資料，則呼叫條件變數producer_con的await方法，阻塞當前執行緒，當消費者執行緒再次呼叫該條件變數producer_con的signal方法，就會喚醒該執行緒。
- 如果沒有資料，則生產資料，並且呼叫條件變數consumer_con的signal方法，喚醒因為呼叫consumer_con的await方法而被阻塞的消費者執行緒。
最終釋放鎖。
消費者，首先獲取了獨佔鎖，判斷是否有資料：
- 如果沒有資料，則呼叫條件變數consumer_con的await方法，阻塞當前執行緒，當生產者執行緒再次呼叫該條件變數consumer_con的signal方法，就會喚醒該執行緒。
- 如果有資料，則消費資料，並且呼叫條件變數producer_con的signal方法，喚醒因為呼叫producer_con的await方法而被阻塞的生產者執行緒。
最終釋放鎖。
這裡有兩點需要特別注意：
- 釋放鎖，一般應該放在finally裡面，以防中間出現異常，鎖沒有被釋放。
- 判斷是否有資料，其實用if在絕大部分情況也可以，但是用while更好一些，可以防止虛假喚醒。
為了加深對條件變數的理解，我們再來看一個例子，兩個執行緒交替列印奇偶數：
public class Test {
private int num = 0;
private Lock lock = new ReentrantLock();
Condition condition = lock.newCondition();
public void add() {
while(num<100) {
try {
lock.lock();
System.out.println(Thread.currentThread().getName() + ":" + num++);
condition.signal();
condition.await();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
}
複製程式碼
public class Main {
public static void main(String[] args) {
Test test=new Test();
new Thread(() -> {
test.add();
}).start();
new Thread(() -> {
test.add();
}).start();
}
}
複製程式碼
執行結果：
翻閱網上的大多數案例是分兩個執行緒方法交替列印，同時開兩個條件變數，其中一個條件變數負責阻塞/喚醒列印奇數的執行緒，一個變數負責阻塞/喚醒列印偶數的執行緒，但是個人覺得沒什麼必要，兩個執行緒共用一個執行緒方法，共用一個條件變數也可以。不知道各位看官是什麼想的？
原始碼解析
當我們點開lock.newCondition，發現它有好幾個實現類：
我們選擇ReentrantLock的實現類，實際上其他實現類也是相同的，只是為了和上面案例中的對應起來，所以先選擇ReentrantLock的實現類：
public Condition newCondition() {
return sync.newCondition();
}
複製程式碼
繼續往下點：
final ConditionObject newCondition() {
return new ConditionObject();
}
複製程式碼
可以看到，當我們呼叫lock.newnewCondition，最終會new出一個ConditionObject物件，而ConditionObject類是AbstractQueuedSynchronizer的內部類，我們先看下ConditionObject的UML圖：
其中firstWaiter儲存的是該條件變數下條件佇列的首節點，lastWaiter儲存的是該條件變數下條件佇列的尾節點。這裡只儲存了條件佇列的首節點和尾節點，中間的節點儲存在哪裡呢？ 讓我們點開await方法:
public final void await() throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
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);
}
複製程式碼
在這裡，我們就搞清楚三個問題即可：
- 完整的條件佇列儲存在哪裡，以什麼方式儲存？
- await方法，是如何釋放鎖的？
- await方法，是如何阻塞執行緒的？
第一個問題在addConditionWaiter方法可以得到答案：
private Node addConditionWaiter() {
Node t = lastWaiter;
// If lastWaiter is cancelled, clean out.
if (t != null && t.waitStatus != Node.CONDITION) {
unlinkCancelledWaiters();
t = lastWaiter;
}
Node node = new Node(Thread.currentThread(), Node.CONDITION);
if (t == null)
firstWaiter = node;
else
t.nextWaiter = node;
lastWaiter = node;
return node;
}
複製程式碼
首先是判斷條件佇列中的尾節點是否被取消了，如果被取消了，執行unlinkCancelledWaiters方法。我們這裡肯定沒有被取消，事實上，如果是第一次呼叫await方法，lastWaiter是為空的，所以肯定不會進入第一個if。隨後，新建一個Node，這個Node類就是上一篇部落格中大量介紹過的，也是AbstractQueuedSynchronizer的內部類，也就是新建了一個Node節點，其中儲存了當前執行緒和Node的型別，這裡Node的型別是CONDITION，如果t==null，則說明新建的Node是第一個節點，所以賦值給firstWaiter ，否則將尾節點的nextWaiter設定為新Node，形成一個單向連結串列，這個nextWaiter在哪裡呢，它是通過node點出來的，也就是它也屬於node類的一個欄位：
這說明了一個比較重要的問題： AQS的阻塞佇列是以雙向的連結串列的形式儲存的，是通過prev和next建立起關係的，但是AQS中的條件佇列是以單向連結串列的形式儲存的，是通過nextWaiter建立起關係的，也就是AQS的阻塞佇列和AQS中的條件佇列並非同一個佇列。
其實，AQS中的條件佇列也是一個阻塞佇列，只是為了方便，所以在本篇部落格中出現的AQS中的條件佇列特指在被條件變數await的，而阻塞佇列特指FIFO雙向連結串列佇列。
第一個問題解決了，我們再來看第二個問題，第二個問題答案在await的第二個方法：
final int fullyRelease(Node node) {
boolean failed = true;
try {
int savedState = getState();
if (release(savedState)) {
failed = false;
return savedState;
} else {
throw new IllegalMonitorStateException();
}
} finally {
if (failed)
node.waitStatus = Node.CANCELLED;
}
}
複製程式碼
首先呼叫getState方法，這個state是什麼，不知大家是否還有印象，對於ReentrantLock來說，state就是重入次數，隨後呼叫release方法，傳入state。也就是不管重入了多少次，這裡是一次性把鎖完全釋放掉。
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方法，這個方法正是需要被重寫的。
當完成了前兩個方法的呼叫後，就會進行一個判斷isOnSyncQueue，一般來說會進入這個if，park這個執行緒，等待喚醒，這就解決了第三個問題。
下面我們再來看看signal方法，同樣的，我們需要解決幾個問題：
- AQS的條件佇列和阻塞佇列既然不是同一個佇列，那麼是不是被await的執行緒永遠不會進入阻塞佇列？
- signal方法是如何喚醒執行緒的？
public final void signal() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node first = firstWaiter;
if (first != null)
doSignal(first);
}
複製程式碼
重點在於doSignal中的transferForSignal方法：
final boolean transferForSignal(Node node) {
/*
* If cannot change waitStatus, the node has been cancelled.
*/
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;
/*
* Splice onto queue and try to set waitStatus of predecessor to
* indicate that thread is (probably) waiting. If cancelled or
* attempt to set waitStatus fails, wake up to resync (in which
* case the waitStatus can be transiently and harmlessly wrong).
*/
Node p = enq(node);
int ws = p.waitStatus;
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
return true;
複製程式碼
在這個方法中，我們可以會呼叫enq方法，把條件佇列的執行緒放入阻塞佇列中，然後呼叫unpark方法，喚醒執行緒。
本篇部落格到這裡也結束了。
經過上下兩篇部落格，相信大家對AQS一定有了一個比較淺顯的理解。聰明的你，可以看出來，其實這兩篇部落格有很多內容都沒有講透，甚至有點模稜兩可，只是“蜻蜓點水”，所以這也符合了我的標題：難以理解的AQS。的確，AQS要深入研究的話，不比執行緒池簡單多少。看，我又再給自己找理由了。希望經過今後的沉澱，我可以把這兩篇部落格重寫下，然後換個標題“徹底理解AQS”，嘿嘿。