摘要：造成當前線程在接到信號被中斷或到達指定最后期限之前一直處于等待狀態。該線程從等待方法返回前必須獲得與相關的鎖。如果線程已經獲取了鎖，則將喚醒條件隊列的首節點。

在前幾篇我們聊了 AQS、CLH、ReentrantLock、ReentrantReadWriteLock等的原理以及其源碼解讀，具體參見專欄 《非學無以廣才》

這章我們一起聊聊顯示的Condition 對象。

在沒有Lock之前，我們使用synchronized來控制同步，配合Object的wait()、wait(long timeout)、notify()、以及notifyAll 等方法可以實現等待/通知模式。

Condition接口也提供了類似于Object的監聽器方法、與Lock接口配合可以實現等待/通知模式，但是兩者還是有很大區別的，下圖是兩者的對比

參考《Java并發編程藝術》

Java API 摘要

Condition 將 Object 監視器方法（wait、notify 和 notifyAll）分解成截然不同的對象，以便通過將這些對象與任意 Lock 實現組合使用，為每個對象提供多個等待 set（wait-set）。其中，Lock 替代了 synchronized 方法和語句的使用，Condition 替代了 Object 監視器方法的使用。
條件（也稱為條件隊列 或條件變量）為線程提供了一個含義，以便在某個狀態條件現在可能為 true 的另一個線程通知它之前，一直掛起該線程（即讓其“等待”）。因為訪問此共享狀態信息發生在不同的線程中，所以它必須受保護，因此要將某種形式的鎖與該條件相關聯。等待提供一個條件的主要屬性是：以原子方式 釋放相關的鎖，并掛起當前線程，就像 Object.wait 做的那樣。
Condition 實例實質上被綁定到一個鎖上。要為特定 Lock 實例獲得 Condition 實例，請使用其 newCondition() 方法。

Condition提供了一系列的方法來對阻塞和喚醒線程：

await()：造成當前線程在接到信號或被中斷之前一直處于等待狀態。

await(long time, TimeUnit unit) ：造成當前線程在接到信號、被中斷或到達指定等待時間之前一直處于等待狀態。

awaitNanos(long nanosTimeout) ：造成當前線程在接到信號、被中斷或到達指定等待時間之前一直處于等待狀態。返回值表示剩余時間，如果在nanosTimesout之前喚醒，那么返回值 = nanosTimeout - 消耗時間，如果返回值 <= 0 ,則可以認定它已經超時了。

awaitUninterruptibly() ：造成當前線程在接到信號之前一直處于等待狀態。【注意：該方法對中斷不敏感】。

awaitUntil(Date deadline) ：造成當前線程在接到信號、被中斷或到達指定最后期限之前一直處于等待狀態。如果沒有到指定時間就被通知，則返回true，否則表示到了指定時間，返回返回false。

signal() ：喚醒一個等待線程。該線程從等待方法返回前必須獲得與Condition相關的鎖。

signalAll() ：喚醒所有等待線程。能夠從等待方法返回的線程必須獲得與Condition相關的鎖。

Condition是一種廣義上的條件隊列。他為線程提供了一種更為靈活的等待/通知模式，線程在調用await方法后執行掛起操作，直到線程等待的某個條件為真時才會被喚醒。Condition必須要配合鎖一起使用，因為對共享狀態變量的訪問發生在多線程環境下。一個Condition的實例必須與一個Lock綁定，因此Condition一般都是作為Lock的內部實現。

獲取一個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;
        
        /** First node of condition queue. */
        private transient Node firstWaiter;
        
        /** Last node of condition queue. */
        private transient Node lastWaiter;

        /**
         * Creates a new {@code ConditionObject} instance.
         */
        public ConditionObject() { }

        // Internal methods
        // 省略方法
}

從上面代碼可以看出Condition擁有首節點（firstWaiter），尾節點（lastWaiter）。

當前線程調用await()方法，將會以當前線程構造成一個節點（Node），并將節點加入到該隊列的尾部。

圖來源《Java 并發編程藝術》

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();
            
            //釋放鎖
            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);
        }

此段代碼的邏輯是：
首先將當前線程新建一個節點同時加入到等待隊列中，然后釋放當前線程持有的同步狀態。

然后則是不斷檢測該節點代表的線程是否出現在CLH同步隊列中（收到signal信號之后就會在AQS隊列中檢測到），如果不存在則一直掛起，否則參與競爭同步狀態。

加入條件隊列（addConditionWaiter()）源碼如下

private Node addConditionWaiter() {
            
            //隊列的尾節點
            Node t = lastWaiter;
            // If lastWaiter is cancelled, clean out.
            
            // 如果該節點的狀態的不是CONDITION，則說明該節點不在等待隊列上，需要 清除
            if (t != null && t.waitStatus != Node.CONDITION) {
                
                // 清除等待隊列中狀態部位 CONDITION 的節點
                unlinkCancelledWaiters();
                
                //清除后從新獲取尾節點
                t = lastWaiter;
            }
            
            // 將當前線程構造成等待節點
            Node node = new Node(Thread.currentThread(), Node.CONDITION);
            
            // 將node 節點添加到等待隊列的尾部
            
            if (t == null)
                firstWaiter = node;
            else
                t.nextWaiter = node;
            lastWaiter = node;
            return node;
        }

該方法主要是將當前線程加入到Condition條件隊列中。當然在加入到尾節點之前會清除所有狀態不為Condition的節點。

fullyRelease(Node node)，負責釋放該線程持有的鎖

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

isOnSyncQueue(Node node)：如果一個節點剛開始在條件隊列上，現在在同步隊列上獲取鎖則返回true。

final boolean isOnSyncQueue(Node node) {
        
        // 狀態為CONDITION 、前驅節點為空，返回false
        if (node.waitStatus == Node.CONDITION || node.prev == null)
            return false;
            
         // 如果后繼節點不為空，則說明節點肯定在同步隊列中
        if (node.next != null) // If has successor, it must be on queue
            return true;
        /*
         * node.prev can be non-null, but not yet on queue because
         * the CAS to place it on queue can fail. So we have to
         * traverse from tail to make sure it actually made it.  It
         * will always be near the tail in calls to this method, and
         * unless the CAS failed (which is unlikely), it will be
         * there, so we hardly ever traverse much.
         */
        return findNodeFromTail(node);
    }

unlinkCancelledWaiters()：負責將條件隊列中狀態不為Condition的節點刪除。

private void unlinkCancelledWaiters() {
            
            // 首節點
            Node t = firstWaiter;
            
            Node trail = null;
            // 從頭開始清除狀態不為CONDITION的節點
            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() {

            // 如果同步是以獨占方式進行的，則返回 true；其他情況則返回 false  
            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)主要是做兩件事：

修改頭節點；

調用transferForSignal(Node first) 方法將節點移動到CLH同步隊列中。

transferForSignal(Node first)源碼如下

final boolean transferForSignal(Node node) {
        /*
         * If cannot change waitStatus, the node has been cancelled.
         */
         // 將該節點的狀態改為初始狀態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).
         */
         
         // 將該節點添加到同步隊列的尾部、返回的是舊的尾部節點，也就是 node.prev節點
        Node p = enq(node);
        
        //如果結點p的狀態為cancel 或者修改waitStatus失敗，則直接喚醒
        int ws = p.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()方法競爭同步狀態。

注：本章參考《Java 并發編程藝術》、《Java 并發編程實戰》

本人技術有限，如果錯誤，歡迎拍磚