詳解Condition的await和signal等待/通知機制

時間 2019-11-30

標籤詳解 condition await signal 等待通知機制欄目 Java 简体版

原文原文鏈接

本人免費整理了Java高級資料，涵蓋了Java、Redis、MongoDB、MySQL、Zookeeper、Spring Cloud、Dubbo高併發分佈式等教程，一共30G，須要本身領取。
傳送門：https://mp.weixin.qq.com/s/JzddfH-7yNudmkjT0IRL8Q

1.Condition簡介

任何一個java對象都自然繼承於Object類，在線程間實現通訊的每每會應用到Object的幾個方法，好比wait(),wait(long timeout),wait(long timeout, int nanos)與notify(),notifyAll()幾個方法實現等待/通知機制，一樣的，在java Lock體系下依然會有一樣的方法實現等待/通知機制。從總體上來看Object的wait和notify/notify是與對象監視器配合完成線程間的等待/通知機制，而Condition與Lock配合完成等待通知機制，前者是java底層級別的，後者是語言級別的，具備更高的可控制性和擴展性。二者除了在使用方式上不一樣外，在功能特性上仍是有不少的不一樣：java

Condition可以支持不響應中斷，而經過使用Object方式不支持；node
Condition可以支持多個等待隊列（new 多個Condition對象），而Object方式只能支持一個；面試
Condition可以支持超時時間的設置，而Object不支持併發

參照Object的wait和notify/notifyAll方法，Condition也提供了一樣的方法：app

針對Object的wait方法

void await() throws InterruptedException:當前線程進入等待狀態，若是其餘線程調用condition的signal或者signalAll方法而且當前線程獲取Lock從await方法返回，若是在等待狀態中被中斷會拋出被中斷異常；less
long awaitNanos(long nanosTimeout)：當前線程進入等待狀態直到被通知，中斷或者超時；分佈式
boolean await(long time, TimeUnit unit)throws InterruptedException：同第二種，支持自定義時間單位ide
boolean awaitUntil(Date deadline) throws InterruptedException：當前線程進入等待狀態直到被通知，中斷或者到了某個時間高併發

針對Object的notify/notifyAll方法

void signal()：喚醒一個等待在condition上的線程，將該線程從等待隊列中轉移到同步隊列中，若是在同步隊列中可以競爭到Lock則能夠從等待方法中返回。學習
void signalAll()：與1的區別在於可以喚醒全部等待在condition上的線程

2.Condition實現原理分析

2.1 等待隊列

要想可以深刻的掌握condition仍是應該知道它的實現原理，如今咱們一塊兒來看看condiiton的源碼。建立一個condition對象是經過lock.newCondition(),而這個方法其實是會new出一個ConditionObject對象，該類是AQS的一個內部類，有興趣能夠去看看。前面咱們說過，condition是要和lock配合使用的也就是condition和Lock是綁定在一塊兒的，而lock的實現原理又依賴於AQS，天然而然ConditionObject做爲AQS的一個內部類無可厚非。

咱們知道在鎖機制的實現上，AQS內部維護了一個同步隊列，若是是獨佔式鎖的話，全部獲取鎖失敗的線程的尾插入到同步隊列，一樣的，condition內部也是使用一樣的方式，內部維護了一個等待隊列，全部調用condition.await方法的線程會加入到等待隊列中，而且線程狀態轉換爲等待狀態。另外注意到ConditionObject中有兩個成員變量：

/** First node of condition queue. */
private transient Node firstWaiter;
/** Last node of condition queue. */
private transient Node lastWaiter;

這樣咱們就能夠看出來ConditionObject經過持有等待隊列的頭尾指針來管理等待隊列。主要注意的是Node類複用了在AQS中的Node類，其節點狀態和相關屬性能夠去看，若是您仔細看完這篇文章對condition的理解易如反掌，對lock體系的實現也會有一個質的提高。Node類有這樣一個屬性：

//後繼節點
Node nextWaiter;

進一步說明，等待隊列是一個單向隊列，而在以前說AQS時知道同步隊列是一個雙向隊列。接下來咱們用一個demo，經過debug進去看是否是符合咱們的猜測：

