Java多線程之JUC包：Condition源碼學習筆記

若有不正之處請多多諒解，並歡迎批評指正。

請尊重作者勞動成果，轉載請標明原文鏈接：

http://www.cnblogs.com/go2sea/p/5630355.html

 

Condition在JUC框架下提供了傳統Java監視器風格的wait、notify和notifyAll相似的功能。

Condition必須被綁定到一個獨占鎖上使用。ReentrantLock中獲取Condition的方法為：

    public Condition newCondition() {
        return sync.newCondition();
    }
        
        final ConditionObject newCondition() {
            return new ConditionObject();
        }

直接初始化並返回了一個AQS提供的ConditionObject對象。因此，Condition實際上是AQS框架的內容。ConditionObject通過維護兩個成員變量：

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

來維護一個Condition等待隊列，並通過signal操作將Condition隊列中的線程移到Sync鎖等待隊列。

源代碼：

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

        /**
         * Adds a new waiter to wait queue.
         * @return its new wait node
         */
        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;
        }

        /**
         * Removes and transfers nodes until hit non-cancelled one or
         * null. Split out from signal in part to encourage compilers
         * to inline the case of no waiters.
         * @param first (non-null) the first node on condition queue
         */
        private void doSignal(Node first) {
            do {
                if ( (firstWaiter = first.nextWaiter) == null)
                    lastWaiter = null;
                first.nextWaiter = null;
            } while (!transferForSignal(first) &&
                     (first = firstWaiter) != null);
        }

        /**
         * Removes and transfers all nodes.
         * @param first (non-null) the first node on condition queue
         */
        private void doSignalAll(Node first) {
            lastWaiter = firstWaiter = null;
            do {
                Node next = first.nextWaiter;
                first.nextWaiter = null;
                transferForSignal(first);
                first = next;
            } while (first != null);
        }

        /**
         * Unlinks cancelled waiter nodes from condition queue.
         * Called only while holding lock. This is called when
         * cancellation occurred during condition wait, and upon
         * insertion of a new waiter when lastWaiter is seen to have
         * been cancelled. This method is needed to avoid garbage
         * retention in the absence of signals. So even though it may
         * require a full traversal, it comes into play only when
         * timeouts or cancellations occur in the absence of
         * signals. It traverses all nodes rather than stopping at a
         * particular target to unlink all pointers to garbage nodes
         * without requiring many re-traversals during cancellation
         * storms.
         */
        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;
            }
        }

        // public methods

        /**
         * Moves the longest-waiting thread, if one exists, from the
         * wait queue for this condition to the wait queue for the
         * owning lock.
         *
         * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
         *         returns {@code false}
         */
        public final void signal() {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            Node first = firstWaiter;
            if (first != null)
                doSignal(first);
        }

        /**
         * Moves all threads from the wait queue for this condition to
         * the wait queue for the owning lock.
         *
         * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
         *         returns {@code false}
         */
        public final void signalAll() {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            Node first = firstWaiter;
            if (first != null)
                doSignalAll(first);
        }

        /**
         * Implements uninterruptible condition wait.
         * <ol>
         * <li> Save lock state returned by {@link #getState}.
         * <li> Invoke {@link #release} with saved state as argument,
         *      throwing IllegalMonitorStateException if it fails.
         * <li> Block until signalled.
         * <li> Reacquire by invoking specialized version of
         *      {@link #acquire} with saved state as argument.
         * </ol>
         */
        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();
        }

        /*
         * For interruptible waits, we need to track whether to throw
         * InterruptedException, if interrupted while blocked on
         * condition, versus reinterrupt current thread, if
         * interrupted while blocked waiting to re-acquire.
         */

        /** Mode meaning to reinterrupt on exit from wait */
        private static final int REINTERRUPT =  1;
        /** Mode meaning to throw InterruptedException on exit from wait */
        private static final int THROW_IE    = -1;

        /**
         * Checks for interrupt, returning THROW_IE if interrupted
         * before signalled, REINTERRUPT if after signalled, or
         * 0 if not interrupted.
         */
        private int checkInterruptWhileWaiting(Node node) {
            return Thread.interrupted() ?
                (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
                0;
        }

        /**
         * Throws InterruptedException, reinterrupts current thread, or
         * does nothing, depending on mode.
         */
        private void reportInterruptAfterWait(int interruptMode)
            throws InterruptedException {
            if (interruptMode == THROW_IE)
                throw new InterruptedException();
            else if (interruptMode == REINTERRUPT)
                selfInterrupt();
        }

        /**
         * Implements interruptible condition wait.
         * <ol>
         * <li> If current thread is interrupted, throw InterruptedException.
         * <li> Save lock state returned by {@link #getState}.
         * <li> Invoke {@link #release} with saved state as argument,
         *      throwing IllegalMonitorStateException if it fails.
         * <li> Block until signalled or interrupted.
         * <li> Reacquire by invoking specialized version of
         *      {@link #acquire} with saved state as argument.
         * <li> If interrupted while blocked in step 4, throw InterruptedException.
         * </ol>
         */
        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);
        }

        /**
         * Implements timed condition wait.
         * <ol>
         * <li> If current thread is interrupted, throw InterruptedException.
         * <li> Save lock state returned by {@link #getState}.
         * <li> Invoke {@link #release} with saved state as argument,
         *      throwing IllegalMonitorStateException if it fails.
         * <li> Block until signalled, interrupted, or timed out.
         * <li> Reacquire by invoking specialized version of
         *      {@link #acquire} with saved state as argument.
         * <li> If interrupted while blocked in step 4, throw InterruptedException.
         * </ol>
         */
        public final long awaitNanos(long nanosTimeout)
                throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            Node node = addConditionWaiter();
            int savedState = fullyRelease(node);
            final long deadline = System.nanoTime() + nanosTimeout;
            int interruptMode = 0;
            while (!isOnSyncQueue(node)) {
                if (nanosTimeout <= 0L) {
                    transferAfterCancelledWait(node);
                    break;
                }
                if (nanosTimeout >= spinForTimeoutThreshold)
                    LockSupport.parkNanos(this, nanosTimeout);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
                nanosTimeout = deadline - System.nanoTime();
            }
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null)
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
            return deadline - System.nanoTime();
        }

        /**
         * Implements absolute timed condition wait.
         * <ol>
         * <li> If current thread is interrupted, throw InterruptedException.
         * <li> Save lock state returned by {@link #getState}.
         * <li> Invoke {@link #release} with saved state as argument,
         *      throwing IllegalMonitorStateException if it fails.
         * <li> Block until signalled, interrupted, or timed out.
         * <li> Reacquire by invoking specialized version of
         *      {@link #acquire} with saved state as argument.
         * <li> If interrupted while blocked in step 4, throw InterruptedException.
         * <li> If timed out while blocked in step 4, return false, else true.
         * </ol>
         */
        public final boolean awaitUntil(Date deadline)
                throws InterruptedException {
            long abstime = deadline.getTime();
            if (Thread.interrupted())
                throw new InterruptedException();
            Node node = addConditionWaiter();
            int savedState = fullyRelease(node);
            boolean timedout = false;
            int interruptMode = 0;
            while (!isOnSyncQueue(node)) {
                if (System.currentTimeMillis() > abstime) {
                    timedout = transferAfterCancelledWait(node);
                    break;
                }
                LockSupport.parkUntil(this, abstime);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
            }
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null)
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
            return !timedout;
        }

        /**
         * Implements timed condition wait.
         * <ol>
         * <li> If current thread is interrupted, throw InterruptedException.
         * <li> Save lock state returned by {@link #getState}.
         * <li> Invoke {@link #release} with saved state as argument,
         *      throwing IllegalMonitorStateException if it fails.
         * <li> Block until signalled, interrupted, or timed out.
         * <li> Reacquire by invoking specialized version of
         *      {@link #acquire} with saved state as argument.
         * <li> If interrupted while blocked in step 4, throw InterruptedException.
         * <li> If timed out while blocked in step 4, return false, else true.
         * </ol>
         */
        public final boolean await(long time, TimeUnit unit)
                throws InterruptedException {
            long nanosTimeout = unit.toNanos(time);
            if (Thread.interrupted())
                throw new InterruptedException();
            Node node = addConditionWaiter();
            int savedState = fullyRelease(node);
            final long deadline = System.nanoTime() + nanosTimeout;
            boolean timedout = false;
            int interruptMode = 0;
            while (!isOnSyncQueue(node)) {
                if (nanosTimeout <= 0L) {
                    timedout = transferAfterCancelledWait(node);
                    break;
                }
                if (nanosTimeout >= spinForTimeoutThreshold)
                    LockSupport.parkNanos(this, nanosTimeout);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
                nanosTimeout = deadline - System.nanoTime();
            }
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null)
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
            return !timedout;
        }

        //  support for instrumentation

        /**
         * Returns true if this condition was created by the given
         * synchronization object.
         *
         * @return {@code true} if owned
         */
        final boolean isOwnedBy(AbstractQueuedSynchronizer sync) {
            return sync == AbstractQueuedSynchronizer.this;
        }

        /**
         * Queries whether any threads are waiting on this condition.
         * Implements {@link AbstractQueuedSynchronizer#hasWaiters(ConditionObject)}.
         *
         * @return {@code true} if there are any waiting threads
         * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
         *         returns {@code false}
         */
        protected final boolean hasWaiters() {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
                if (w.waitStatus == Node.CONDITION)
                    return true;
            }
            return false;
        }

        /**
         * Returns an estimate of the number of threads waiting on
         * this condition.
         * Implements {@link AbstractQueuedSynchronizer#getWaitQueueLength(ConditionObject)}.
         *
         * @return the estimated number of waiting threads
         * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
         *         returns {@code false}
         */
        protected final int getWaitQueueLength() {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            int n = 0;
            for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
                if (w.waitStatus == Node.CONDITION)
                    ++n;
            }
            return n;
        }

        /**
         * Returns a collection containing those threads that may be
         * waiting on this Condition.
         * Implements {@link AbstractQueuedSynchronizer#getWaitingThreads(ConditionObject)}.
         *
         * @return the collection of threads
         * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
         *         returns {@code false}
         */
        protected final Collection<Thread> getWaitingThreads() {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            ArrayList<Thread> list = new ArrayList<Thread>();
            for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
                if (w.waitStatus == Node.CONDITION) {
                    Thread t = w.thread;
                    if (t != null)
                        list.add(t);
                }
            }
            return list;
        }
    }

