Java多執行緒之JUC包：Condition原始碼學習筆記
阿新 • • 發佈：2019-01-13
若有不正之處請多多諒解，並歡迎批評指正。
請尊重作者勞動成果，轉載請標明原文連結：
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;View Code/** 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(); }