死磕java concurrent包系列（三）基於ReentrantLock理解AQS的條件佇列
基於Codition分析AQS的條件佇列
前言
上一篇我們講了AQS中的同步佇列佇列，現在我們研究一下條件佇列。
在java中最常見的加鎖方式就是synchorinzed和Reentrantlock，我們都說Reentrantlock比synchorinzed更加靈活，其實就靈活在Reentrantlock中的條件佇列的用法上。
Condition介面
它是在java1.5中引入的一個介面，主要是為了替代object類中的wait、notify方法，以一種更靈活的方式解決執行緒之間的通訊問題：
public interface Condition {
/**
* 使當前執行緒進入等待狀態直到被通知(signal)或中斷
* 當其他執行緒呼叫singal()或singalAll()方法時，該執行緒將被喚醒
* 當其他執行緒呼叫interrupt()方法中斷當前執行緒
* await()相當於synchronized等待喚醒機制中的wait 最重要的是await方法使執行緒進入等待狀態，再通過signal方法喚醒。接下來我們結合實際例子分析。
Condition可以解決什麼問題
假設有一個生產者-消費者的場景：
1、生產者有兩個執行緒產生烤雞；消費者有兩個執行緒消費烤雞
2、四個執行緒一起執行，但同時只能有一個生產者執行緒生成烤雞，一個消費者執行緒消費烤雞。
3、只有產生了烤雞，才能通知消費執行緒去消費，否則只能等著；
4、只有消費了烤雞，才能通知生產者執行緒去生產，否則只能等著
於是乎，我們使用ReentrantLock控制併發，並使用它生成兩組Condition物件，productCondition和consumeCondition：前者控制生產者執行緒，後者控制消費者執行緒。當isHaveChicken為true時，代表烤雞生成完畢，生產執行緒必須進入等待狀態同時喚醒消費執行緒進行消費，消費執行緒消費完畢後將flag設定為false，代表烤雞消費完成，進入等待狀態，同時喚醒生產執行緒生產烤雞。。。。。。
package com.springsingleton.demo.Chicken;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
public class ChikenStore {
ReentrantLock reentrantLock = new ReentrantLock();
Condition productCondition = reentrantLock.newCondition();
Condition consumeCondition = reentrantLock.newCondition();
private int count = 0;
private volatile boolean isHaveChicken = false;
//生產
public void ProductChicken() {
reentrantLock.lock();
while (isHaveChicken) {
try {
System.out.println("有烤雞了" + Thread.currentThread().getName() + "不生產了");
productCondition.await();
} catch (Exception e) {
System.out.println("error" + e.getMessage());
}
}
count++;
System.out.println(Thread.currentThread().getName() + "產生了第" + count + "個烤雞，趕緊開始賣");
isHaveChicken = true;
consumeCondition.signal();
reentrantLock.unlock();
}
public void SellChicken() {
reentrantLock.lock();
while (!isHaveChicken) {
try {
System.out.println("沒有烤雞了" + Thread.currentThread().getName() + "不賣了");
consumeCondition.await();
} catch (Exception e) {
System.out.println("error" + e.getMessage());
}
}
count--;
isHaveChicken = false;
System.out.println(Thread.currentThread().getName() + "賣掉了第" + count + 1 + "個烤雞，趕緊開始生產");
productCondition.signal();
reentrantLock.unlock();
}
public static void main(String[] args) {
ChikenStore chikenStore = new ChikenStore();
new Thread(() -> {
Thread.currentThread().setName("生產者1號");
while (true) {
chikenStore.ProductChicken();
}
}).start();
new Thread(() -> {
Thread.currentThread().setName("生產者2號");
for (; ; ) {
chikenStore.ProductChicken();
}
}).start();
new Thread(() -> {
Thread.currentThread().setName("消費者1號");
while (true) {
chikenStore.SellChicken();
}
}).start();
new Thread(() -> {
Thread.currentThread().setName("消費者2號");
while (true) {
chikenStore.SellChicken();
}
}).start();
}
}
複製程式碼輸出：
生產者1號產生了第1個烤雞，趕緊開始賣
有烤雞了生產者1號不生產了
有烤雞了生產者2號不生產了
消費者1號賣掉了第01個烤雞，趕緊開始生產
沒有烤雞了消費者1號不賣了
生產者1號產生了第1個烤雞，趕緊開始賣
有烤雞了生產者1號不生產了
消費者1號賣掉了第01個烤雞，趕緊開始生產
沒有烤雞了消費者1號不賣了
沒有烤雞了消費者2號不賣了
生產者2號產生了第1個烤雞，趕緊開始賣
有烤雞了生產者2號不生產了
消費者1號賣掉了第01個烤雞，趕緊開始生產
沒有烤雞了消費者1號不賣了
生產者1號產生了第1個烤雞，趕緊開始賣
有烤雞了生產者1號不生產了
消費者2號賣掉了第01個烤雞，趕緊開始生產
沒有烤雞了消費者2號不賣了
複製程式碼如果用synchorinzed的話：
package com.springsingleton.demo.Chicken;
public class ChickenStoreSync {
private int count = 0;
private volatile boolean isHaveChicken = false;
public synchronized void ProductChicken() {
while (isHaveChicken) {
try {
System.out.println("有烤雞了" + Thread.currentThread().getName() + "不生產了");
this.wait();
} catch (Exception e) {
System.out.println("error" + e.getMessage());
}
}
count++;
System.out.println(Thread.currentThread().getName() + "產生了第" + count + "個烤雞，趕緊開始賣");
isHaveChicken = true;
notifyAll();
}
public synchronized void SellChicken() {
while (!isHaveChicken) {
try {
System.out.println("沒有烤雞了" + Thread.currentThread().getName() + "不賣了");
this.wait();
} catch (Exception e) {
System.out.println("error" + e.getMessage());
}
}
count--;
isHaveChicken = false;
System.out.println(Thread.currentThread().getName() + "賣掉了第" + count + 1 + "個烤雞，趕緊開始生產");
notifyAll();
}
public static void main(String[] args) {
ChickenStoreSync chikenStore = new ChickenStoreSync();
new Thread(() -> {
Thread.currentThread().setName("生產者1號");
while (true) {
chikenStore.ProductChicken();
}
}).start();
new Thread(() -> {
Thread.currentThread().setName("生產者2號");
for (; ; ) {
chikenStore.ProductChicken();
}
}).start();
new Thread(() -> {
Thread.currentThread().setName("消費者1號");
while (true) {
chikenStore.SellChicken();
}
}).start();
new Thread(() -> {
Thread.currentThread().setName("消費者2號");
while (true) {
chikenStore.SellChicken();
}
}).start();
}
}
複製程式碼如上程式碼，在呼叫notify()或者 notifyAll()方法時，由於synchronized等待佇列中同時存在生產者執行緒和消費者執行緒，所以我們並不能保證被喚醒的到底是消費者執行緒還是生產者執行緒，而Codition則可以避免這種情況。
AQS中Condition的實現原理
Condition的具體實現類是AQS的內部類ConditionObject，前面我們分析過AQS中存在兩種佇列，一種是同步佇列，一種是條件佇列，而條件佇列就相對於Condition而言的。注意在使用Condition前必須獲得鎖，同時在Condition的條件佇列上的結點與前面同步佇列的結點是同一個類即Node，其結點的waitStatus的值為CONDITION。在實現類ConditionObject中有兩個結點分別是firstWaiter和lastWaiter，firstWaiter代表等待佇列第一個等待結點，lastWaiter代表等待佇列最後一個等待結點
public class ConditionObject implements Condition, java.io.Serializable {
//等待佇列第一個等待結點
private transient Node firstWaiter;
//等待佇列最後一個等待結點
private transient Node lastWaiter;
//省略.......
}
複製程式碼每個Condition都對應著一個條件佇列，也就是說如果一個鎖上建立了多個Condition物件，那麼也就存在多個等待佇列。等待佇列是一個FIFO的佇列，在佇列中每一個節點都包含了一個執行緒的引用，而該執行緒就是Condition物件上等待的執行緒。當一個執行緒呼叫了await()相關的方法，那麼該執行緒將會釋放鎖，並構建一個Node節點封裝當前執行緒的相關資訊加入到條件佇列中進行等待，直到被喚醒、中斷、超時才從佇列中移出。Condition中的等待佇列模型如下
正如圖所示，Node節點的資料結構，在等待佇列中使用的變數與同步佇列是不同的，Condtion中等待佇列的結點只有直接指向的後繼結點並沒有指明前驅結點，而且使用的變數是nextWaiter而不是next，這點我們在前面分析結點Node的資料結構時講過。firstWaiter指向條件佇列的頭結點，lastWaiter指向條件佇列的尾結點，等待佇列中結點的狀態只有兩種即CANCELLED和CONDITION，前者表示執行緒已結束需要從等待佇列中移除，後者表示條件結點等待被喚醒。
再次強調每個Codition物件對於一個等待佇列，也就是說AQS中只能存在一個同步佇列，但可擁有多個等待佇列。下面從程式碼層面看看被呼叫await()方法(其他await()實現原理類似)的執行緒是如何加入等待佇列的，而又是如何從等待佇列中被喚醒的。
public final void await() throws InterruptedException {
//判斷執行緒是否被中斷
if (Thread.interrupted())
throw new InterruptedException();
//建立新結點加入等待佇列並返回
Node node = addConditionWaiter();
//釋放當前執行緒鎖即釋放同步狀態
int savedState = fullyRelease(node);
int interruptMode = 0;
//判斷結點是否同步佇列(SyncQueue)中,即是否被喚醒
while (!isOnSyncQueue(node)) {
//掛起執行緒
LockSupport.park(this);
//判斷是否被中斷喚醒，如果是退出迴圈。
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
//被喚醒後執行自旋操作爭取獲得鎖，同時判斷執行緒是否被中斷
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
// clean up if cancelled
if (node.nextWaiter != null)
//清理等待佇列中不為CONDITION狀態的結點
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}
複製程式碼再看看addConditionWaiter方法，新增到等待佇列：
private Node addConditionWaiter() {
Node t = lastWaiter;
// 判斷是否為結束狀態的結點並移除
if (t != null && t.waitStatus != Node.CONDITION) {
unlinkCancelledWaiters();
t = lastWaiter;
}
//建立新結點狀態為CONDITION
Node node = new Node(Thread.currentThread(), Node.CONDITION);
//加入等待佇列
if (t == null)
firstWaiter = node;
else
t.nextWaiter = node;
lastWaiter = node;
return node;
}
複製程式碼await()方法主要做了3件事：
一是呼叫addConditionWaiter()方法將當前執行緒封裝成node結點加入等待佇列。
二是呼叫fullyRelease(node)方法釋放同步狀態並喚醒後繼結點的執行緒。
三是呼叫isOnSyncQueue(node)方法判斷結點是否在同步佇列中，注意是個while迴圈，如果同步佇列中沒有該結點就直接掛起該執行緒，需要明白的是如果執行緒被喚醒後就呼叫acquireQueued(node, savedState)執行自旋操作爭取鎖，即當前執行緒結點從等待佇列轉移到同步佇列並開始努力獲取鎖。
接下來看看Singnal
public final void signal() {
//判斷是否持有獨佔鎖，如果不是丟擲異常
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node first = firstWaiter;
//喚醒等待佇列第一個結點的執行緒
if (first != null)
doSignal(first);
}
複製程式碼這裡signal()方法做了兩件事：
一是判斷當前執行緒是否持有獨佔鎖，沒有就丟擲異常，從這點也可以看出只有獨佔模式（例如ReentrantLock）先採用等待佇列，而共享模式（例如Semaphor）下是沒有等待佇列的，也就沒法使用Condition。
二是喚醒等待佇列的第一個結點，即執行doSignal(first)
private void doSignal(Node first) {
do {
//移除條件等待佇列中的第一個結點，
//如果後繼結點為null，那麼說沒有其他結點將尾結點也設定為null
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
first.nextWaiter = null;
//如果被通知節點沒有進入到同步佇列並且條件等待佇列還有不為空的節點，則繼續迴圈通知後續結點
} while (!transferForSignal(first) &&
(first = firstWaiter) != null);
}
//transferForSignal方法
final boolean transferForSignal(Node node) {
//嘗試設定喚醒結點的waitStatus為0，即初始化狀態
//如果設定失敗（compareAndSetWaitStatus返回false)，說明當期結點node的waitStatus已不為
//CONDITION狀態，那麼只能是結束狀態了，因此返回false
//返回doSignal()方法中繼續喚醒其他結點的執行緒，注意這裡並
//不涉及併發問題，所以CAS操作失敗只可能是預期值不為CONDITION，
//而不是多執行緒設定導致預期值變化，畢竟操作該方法的執行緒是持有鎖的。
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0)){
return false;
}
//加入同步佇列並返回前驅結點p
Node p = enq(node);
int ws = p.waitStatus;
//判斷前驅結點是否為結束結點(CANCELLED=1)或者在設定
//前驅節點狀態為Node.SIGNAL狀態失敗時，喚醒被通知節點代表的執行緒
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)){
//喚醒node結點的執行緒
LockSupport.unpark(node.thread);
return true;
}
}
複製程式碼doSignal(first)方法中做了2件事：
一是從條件佇列移除被喚醒的節點，然後重新維護條件條件佇列的firstWaiter和lastWaiter的指向。
二是將從條件佇列移除的結點加入同步佇列(在transferForSignal()方法中完成的)，如果進入到同步佇列失敗並且條件佇列還有不為空的節點，則繼續迴圈喚醒後續其他結點的執行緒。
總結：
signal()被呼叫後，先判斷當前執行緒是否持有獨佔鎖，如果有，那麼喚醒當前Condition物件中條件佇列的第一個結點的執行緒，並從條件佇列中移除該結點，移動到同步佇列中，如果加入同步佇列失敗（此時只有可能執行緒被取消），那麼繼續迴圈喚醒條件佇列中的其他結點的執行緒，如果成功加入同步佇列，那麼如果其前驅結點是否已結束或者設定前驅節點狀態為Node.SIGNAL狀態失敗，則通過LockSupport.unpark()喚醒被通知節點代表的執行緒，到此signal()任務完成，注意被喚醒後的執行緒，將從前面的await()方法中的while迴圈中退出，因為此時該執行緒的結點已在同步佇列中，那麼while (!isOnSyncQueue(node))將不在符合迴圈條件，進而呼叫AQS的acquireQueued()方法加入獲取同步狀態的競爭中，這就是等待喚醒機制的整個流程實現原理，流程如下圖（注意無論是同步佇列還是條件佇列使用的Node資料結構都是同一個，不過是使用的內部變數不同罷了）