下面我們就來分析下Condition的工作流程。

一、await 在條件變量上等待

分別是Condition隊列的頭結點和尾節點。Condition在調用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;
        }

顧名思義，此方法在Condition隊列中添加一個等待線程。首先，方法先檢查一下隊列尾節點是否還在等待Condition（如果被signal或者中斷，waitStatus會被修改為0或者CANCELLED）。如果尾節點被取消或者中斷，調用unlinkCancelledWaiters方法刪除Condition隊列中被cancel的節點。然后將當前線程封裝在一個Node中，添加到Condition隊列的尾部。這里由於我們在操縱Condition隊列的時候已經獲取了一個獨占鎖，因此不會發生競爭。

值得注意的是，Condition隊列與Sync隊列（鎖等待隊列）有幾點不同：①Condition隊列是一個單向鏈表，而Sync隊列是一個雙向鏈表；②Sync隊列在初始化的時候，會在隊列頭部添加一個空的dummy節點，它不持有任何線程，而Condition隊列初始化時，頭結點就開始持有等待線程了。

我們有必要在這里提一下Node對象中的nextWaiter成員、SHARED成員和EXCLUSIVE成員：

        /** Marker to indicate a node is waiting in shared mode */
        static final Node SHARED = new Node();
        /** Marker to indicate a node is waiting in exclusive mode */
        static final Node EXCLUSIVE = null;

        Node nextWaiter;