public static void main(String[] args) {
    for (int i = 0; i < 10; i++) {
        Thread thread = new Thread(() -> {
            lock.lock();
            try {
                condition.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }finally {
                lock.unlock();
            }
        });
        thread.start();
    }
}

這段代碼沒有任何實際意義，甚至很臭，只是想說明下咱們剛纔所想的。新建了10個線程，沒有線程先獲取鎖，而後調用condition.await方法釋放鎖將當前線程加入到等待隊列中，經過debug控制當走到第10個線程的時候查看firstWaiter即等待隊列中的頭結點，debug模式下情景圖以下：

從這個圖咱們能夠很清楚的看到這樣幾點：
1. 調用condition.await方法後線程依次尾插入到等待隊列中，如圖隊列中的線程引用依次爲Thread-0,Thread-1,Thread-2....Thread-8；
2. 等待隊列是一個單向隊列。經過咱們的猜測而後進行實驗驗證，咱們能夠得出等待隊列的示意圖以下圖所示：

同時還有一點須要注意的是：咱們能夠屢次調用lock.newCondition()方法建立多個condition對象，也就是一個lock能夠持有多個等待隊列。而在以前利用Object的方式其實是指在對象Object對象監視器上只能擁有一個同步隊列和一個等待隊列，而併發包中的Lock擁有一個同步隊列和多個等待隊列。示意圖以下：

如圖所示，ConditionObject是AQS的內部類，所以每一個ConditionObject可以訪問到AQS提供的方法，至關於每一個Condition都擁有所屬同步器的引用。

2.2 await實現原理

當調用condition.await()方法後會使得當前獲取lock的線程進入到等待隊列，若是該線程可以從await()方法返回的話必定是該線程獲取了與condition相關聯的lock。接下來，咱們仍是從源碼的角度去看，只有熟悉了源碼的邏輯咱們的理解纔是最深的。await()方法源碼爲：

public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    // 1\. 將當前線程包裝成Node，尾插入到等待隊列中
    Node node = addConditionWaiter();
    // 2\. 釋放當前線程所佔用的lock，在釋放的過程當中會喚醒同步隊列中的下一個節點
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    while (!isOnSyncQueue(node)) {
        // 3\. 當前線程進入到等待狀態
        LockSupport.park(this);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    // 4\. 自旋等待獲取到同步狀態（即獲取到lock）
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    // 5\. 處理被中斷的狀況
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

代碼的主要邏輯請看註釋，咱們都知道噹噹前線程調用condition.await()方法後，會使得當前線程釋放lock而後加入到等待隊列中，直至被signal/signalAll後會使得當前線程從等待隊列中移至到同步隊列中去，直到得到了lock後纔會從await方法返回，或者在等待時被中斷會作中斷處理。那麼關於這個實現過程咱們會有這樣幾個問題：1. 是怎樣將當前線程添加到等待隊列中去的？2.釋放鎖的過程？3.怎樣才能從await方法退出？而這段代碼的邏輯就是告訴咱們這三個問題的答案。具體請看註釋，在第1步中調用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 node = new Node(Thread.currentThread(), Node.CONDITION);
    if (t == null)
        firstWaiter = node;
    else
        //尾插入
        t.nextWaiter = node;
    //更新lastWaiter
    lastWaiter = node;
    return node;
}

這段代碼就很容易理解了，將當前節點包裝成Node，若是等待隊列的firstWaiter爲null的話（等待隊列爲空隊列），則將firstWaiter指向當前的Node,不然，更新lastWaiter(尾節點)便可。就是經過尾插入的方式將當前線程封裝的Node插入到等待隊列中便可，同時能夠看出等待隊列是一個不帶頭結點的鏈式隊列，以前咱們學習AQS時知道同步隊列是一個帶頭結點的鏈式隊列，這是二者的一個區別。將當前節點插入到等待對列以後，會使當前線程釋放lock，由fullyRelease方法實現，fullyRelease源碼爲：

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;
    }
}

