【死磕Java併發】-----J.U.C之Condition
此篇部落格所有原始碼均來自JDK 1.8
在沒有Lock之前，我們使用synchronized來控制同步，配合Object的wait()、notify()系列方法可以實現等待/通知模式。在Java SE5後，Java提供了Lock介面，相對於Synchronized而言，Lock提供了條件Condition，對執行緒的等待、喚醒操作更加詳細和靈活。下圖是Condition與Object的監視器方法的對比（摘自《Java併發程式設計的藝術》）：
Condition提供了一系列的方法來對阻塞和喚醒執行緒：
- await() ：造成當前執行緒在接到訊號或被中斷之前一直處於等待狀態。
- await(long time, TimeUnit unit) ：造成當前執行緒在接到訊號、被中斷或到達指定等待時間之前一直處於等待狀態。
- awaitNanos(long nanosTimeout) ：造成當前執行緒在接到訊號、被中斷或到達指定等待時間之前一直處於等待狀態。返回值表示剩餘時間，如果在nanosTimesout之前喚醒，那麼返回值 = nanosTimeout - 消耗時間，如果返回值 <= 0 ,則可以認定它已經超時了。
- awaitUninterruptibly() ：造成當前執行緒在接到訊號之前一直處於等待狀態。【注意：該方法對中斷不敏感】。
- awaitUntil(Date deadline) ：造成當前執行緒在接到訊號、被中斷或到達指定最後期限之前一直處於等待狀態。如果沒有到指定時間就被通知，則返回true，否則表示到了指定時間，返回返回false。
- signal()：喚醒一個等待執行緒。該執行緒從等待方法返回前必須獲得與Condition相關的鎖。
- signal()All：喚醒所有等待執行緒。能夠從等待方法返回的執行緒必須獲得與Condition相關的鎖。
Condition是一種廣義上的條件佇列。他為執行緒提供了一種更為靈活的等待/通知模式，執行緒在呼叫await方法後執行掛起操作，直到執行緒等待的某個條件為真時才會被喚醒。Condition必須要配合鎖一起使用，因為對共享狀態變數的訪問發生在多執行緒環境下。一個Condition的例項必須與一個Lock繫結，因此Condition一般都是作為Lock的內部實現。
Condtion的實現
獲取一個Condition必須要通過Lock的newCondition()方法。該方法定義在介面Lock下面，返回的結果是繫結到此 Lock 例項的新 Condition 例項。Condition為一個介面，其下僅有一個實現類ConditionObject，由於Condition的操作需要獲取相關的鎖，而AQS則是同步鎖的實現基礎，所以ConditionObject則定義為AQS的內部類。定義如下：
public class ConditionObject implements Condition, java.io.Serializable {
}等待佇列
每個Condition物件都包含著一個FIFO佇列，該佇列是Condition物件通知/等待功能的關鍵。在佇列中每一個節點都包含著一個執行緒引用，該執行緒就是在該Condition物件上等待的執行緒。我們看Condition的定義就明白了：
public class ConditionObject implements Condition, java.io.Serializable {
private static final long serialVersionUID = 1173984872572414699L;
//頭節點
private transient Node firstWaiter;
//尾節點
private transient Node lastWaiter;
public ConditionObject() {
}
/** 省略方法 **/
}從上面程式碼可以看出Condition擁有首節點（firstWaiter），尾節點（lastWaiter）。當前執行緒呼叫await()方法，將會以當前執行緒構造成一個節點（Node），並將節點加入到該佇列的尾部。結構如下：
Node裡面包含了當前執行緒的引用。Node定義與AQS的CLH同步佇列的節點使用的都是同一個類（AbstractQueuedSynchronized.Node靜態內部類）。
Condition的佇列結構比CLH同步佇列的結構簡單些，新增過程較為簡單隻需要將原尾節點的nextWaiter指向新增節點，然後更新lastWaiter即可。
等待
呼叫Condition的await()方法會使當前執行緒進入等待狀態，同時會加入到Condition等待佇列同時釋放鎖。當從await()方法返回時，當前執行緒一定是獲取了Condition相關連的鎖。
public final void await() throws InterruptedException {
// 當前執行緒中斷
if (Thread.interrupted())
throw new InterruptedException();
//當前執行緒加入等待佇列
Node node = addConditionWaiter();
//釋放鎖
long 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);
}此段程式碼的邏輯是：首先將當前執行緒新建一個節點同時加入到條件佇列中，然後釋放當前執行緒持有的同步狀態。然後則是不斷檢測該節點代表的執行緒釋放出現在CLH同步佇列中（收到signal訊號之後就會在AQS佇列中檢測到），如果不存在則一直掛起，否則參與競爭同步狀態。
加入條件佇列（addConditionWaiter()）原始碼如下：
private Node addConditionWaiter() {
Node t = lastWaiter; //尾節點
//Node的節點狀態如果不為CONDITION，則表示該節點不處於等待狀態，需要清除節點
if (t != null && t.waitStatus != Node.CONDITION) {
//清除條件佇列中所有狀態不為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;
}該方法主要是將當前執行緒加入到Condition條件佇列中。當然在加入到尾節點之前會清楚所有狀態不為Condition的節點。
fullyRelease(Node node)，負責釋放該執行緒持有的鎖。
final long fullyRelease(Node node) {
boolean failed = true;
try {
//節點狀態--其實就是持有鎖的數量
long savedState = getState();
//釋放鎖
if (release(savedState)) {
failed = false;
return savedState;
} else {
throw new IllegalMonitorStateException();
}
} finally {
if (failed)
node.waitStatus = Node.CANCELLED;
}
}isOnSyncQueue(Node node)：如果一個節點剛開始在條件佇列上，現在在同步佇列上獲取鎖則返回true
final boolean isOnSyncQueue(Node node) {
//狀態為Condition，獲取前驅節點為null，返回false
if (node.waitStatus == Node.CONDITION || node.prev == null)
return false;
//後繼節點不為null，肯定在CLH同步佇列中
if (node.next != null)
return true;
return findNodeFromTail(node);
}unlinkCancelledWaiters()：負責將條件佇列中狀態不為Condition的節點刪除
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;
}
}通知
呼叫Condition的signal()方法，將會喚醒在等待佇列中等待最長時間的節點（條件佇列裡的首節點），在喚醒節點前，會將節點移到CLH同步佇列中。
public final void signal() {
//檢測當前執行緒是否為擁有鎖的獨
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
//頭節點，喚醒條件佇列中的第一個節點
Node first = firstWaiter;
if (first != null)
doSignal(first); //喚醒
}該方法首先會判斷當前執行緒是否已經獲得了鎖，這是前置條件。然後喚醒條件佇列中的頭節點。
doSignal(Node first)：喚醒頭節點
private void doSignal(Node first) {
do {
//修改頭結點，完成舊頭結點的移出工作
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
first.nextWaiter = null;
} while (!transferForSignal(first) &&
(first = firstWaiter) != null);
}doSignal(Node first)主要是做兩件事：1.修改頭節點，2.呼叫transferForSignal(Node first) 方法將節點移動到CLH同步佇列中。transferForSignal(Node first)原始碼如下：
final boolean transferForSignal(Node node) {
//將該節點從狀態CONDITION改變為初始狀態0,
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;
//將節點加入到syn佇列中去，返回的是syn佇列中node節點前面的一個節點
Node p = enq(node);
int ws = p.waitStatus;
//如果結點p的狀態為cancel 或者修改waitStatus失敗，則直接喚醒
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
return true;
}整個通知的流程如下：
- 判斷當前執行緒是否已經獲取了鎖，如果沒有獲取則直接丟擲異常，因為獲取鎖為通知的前置條件。
- 如果執行緒已經獲取了鎖，則將喚醒條件佇列的首節點
- 喚醒首節點是先將條件佇列中的頭節點移出，然後呼叫AQS的enq(Node node)方法將其安全地移到CLH同步佇列中
- 最後判斷如果該節點的同步狀態是否為Cancel，或者修改狀態為Signal失敗時，則直接呼叫LockSupport喚醒該節點的執行緒。
總結
一個執行緒獲取鎖後，通過呼叫Condition的await()方法，會將當前執行緒先加入到條件佇列中，然後釋放鎖，最後通過isOnSyncQueue(Node node)方法不斷自檢看節點是否已經在CLH同步隊列了，如果是則嘗試獲取鎖，否則一直掛起。當執行緒呼叫signal()方法後，程式首先檢查當前執行緒是否獲取了鎖，然後通過doSignal(Node first)方法喚醒CLH同步佇列的首節點。被喚醒的執行緒，將從await()方法中的while迴圈中退出來，然後呼叫acquireQueued()方法競爭同步狀態。
Condition的應用
只知道原理，如果不知道使用那就坑爹了，下面是用Condition實現的生產者消費者問題：
public class ConditionTest {
private LinkedList<String> buffer; //容器
private int maxSize ; //容器最大
private Lock lock;
private Condition fullCondition;
private Condition notFullCondition;
ConditionTest(int maxSize){
this.maxSize = maxSize;
buffer = new LinkedList<String>();
lock = new ReentrantLock();
fullCondition = lock.newCondition();
notFullCondition = lock.newCondition();
}
public void set(String string) throws InterruptedException {
lock.lock(); //獲取鎖
try {
while (maxSize == buffer.size()){
notFullCondition.await(); //滿了，新增的執行緒進入等待狀態
}
buffer.add(string);
fullCondition.signal();
} finally {
lock.unlock(); //記得釋放鎖
}
}
public String get() throws InterruptedException {
String string;
lock.lock();
try {
while (buffer.size() == 0){
fullCondition.await();
}
string = buffer.poll();
notFullCondition.signal();
} finally {
lock.unlock();
}
return string;
}
}歡迎掃一掃我的公眾號關注 — 及時得到部落格訂閱哦！
–— Java成神之路： 488391811（一起走向Java成神） –—