nextWaiter在共享模式下，被設置為SHARED，SHARED為一個final的空節點，用來表示當前模式是共享模式；默認情況下nextWaiter是null，EXCLUSIVE成員是一個final的null，因此默認模式是獨占模式。在Condition隊列中nextWaiter被用來指向隊列里的下一個等待線程。在一個線程從Condition隊列中被移除之后，nextWaiter被設置為空（EXCLUSIVE）。這再次表明：Condition必須被綁定在一個獨占鎖上使用。

我們來看一下unlinkCancelledWaiters方法：

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

unlinkCancelledWaiters方法很簡單，從頭到尾遍歷Condition隊列，移除被cancel或被中斷的節點。由於這里我們在操縱Condition隊列的時候已經獲取了所綁定的獨占鎖，因此不用擔心競爭的發生。

我們再來看一下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;
        }
    }

方法首先獲取了state的值，這個值表示可鎖被“重入”深度，並調用release釋放全部的重入獲取，如果成功，返貨這個深度，如果失敗，要將當前線程的waitStatus設置為CANCELLED。

我們再來看一下isOnSyncQueue方法，這個方法返節點是否在Sync隊列中等待鎖：

    final boolean isOnSyncQueue(Node node) {
        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);
    }

node從Condition隊列移除的第一步，就是設置waitStatus為其他值，因此是否等於Node.CONDITON可以作為判斷標志，如果等於，說明還在Condition隊列中，即不再Sync隊列里。在node被放入Sync隊列時，第一步就是設置node的prev為當前獲取到的尾節點，所以如果發現node的prev為null的話，可以確定node尚未被加入Sync隊列。