這段代碼就很容易理解了，調用AQS的模板方法release方法釋放AQS的同步狀態而且喚醒在同步隊列中頭結點的後繼節點引用的線程，若是釋放成功則正常返回，若失敗的話就拋出異常。到目前爲止，這兩段代碼已經解決了前面的兩個問題的答案了，還剩下第三個問題，怎樣從await方法退出？如今回過頭再來看await方法有這樣一段邏輯：

while (!isOnSyncQueue(node)) {
    // 3\. 當前線程進入到等待狀態
    LockSupport.park(this);
    if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
        break;
}

很顯然，當線程第一次調用condition.await()方法時，會進入到這個while()循環中，而後經過LockSupport.park(this)方法使得當前線程進入等待狀態，那麼要想退出這個await方法第一個前提條件天然而然的是要先退出這個while循環，出口就只剩下兩個地方：1. 邏輯走到break退出while循環；2. while循環中的邏輯判斷爲false。

再看代碼出現第1種狀況的條件是當前等待的線程被中斷後代碼會走到break退出，第二種狀況是當前節點被移動到了同步隊列中（即另外線程調用的condition的signal或者signalAll方法），while中邏輯判斷爲false後結束while循環。總結下，就是當前線程被中斷或者調用condition.signal/condition.signalAll方法當前節點移動到了同步隊列後，這是當前線程退出await方法的前提條件。

當退出while循環後就會調用acquireQueued(node, savedState)，這個方法在介紹AQS的底層實現時說過了，若感興趣的話能夠去，該方法的做用是在自旋過程當中線程不斷嘗試獲取同步狀態，直至成功（線程獲取到lock）。這樣也說明了退出await方法必須是已經得到了condition引用（關聯）的lock。到目前爲止，開頭的三個問題咱們經過閱讀源碼的方式已經徹底找到了答案，也對await方法的理解加深。await方法示意圖以下圖：

如圖，調用condition.await方法的線程必須是已經得到了lock，也就是當前線程是同步隊列中的頭結點。調用該方法後會使得當前線程所封裝的Node尾插入到等待隊列中。

超時機制的支持

condition還額外支持了超時機制，使用者可調用方法awaitNanos,awaitUtil。這兩個方法的實現原理，基本上與AQS中的tryAcquire方法一模一樣，關於tryAcquire能夠仔細閱讀。

不響應中斷的支持

要想不響應中斷能夠調用condition.awaitUninterruptibly()方法，該方法的源碼爲：

public final void awaitUninterruptibly() {
    Node node = addConditionWaiter();
    int savedState = fullyRelease(node);
    boolean interrupted = false;
    while (!isOnSyncQueue(node)) {
        LockSupport.park(this);
        if (Thread.interrupted())
            interrupted = true;
    }
    if (acquireQueued(node, savedState) || interrupted)
        selfInterrupt();
}

這段方法與上面的await方法基本一致，只不過減小了對中斷的處理，並省略了reportInterruptAfterWait方法拋被中斷的異常。

2.3 signal/signalAll實現原理

調用condition的signal或者signalAll方法能夠將等待隊列中等待時間最長的節點移動到同步隊列中，使得該節點可以有機會得到lock。按照等待隊列是先進先出（FIFO）的，因此等待隊列的頭節點必然會是等待時間最長的節點，也就是每次調用condition的signal方法是將頭節點移動到同步隊列中。咱們來經過看源碼的方式來看這樣的猜測是否是對的，signal方法源碼爲：

public final void signal() {
    //1\. 先檢測當前線程是否已經獲取lock
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    //2\. 獲取等待隊列中第一個節點，以後的操做都是針對這個節點
    Node first = firstWaiter;
    if (first != null)
        doSignal(first);
}

signal方法首先會檢測當前線程是否已經獲取lock，若是沒有獲取lock會直接拋出異常，若是獲取的話再獲得等待隊列的頭指針引用的節點，以後的操做的doSignal方法也是基於該節點。下面咱們來看看doSignal方法作了些什麼事情，doSignal方法源碼爲：

private void doSignal(Node first) {
    do {
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        //1\. 將頭結點從等待隊列中移除
        first.nextWaiter = null;
        //2\. while中transferForSignal方法對頭結點作真正的處理
    } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
}