相似的，node被放入Sync隊列的最后一步是設置node的next，如果發現node的next不為null，說明已經完成了放入Sync隊列的過程，因此可以返回true。

當我們執行完兩個if而仍未返回時，node的prev一定不為null，next一定為null，這個時候可以認為node正處於放入Sync隊列的執行CAS操作執行過程中。而這個CAS操作有可能失敗，因此我們再給node一次機會，調用findNodeFromTail來檢測：

    private boolean findNodeFromTail(Node node) {
        Node t = tail;
        for (;;) {
            if (t == node)
                return true;
            if (t == null)
                return false;
            t = t.prev;
        }
    }

findNodeFromTail方法從尾部遍歷Sync隊列，如果檢查node是否在隊列中，如果還不在，此時node也許在CAS自旋中，在不久的將來可能會進到Sync隊列里。但我們已經等不了了，直接放回false。

我們再來看一下checkInterruptWhileWaiting方法：

        /** Mode meaning to reinterrupt on exit from wait */
        private static final int REINTERRUPT =  1;
        /** Mode meaning to throw InterruptedException on exit from wait */
        private static final int THROW_IE    = -1;

        /**
         * Checks for interrupt, returning THROW_IE if interrupted
         * before signalled, REINTERRUPT if after signalled, or
         * 0 if not interrupted.
         */
        private int checkInterruptWhileWaiting(Node node) {
            return Thread.interrupted() ?
                (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
                0;
        }

此方法在線程從park中醒來后調用，它的返回值有三種：0代表在park過程中沒有發生中斷；THORW_IE（1）代表發生了中斷，且在后續我們需要拋出中斷異常；REINTERRUPT表示發生了中斷，但在后續我們不拋出中斷異常，而是“補上”這次中斷。當沒有發生中斷時，我們返回0即可，當中斷發生時，返回THROW_IE or REINTERRUPT由transferAfterCancelledWait方法判斷：

    final boolean transferAfterCancelledWait(Node node) {
        if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
            enq(node);
            return true;
        }
        /*
         * If we lost out to a signal(), then we can't proceed
         * until it finishes its enq().  Cancelling during an
         * incomplete transfer is both rare and transient, so just
         * spin.
         */
        while (!isOnSyncQueue(node))
            Thread.yield();
        return false;
    }

transferAfterCancelledWait方法並不在ConditionObject中定義，而是由AQS提供。這個方法根據是否中斷發生時，是否有signal操作來“摻和”來返回結果。方法調用CAS操作將node的waitStatus從CONDITION設置為0，如果成功，說明當中斷發生時，說明沒有signal發生（signal的第一步是將node的waitStatus設置為0），在調用enq將線程放入Sync隊列后直接返回true，表示中斷先於signal發生，即中斷在await等待過程中發生，根據await的語義，在遇到中斷時需要拋出中斷異常，返回true告訴上層方法返回THROW_IT，后續會根據這個返回值做拋出中斷異常的處理。

如果CAS操作失敗，是否說明中斷后於signal發生呢？只能說這時候我們不能確定中斷和signal到底誰先發生，只是在我們做CAS操作的時候，他們倆已經都發生了（中斷->interrupted檢測->signal->CAS，或者signal->中斷->interrupted檢測->CAS都有可能），這時候我們無法判斷到底順序是怎樣，這里的處理是不管怎樣都返回false告訴上層方法返回REINTERRUPT，當做是signal先發生（線程被signal喚醒）來處理，后續根據這個返回值做“補上”中斷的處理。在返回false之前，我們要先做一下等待，直到當前線程被成功放入Sync鎖等待隊列。

因此，我們可以這樣總結：transferAfterCancelledWait的返回值表示了線程是否因為中斷從park中喚醒。

至此，我們終於可以正式來看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方法是及時響應中斷的。它首先檢查了一下中斷標志。然后調用addConditionWaiter將當前線程放入Condition隊列的尾，並順手清理了一下隊列里的無用節點。緊接着調用fullyRelease方法釋放當前線程持有的鎖。然后是一個while循環，這個循環會循環檢測線程的狀態，直到線程被signal或者中斷喚醒且被放入Sync鎖等待隊列。如果中斷發生的話，還需要調用checkInterruptWhileWaiting方法，根據中斷發生的時機確定后去處理這次中斷的方式，如果發生中斷，退出while循環。

退出while循環后，我們調用acquireQueued方法來獲取鎖，注意，acquireQueued方法的返回值表示在等待獲取鎖的過程中是否發生中斷，如果發生中斷 且 原來沒有需要做拋出處理的中斷發生時，我們將后續處理方式設置為REINTERRUPT（如果原來在await狀態有中斷發生，即interrruptMode==THROW_IE，依然保持THROW_IE）。

如果是應為中斷從park中喚醒（interruptMode==THROT_IE），當前線程仍在Condition隊列中，但waitStatus已經變成0了，這里在調用unlinkCancelledWaiters做一次清理。

最后，根據interruptMode的值，調用reportInterruptAfterWait做出相應處理：

        private void reportInterruptAfterWait(int interruptMode)
            throws InterruptedException {
            if (interruptMode == THROW_IE)
                throw new InterruptedException();
            else if (interruptMode == REINTERRUPT)
                selfInterrupt();
        }

如果interruptMod==0，donothing，如果是THROW_IE，說明在await狀態下發生中斷，拋出中斷異常，如果是REINTERRUPT，說明是signal“摻和”了中斷，我們無法分辨具體的先后順序，於是統一按照先signal再中斷來處理，即成功獲取鎖之后要調用selfInterrupt“補上”這次中斷。

二、awaitNanos 限時的在條件變量上等待

        public final long awaitNanos(long nanosTimeout)
                throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            Node node = addConditionWaiter();
            int savedState = fullyRelease(node);
            final long deadline = System.nanoTime() + nanosTimeout;
            int interruptMode = 0;
            while (!isOnSyncQueue(node)) {
                if (nanosTimeout <= 0L) {
                    transferAfterCancelledWait(node);
                    break;
                }
                if (nanosTimeout >= spinForTimeoutThreshold)
                    LockSupport.parkNanos(this, nanosTimeout);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
                nanosTimeout = deadline - System.nanoTime();
            }
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null)
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
            return deadline - System.nanoTime();
        }

awaitNanos方法與await方法大致相同，區別在於每次park是定時的，當被喚醒時，比較一下剩余等待時間Timeout與spinForTimeoutThreshold閾值的大小，如果小於，將不再park，spinForTimeoutThreshold閾值的作用在筆者的另一篇博文Semaphore源碼學習筆記中已經分析過，作用是提高短時長的等待的相應效率。

注意：當已經到達等待的deadline時，調用transferAfterCancelledWait方法，注意，此時可能發生中斷（上次調用checkInterruptWhileWaiting之后被中斷），再次的，我們無法判斷這次中斷與到時這兩個的先后順序，我們在這里的處理方式是直接忽略這次中斷，統一認為是先到時后中斷（體現在沒有記錄transferAfterCancelledWait方法的返回值），但在transferAfterCancelledWait方法中的處理是考慮了被中斷的情況的，只不過這個中斷標志位沒有檢測，留給后續來處理了。這個中斷標志將會在調用acquireQueued方法並成功獲取鎖之后被檢測並返回，最終影響interruptMode的值，並在reportInterruptAfterWait方法中被處理。可見，這次中斷最終沒有被遺漏，只是我們先處理的signal，回過頭來再去處理它。