具體邏輯請看註釋，真正對頭節點作處理的邏輯在transferForSignal放，該方法源碼爲：

final boolean transferForSignal(Node node) {
    /*
     * If cannot change waitStatus, the node has been cancelled.
     */
    //1\. 更新狀態爲0
    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).
     */
    //2.將該節點移入到同步隊列中去
    Node p = enq(node);
    int ws = p.waitStatus;
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);
    return true;
}

關鍵邏輯請看註釋，這段代碼主要作了兩件事情1.將頭結點的狀態更改成CONDITION；2.調用enq方法，將該節點尾插入到同步隊列中，關於enq方法請看AQS的底層實現這篇文章。如今咱們能夠得出結論：調用condition的signal的前提條件是當前線程已經獲取了lock，該方法會使得等待隊列中的頭節點即等待時間最長的那個節點移入到同步隊列，而移入到同步隊列後纔有機會使得等待線程被喚醒，即從await方法中的LockSupport.park(this)方法中返回，從而纔有機會使得調用await方法的線程成功退出。signal執行示意圖以下圖：

signalAll

sigllAll與sigal方法的區別體如今doSignalAll方法上，前面咱們已經知道doSignal方法只會對等待隊列的頭節點進行操做，，而doSignalAll的源碼爲：

private void doSignalAll(Node first) {
    lastWaiter = firstWaiter = null;
    do {
        Node next = first.nextWaiter;
        first.nextWaiter = null;
        transferForSignal(first);
        first = next;
    } while (first != null);
}

該方法只不過期間等待隊列中的每個節點都移入到同步隊列中，即「通知」當前調用condition.await()方法的每個線程。

3. await與signal/signalAll的結合思考

文章開篇提到等待/通知機制，經過使用condition提供的await和signal/signalAll方法就能夠實現這種機制，而這種機制可以解決最經典的問題就是「生產者與消費者問題」，關於「生產者消費者問題」以後會用單獨的一篇文章進行講解，這也是面試的高頻考點。await和signal和signalAll方法就像一個開關控制着線程A（等待方）和線程B（通知方）。它們之間的關係能夠用下面一個圖來表現得更加貼切：

如圖，線程awaitThread先經過lock.lock()方法獲取鎖成功後調用了condition.await方法進入等待隊列，而另外一個線程signalThread經過lock.lock()方法獲取鎖成功後調用了condition.signal或者signalAll方法，使得線程awaitThread可以有機會移入到同步隊列中，當其餘線程釋放lock後使得線程awaitThread可以有機會獲取lock，從而使得線程awaitThread可以從await方法中退出執行後續操做。若是awaitThread獲取lock失敗會直接進入到同步隊列。

3. 一個例子

咱們用一個很簡單的例子說說condition的用法：

public class AwaitSignal {
    private static ReentrantLock lock = new ReentrantLock();
    private static Condition condition = lock.newCondition();
    private static volatile boolean flag = false;

    public static void main(String[] args) {
        Thread waiter = new Thread(new waiter());
        waiter.start();
        Thread signaler = new Thread(new signaler());
        signaler.start();
    }

    static class waiter implements Runnable {

        @Override
        public void run() {
            lock.lock();
            try {
                while (!flag) {
                    System.out.println(Thread.currentThread().getName() + "當前條件不知足等待");
                    try {
                        condition.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                System.out.println(Thread.currentThread().getName() + "接收到通知條件知足");
            } finally {
                lock.unlock();
            }
        }
    }

    static class signaler implements Runnable {

        @Override
        public void run() {
            lock.lock();
            try {
                flag = true;
                condition.signalAll();
            } finally {
                lock.unlock();
            }
        }
    }
}

輸出結果爲：

Thread-0當前條件不知足等待
Thread-0接收到通知，條件知足

開啓了兩個線程waiter和signaler，waiter線程開始執行的時候因爲條件不知足，執行condition.await方法使該線程進入等待狀態同時釋放鎖，signaler線程獲取到鎖以後更改條件，並通知全部的等待線程後釋放鎖。這時，waiter線程獲取到鎖，並因爲signaler線程更改了條件此時相對於waiter來講條件知足，繼續執行。