最后方法的返回值是拍喚醒后的剩余等待時間，這個時間可能小於0。

await(long time, TimeUnit unit)方法與awaitNanos方法十分類似，不再贅述。

三、awaitUtil 指定結束時刻的在條件變量上等待

        public final boolean awaitUntil(Date deadline)
                throws InterruptedException {
            long abstime = deadline.getTime();
            if (Thread.interrupted())
                throw new InterruptedException();
            Node node = addConditionWaiter();
            int savedState = fullyRelease(node);
            boolean timedout = false;
            int interruptMode = 0;
            while (!isOnSyncQueue(node)) {
                if (System.currentTimeMillis() > abstime) {
                    timedout = transferAfterCancelledWait(node);
                    break;
                }
                LockSupport.parkUntil(this, abstime);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
            }
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null)
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
            return !timedout;
        }

awaitUtil方法在原理上與awaitNanos方法是也十分相似，只不過park操作調用的是LockSupportparkUtil方法，且沒有spinForTimeoutThreshold閾值的應用。在返回值上也有些許差別：返回值timedout記錄了transferAfterCancelledWait方法的返回值——線程是否因為中斷從park中喚醒，如果是的話，表示還沒有到等待的deadline。

四、signal 喚醒Condition隊列的頭節點持有的線程

        public final void signal() {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            Node first = firstWaiter;
            if (first != null)
                doSignal(first);
        }

調用signal之前也需要獲取鎖，因此signal方法首先檢測了一下當前線程是否獲取了獨占鎖。然后調用doSignal喚醒隊列中第一個等待線程。注意，這里的“喚醒”意思是將線程從Condition隊列移到Sync隊列，表示已經完成Condition的等待，具有了去競爭鎖的資格。至此，我們可以發現，由於await會直接把線程放入Condition等待隊列的尾部，因此Condition是公平的，即按照入列的順序來signal。

        private void doSignal(Node first) {
            do {
                if ( (firstWaiter = first.nextWaiter) == null)
                    lastWaiter = null;
                first.nextWaiter = null;
            } while (!transferForSignal(first) &&
                     (first = firstWaiter) != null);
        }

doSignal方法先將first節點從隊列中摘下，然后調用transferForSignal去改變first節點的waitStatus（所謂喚醒線程），這個方法有可能失敗，因為等待線程可能已經到時或者被中斷，因此while循環這個操作直到成功喚醒或隊列為空。我們來看下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;
    }

這個方法並不在ConditionObject中定義，而是由AQS提供。方法首先調用CAS操作修改node的waitStatus，如果失敗，表示線程已經放棄等待（到時或被中斷），直接返回false。如果成功，調用enq方法將它放入Sync鎖等待隊列，返回值p是node在Sync隊列中的前驅節點。緊接着檢測一下前驅p的waitStatus，如果發現不為SIGNAL，需要將node持有的線程（注意不是當前線程）unpark，這里必須搞清楚，node線程是在哪里park的，顯然，他還在await方法的那個while循環里。unpark之后，node線程將會從while循環中退出，然后去調用acquireQueued方法，這個方法是一個自旋，弄得線程會在自旋過程中清除已經為CANCELLED狀態的前驅，然后注冊前驅節點的waitStatus為SIGNAL。

至此，signal方法已經完成了所有該做的，“喚醒”的線程已經成功加入Sync隊列並已經參與鎖的競爭了，返回true。

五、signalAll 喚醒Condition隊列的所有等待線程

        public final void signalAll() {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            Node first = firstWaiter;
            if (first != null)
                doSignalAll(first);
        }

signalAll方法同樣先檢測是否持有獨占鎖，然后對奧用doSignalAll方法：

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

doSignalAll方法循環調用transferForSignal方法“喚醒”隊列的頭結點，直到隊列為空。

 

總結：ConditionObject由AQS提供，它實現了類似wiat、notify和notifyAll類似的功能。Condition必須與一個獨占鎖綁定使用，在await或signal之前必須現持有獨占鎖。Condition隊列是一個單向鏈表，他是公平的，按照先進先出的順序從隊列中被“喚醒”，所謂喚醒指的是完成Condition對象上的等待，被移到Sync鎖等待隊列中，有參與競爭鎖的資格（Sync隊列有公平&非公平兩種模式，注意區別）。