【JUC】JDK1.8源碼分析之SynchronousQueue（九）
1、前言php
本篇是在分析Executors源碼時，發現JUC集合框架中的一個重要類沒有分析，SynchronousQueue，該類在線程池中的做用是很是明顯的，因此頗有必要單獨拿出來分析一番，這對於以後理解線程池有頗有好處，SynchronousQueue是一種阻塞隊列，其中每一個插入操做必須等待另外一個線程的對應移除操做 ，反之亦然。同步隊列沒有任何內部容量，甚至連一個隊列的容量都沒有。css
2、SynchronousQueue數據結構java
因爲SynchronousQueue的支持公平策略和非公平策略，因此底層可能兩種數據結構：隊列（實現公平策略）和棧（實現非公平策略），隊列與棧都是經過鏈表來實現的。具體的數據結構以下node
說明：數據結構有兩種類型，棧和隊列；棧有一個頭結點，隊列有一個頭結點和尾結點；棧用於實現非公平策略，隊列用於實現公平策略。npm
3、SynchronousQueue源碼分析vim
3.1 類的繼承關係 數組
public class SynchronousQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable {}
說明：SynchronousQueue繼承了AbstractQueue抽象類，AbstractQueue定義了對隊列的基本操做；同時實現了BlockingQueue接口，BlockingQueue表示阻塞型的隊列，其對隊列的操做可能會拋出異常；同時也實現了Searializable接口，表示能夠被序列化。緩存
3.2 類的內部類數據結構
SynchronousQueue的內部類框架圖以下app
說明：其中比較重要的類是左側的三個類，Transferer是TransferStack棧和TransferQueue隊列的公共類，定義了轉移數據的公共操做，由TransferStack和TransferQueue具體實現，WaitQueue、LifoWaitQueue、FifoWaitQueue表示爲了兼容JDK1.5版本中的SynchronousQueue的序列化策略所遺留的，這裏不作具體的講解。下面着重看左側的三個類。
① Transferer
abstract static class Transferer<E> { /** * Performs a put or take. * * @param e if non-null, the item to be handed to a consumer; * if null, requests that transfer return an item * offered by producer. * @param timed if this operation should timeout * @param nanos the timeout, in nanoseconds * @return if non-null, the item provided or received; if null, * the operation failed due to timeout or interrupt -- * the caller can distinguish which of these occurred * by checking Thread.interrupted. */ // 轉移數據，put或者take操做 abstract E transfer(E e, boolean timed, long nanos); }
說明：Transferer定義了transfer操做，用於take或者put數據。transfer方法由子類實現。
② TransfererStack
1. 類的繼承關係
static final class TransferStack<E> extends Transferer<E> {}
說明：TransferStack繼承Transferer抽象類，其實現了transfer方法。
2. 類的屬性
static final class TransferStack<E> extends Transferer<E> { /* * This extends Scherer-Scott dual stack algorithm, differing, * among other ways, by using "covering" nodes rather than * bit-marked pointers: Fulfilling operations push on marker * nodes (with FULFILLING bit set in mode) to reserve a spot * to match a waiting node. */ /* Modes for SNodes, ORed together in node fields */ /** Node represents an unfulfilled consumer */ // 表示消費數據的消費者 static final int REQUEST = 0; /** Node represents an unfulfilled producer */ // 表示生產數據的生產者 static final int DATA = 1; /** Node is fulfilling another unfulfilled DATA or REQUEST */ // 表示匹配另外一個生產者或消費者 static final int FULFILLING = 2; /** The head (top) of the stack */ // 頭結點 volatile SNode head; }
說明：TransferStack有三種不一樣的狀態，REQUEST，表示消費數據的消費者；DATA，表示生產數據的生產者；FULFILLING，表示匹配另外一個生產者或消費者。任何線程對TransferStack的操做都屬於上述3種狀態中的一種。同時還包含一個head域，表示頭結點。
3. 類的內部類
SNode類
1. 類的屬性
static final class SNode { // 下一個結點 volatile SNode next; // next node in stack // 相匹配的結點 volatile SNode match; // the node matched to this // 等待的線程 volatile Thread waiter; // to control park/unpark // 元素項 Object item; // data; or null for REQUESTs // 模式 int mode; // Note: item and mode fields don't need to be volatile // since they are always written before, and read after, // other volatile/atomic operations. // item域和mode域不須要使用volatile修飾，由於它們在volatile/atomic操做以前寫，以後讀 // Unsafe mechanics // 反射機制 private static final sun.misc.Unsafe UNSAFE; // match域的內存偏移地址 private static final long matchOffset; // next域的偏移地址 private static final long nextOffset; static { try { UNSAFE = sun.misc.Unsafe.getUnsafe(); Class<?> k = SNode.class; matchOffset = UNSAFE.objectFieldOffset (k.getDeclaredField("match")); nextOffset = UNSAFE.objectFieldOffset (k.getDeclaredField("next")); } catch (Exception e) { throw new Error(e); } } }
說明：SNode類表示棧中的結點，使用了反射機制和CAS來保證原子性的改變相應的域值。
2. 類的構造函數
SNode(Object item) { this.item = item; }
說明：該構造函數僅僅設置了SNode的item域，其餘域爲默認值。
3. 核心函數分析
3.1. tryMatch函數
boolean tryMatch(SNode s) { if (match == null && UNSAFE.compareAndSwapObject(this, matchOffset, null, s)) { // 本結點的match域爲null而且比較並替換match域成功 // 獲取本節點的等待線程 Thread w = waiter; if (w != null) { // 存在等待的線程 // waiters need at most one unpark // 將本結點的等待線程從新置爲null waiter = null; // unpark等待線程 LockSupport.unpark(w); } return true; } // 若是match不爲null或者CAS設置失敗，則比較match域是否等於s結點，若相等，則表示已經完成匹配，匹配成功 return match == s; }
說明：將s結點與本結點進行匹配，匹配成功，則unpark等待線程。具體流程以下
① 判斷本結點的match域是否爲null，若爲null，則進入步驟②，不然，進入步驟⑤
② CAS設置本結點的match域爲s結點，若成功，則進入步驟③，不然，進入步驟⑤
③ 判斷本結點的waiter域是否爲null，若不爲null，則進入步驟④，不然，進入步驟⑤
④ 從新設置本結點的waiter域爲null，而且unparkwaiter域所表明的等待線程。進入步驟⑥
⑤ 比較本結點的match域是否爲本結點，如果，則進入步驟⑥，不然，進入步驟⑦
⑥ 返回true
⑦ 返回false
4. 核心函數分析
4.1 isFulfilling函數
static boolean isFulfilling(int m) { return (m & FULFILLING) != 0; }
說明：表示是否包含FULFILLING標記。
4.2 transfer函數
E transfer(E e, boolean timed, long nanos) { /* * Basic algorithm is to loop trying one of three actions: * * 1. If apparently empty or already containing nodes of same * mode, try to push node on stack and wait for a match, * returning it, or null if cancelled. * * 2. If apparently containing node of complementary mode, * try to push a fulfilling node on to stack, match * with corresponding waiting node, pop both from * stack, and return matched item. The matching or * unlinking might not actually be necessary because of * other threads performing action 3: * * 3. If top of stack already holds another fulfilling node, * help it out by doing its match and/or pop * operations, and then continue. The code for helping * is essentially the same as for fulfilling, except * that it doesn't return the item. */ SNode s = null; // constructed/reused as needed // 根據e肯定這次轉移的模式（是put or take） int mode = (e == null) ? REQUEST : DATA; for (;;) { // 無限循環 // 保存頭結點 SNode h = head; if (h == null || h.mode == mode) { // 頭結點爲null或者頭結點的模式與這次轉移的模式相同 // empty or same-mode if (timed && nanos <= 0) { // 設置了timed而且等待時間小於等於0，表示不能等待，須要當即操做 // can't wait if (h != null && h.isCancelled()) // 頭結點不爲null而且頭結點被取消 casHead(h, h.next); // 從新設置頭結點（彈出以前的頭結點） // pop cancelled node else // 頭結點爲null或者頭結點沒有被取消 // 返回null return null; } else if (casHead(h, s = snode(s, e, h, mode))) { // 生成一個SNode結點；將原來的head頭結點設置爲該結點的next結點；將head頭結點設置爲該結點 // Spins/blocks until node s is matched by a fulfill operation. // 空旋或者阻塞直到s結點被FulFill操做所匹配 SNode m = awaitFulfill(s, timed, nanos); if (m == s) { // 匹配的結點爲s結點（s結點被取消） // wait was cancelled // 清理s結點 clean(s); // 返回 return null; } if ((h = head) != null && h.next == s) // h從新賦值爲head頭結點，而且不爲null；頭結點的next域爲s結點，表示有結點插入到s結點以前，完成了匹配 // 比較並替換head域（移除插入在s以前的結點和s結點） casHead(h, s.next); // help s's fulfiller // 根據這次轉移的類型返回元素 return (E) ((mode == REQUEST) ? m.item : s.item); } } else if (!isFulfilling(h.mode)) { // 沒有FULFILLING標記，嘗試匹配 // try to fulfill if (h.isCancelled()) // 被取消 // already cancelled // 比較並替換head域（彈出頭結點） casHead(h, h.next); // pop and retry else if (casHead(h, s=snode(s, e, h, FULFILLING|mode))) { // 生成一個SNode結點；將原來的head頭結點設置爲該結點的next結點；將head頭結點設置爲該結點 for (;;) { // 無限循環 // loop until matched or waiters disappear // 保存s的next結點 SNode m = s.next; // m is s's match if (m == null) { // next域爲null // all waiters are gone // 比較並替換head域 casHead(s, null); // pop fulfill node // 賦值s爲null s = null; // use new node next time break; // restart main loop } // m結點的next域 SNode mn = m.next; if (m.tryMatch(s)) { // 嘗試匹配，而且成功 // 比較並替換head域（彈出s結點和m結點） casHead(s, mn); // pop both s and m // 根據這次轉移的類型返回元素 return (E) ((mode == REQUEST) ? m.item : s.item); } else // 匹配不成功 // lost match // 比較並替換next域（彈出m結點） s.casNext(m, mn); // help unlink } } } else { // 頭結點正在匹配 // help a fulfiller // 保存頭結點的next域 SNode m = h.next; // m與h能夠匹配 // m is h's match if (m == null) // next域爲null // waiter is gone // 比較並替換head域（m被其餘結點匹配了，須要彈出h） casHead(h, null); // pop fulfilling node else { // next域不爲null // 獲取m結點的next域 SNode mn = m.next; if (m.tryMatch(h)) // m與h匹配成功 // help match // 比較並替換head域（彈出h和m結點） casHead(h, mn); // pop both h and m else // 匹配不成功 // lost match // 比較並替換next域（移除m結點） h.casNext(m, mn); // help unlink } } } }
說明：此函數用於生產或者消費一個元素，而且transfer函數調用了awaitFulfill函數，以後會經過一個例子給出流程。
4.3 awaitFulfill函數
SNode awaitFulfill(SNode s, boolean timed, long nanos) { /* * When a node/thread is about to block, it sets its waiter * field and then rechecks state at least one more time * before actually parking, thus covering race vs * fulfiller noticing that waiter is non-null so should be * woken. * * When invoked by nodes that appear at the point of call * to be at the head of the stack, calls to park are * preceded by spins to avoid blocking when producers and * consumers are arriving very close in time. This can * happen enough to bother only on multiprocessors. * * The order of checks for returning out of main loop * reflects fact that interrupts have precedence over * normal returns, which have precedence over * timeouts. (So, on timeout, one last check for match is * done before giving up.) Except that calls from untimed * SynchronousQueue.{poll/offer} don't check interrupts * and don't wait at all, so are trapped in transfer * method rather than calling awaitFulfill. */ // 根據timed標識計算截止時間 final long deadline = timed ? System.nanoTime() + nanos : 0L; // 獲取當前線程 Thread w = Thread.currentThread(); // 根據s肯定空旋等待的時間 int spins = (shouldSpin(s) ? (timed ? maxTimedSpins : maxUntimedSpins) : 0); for (;;) { // 無限循環，確保操做成功 if (w.isInterrupted()) // 當前線程被中斷 // 取消s結點 s.tryCancel(); // 獲取s結點的match域 SNode m = s.match; if (m != null) // m不爲null，存在匹配結點 // 返回m結點 return m; if (timed) { // 設置了timed // 肯定繼續等待的時間 nanos = deadline - System.nanoTime(); if (nanos <= 0L) { // 繼續等待的時間小於等於0，等待超時 // 取消s結點 s.tryCancel(); // 跳事後面的部分，繼續 continue; } } if (spins > 0) // 空旋等待的時間大於0 // 確實是否還須要繼續空旋等待 spins = shouldSpin(s) ? (spins-1) : 0; else if (s.waiter == null) // 等待線程爲null // 設置waiter線程爲當前線程 s.waiter = w; // establish waiter so can park next iter else if (!timed) // 沒有設置timed標識 // 禁用當前線程並設置了阻塞者 LockSupport.park(this); else if (nanos > spinForTimeoutThreshold) // 繼續等待的時間大於閾值 // 禁用當前線程，最多等待指定的等待時間，除非許可可用 LockSupport.parkNanos(this, nanos); } }
說明：此函數表示當前線程自旋或阻塞，直到結點被匹配。awaitFulfill函數調用了shouldSpin函數
4.4 shouldSpin函數
boolean shouldSpin(SNode s) { // 獲取頭結點 SNode h = head; // s爲頭結點或者頭結點爲null或者h包含FULFILLING標記,返回true return (h == s || h == null || isFulfilling(h.mode)); }
說明：此函數表示是當前結點所包含的線程（當前線程）進行空旋等待，有以下狀況須要進行空旋等待
① 當前結點爲頭結點
② 頭結點爲null
③ 頭結點正在匹配中
4.5 clean函數
void clean(SNode s) { // s結點的item設置爲null s.item = null; // forget item // waiter域設置爲null s.waiter = null; // forget thread /* * At worst we may need to traverse entire stack to unlink * s. If there are multiple concurrent calls to clean, we * might not see s if another thread has already removed * it. But we can stop when we see any node known to * follow s. We use s.next unless it too is cancelled, in * which case we try the node one past. We don't check any * further because we don't want to doubly traverse just to * find sentinel. */ // 獲取s結點的next域 SNode past = s.next; if (past != null && past.isCancelled()) // next域不爲null而且next域被取消 // 從新設置past past = past.next; // Absorb cancelled nodes at head SNode p; while ((p = head) != null && p != past && p.isCancelled()) // 從棧頂頭結點開始到past結點（不包括），將連續的取消結點移除 // 比較並替換head域（彈出取消的結點） casHead(p, p.next); // Unsplice embedded nodes while (p != null && p != past) { // 移除上一步驟沒有移除的非連續的取消結點 // 獲取p的next域 SNode n = p.next; if (n != null && n.isCancelled()) // n不爲null而且n被取消 // 比較並替換next域 p.casNext(n, n.next); else // 設置p爲n p = n; } }
說明：此函數用於移除從棧頂頭結點開始到該結點（不包括）之間的全部已取消結點。
③ TransferQueue
1. 類的繼承關係
static final class TransferQueue<E> extends Transferer<E> {}
說明：TransferQueue繼承Transferer抽象類，其實現了transfer方法。
2. 類的屬性
static final class TransferQueue<E> extends Transferer<E> { /* * This extends Scherer-Scott dual queue algorithm, differing, * among other ways, by using modes within nodes rather than * marked pointers. The algorithm is a little simpler than * that for stacks because fulfillers do not need explicit * nodes, and matching is done by CAS'ing QNode.item field * from non-null to null (for put) or vice versa (for take). */ /** Head of queue */ // 隊列的頭結點 transient volatile QNode head; /** Tail of queue */ // 隊列的尾結點 transient volatile QNode tail; /** * Reference to a cancelled node that might not yet have been * unlinked from queue because it was the last inserted node * when it was cancelled. */ // 指向一個取消的結點，當一個結點是最後插入隊列時，當被取消時，它可能尚未離開隊列 transient volatile QNode cleanMe; }
說明：隊列存在一個頭結點和一個尾節點，分別指示隊頭和隊尾，還包含了一個指示取消結點的域。
3. 類的內部類
QNode類
QNode的源碼以下
static final class QNode { // 下一個結點 volatile QNode next; // next node in queue // 元素項 volatile Object item; // CAS'ed to or from null // 等待線程 volatile Thread waiter; // to control park/unpark // 是否爲數據 final boolean isData; // 構造函數 QNode(Object item, boolean isData) { // 初始化item域 this.item = item; // 初始化isData域 this.isData = isData; } // 比較並替換next域 boolean casNext(QNode cmp, QNode val) { return next == cmp && UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val); } // 比較並替換item域 boolean casItem(Object cmp, Object val) { return item == cmp && UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val); } /** * Tries to cancel by CAS'ing ref to this as item. */ // 取消本結點，將item域設置爲自身 void tryCancel(Object cmp) { UNSAFE.compareAndSwapObject(this, itemOffset, cmp, this); } // 是否被取消 boolean isCancelled() { // item域是否等於自身 return item == this; } /** * Returns true if this node is known to be off the queue * because its next pointer has been forgotten due to * an advanceHead operation. */ // 是否不在隊列中 boolean isOffList() { // next與是否等於自身 return next == this; } // Unsafe mechanics // 反射機制 private static final sun.misc.Unsafe UNSAFE; // item域的偏移地址 private static final long itemOffset; // next域的偏移地址 private static final long nextOffset; static { try { UNSAFE = sun.misc.Unsafe.getUnsafe(); Class<?> k = QNode.class; itemOffset = UNSAFE.objectFieldOffset (k.getDeclaredField("item")); nextOffset = UNSAFE.objectFieldOffset (k.getDeclaredField("next")); } catch (Exception e) { throw new Error(e); } } }
說明：QNode表示隊列中的結點，而且經過反射和CAS原子性的修改對應的域值。
4. 類的構造函數
TransferQueue() { // 初始化一個哨兵結點 QNode h = new QNode(null, false); // initialize to dummy node. // 設置頭結點 head = h; // 設置尾結點 tail = h; }
說明：該構造函數用於初始化一個隊列，而且初始化了一個哨兵結點，頭結點與尾節點均指向該哨兵結點。
5. 核心函數分析
5.1 transfer函數
E transfer(E e, boolean timed, long nanos) { /* Basic algorithm is to loop trying to take either of * two actions: * * 1. If queue apparently empty or holding same-mode nodes, * try to add node to queue of waiters, wait to be * fulfilled (or cancelled) and return matching item. * * 2. If queue apparently contains waiting items, and this * call is of complementary mode, try to fulfill by CAS'ing * item field of waiting node and dequeuing it, and then * returning matching item. * * In each case, along the way, check for and try to help * advance head and tail on behalf of other stalled/slow * threads. * * The loop starts off with a null check guarding against * seeing uninitialized head or tail values. This never * happens in current SynchronousQueue, but could if * callers held non-volatile/final ref to the * transferer. The check is here anyway because it places * null checks at top of loop, which is usually faster * than having them implicitly interspersed. */ QNode s = null; // constructed/reused as needed // 肯定這次轉移的類型（put or take） boolean isData = (e != null); for (;;) { // 無限循環，確保操做成功 // 獲取尾結點 QNode t = tail; // 獲取頭結點 QNode h = head; if (t == null || h == null) // 看到未初始化的頭尾結點 // saw uninitialized value // 跳事後面的部分，繼續 continue; // spin if (h == t || t.isData == isData) { // 頭結點與尾結點相等或者尾結點的模式與當前結點模式相同 // empty or same-mode // 獲取尾結點的next域 QNode tn = t.next; if (t != tail) // t不爲尾結點，不一致，重試 // inconsistent read continue; if (tn != null) { // tn不爲null，有其餘線程添加了tn結點 // lagging tail // 設置新的尾結點爲tn advanceTail(t, tn); // 跳事後面的部分，繼續 continue; } if (timed && nanos <= 0) // 設置了timed而且等待時間小於等於0，表示不能等待，須要當即操做 // can't wait // 返回null return null; if (s == null) // s爲null // 新生一個結點並賦值給s s = new QNode(e, isData); if (!t.casNext(null, s)) // 設置t結點的next域不成功 // failed to link in // 跳事後面的部分，繼續 continue; // 設置新的尾結點 advanceTail(t, s); // swing tail and wait // Spins/blocks until node s is fulfilled // 空旋或者阻塞直到s結點被匹配 Object x = awaitFulfill(s, e, timed, nanos); if (x == s) { // x與s相等，表示已經取消 // wait was cancelled // 清除 clean(t, s); // 返回null return null; } if (!s.isOffList()) { // s結點還沒離開隊列 // not already unlinked // 設置新的頭結點 advanceHead(t, s); // unlink if head if (x != null) // x不爲null // and forget fields // 設置s結點的item s.item = s; // 設置s結點的waiter域爲null s.waiter = null; } return (x != null) ? (E)x : e; } else { // 模式互補 // complementary-mode // 獲取頭結點的next域（匹配的結點） QNode m = h.next; // node to fulfill if (t != tail || m == null || h != head) // t不爲尾結點或者m爲null或者h不爲頭結點（不一致） // 跳事後面的部分，繼續 continue; // inconsistent read // 獲取m結點的元素域 Object x = m.item; if (isData == (x != null) || // m結點被匹配 // m already fulfilled x == m || // m結點被取消 // m cancelled !m.casItem(x, e)) { // CAS操做失敗 // lost CAS advanceHead(h, m); // 隊列頭結點出隊列，並重試 // dequeue and retry continue; } // 匹配成功，設置新的頭結點 advanceHead(h, m); // successfully fulfilled // unpark m結點對應的等待線程 LockSupport.unpark(m.waiter); return (x != null) ? (E)x : e; } } }
說明：此函數用於生產或者消費一個元素，而且transfer函數調用了awaitFulfill函數，以後會經過一個例子給出流程。
5.2 awaitFulfill函數
Object awaitFulfill(QNode s, E e, boolean timed, long nanos) { /* Same idea as TransferStack.awaitFulfill */ // 根據timed標識計算截止時間 final long deadline = timed ? System.nanoTime() + nanos : 0L; // 獲取當前線程 Thread w = Thread.currentThread(); // 計算空旋時間 int spins = ((head.next == s) ? (timed ? maxTimedSpins : maxUntimedSpins) : 0); for (;;) { // 無限循環，確保操做成功 if (w.isInterrupted()) // 當前線程被中斷 // 取消 s.tryCancel(e); // 獲取s的元素域 Object x = s.item; if (x != e) // 元素不爲e // 返回 return x; if (timed) { // 設置了timed // 計算繼續等待的時間 nanos = deadline - System.nanoTime(); if (nanos <= 0L) { // 繼續等待的時間小於等於0 // 取消 s.tryCancel(e); // 跳事後面的部分，繼續 continue; } } if (spins > 0) // 空旋時間大於0 // 減小空旋時間 --spins; else if (s.waiter == null) // 等待線程爲null // 設置等待線程 s.waiter = w; else if (!timed) // 沒有設置timed標識 // 禁用當前線程並設置了阻塞者 LockSupport.park(this); else if (nanos > spinForTimeoutThreshold) // 繼續等待的時間大於閾值 // 禁用當前線程，最多等待指定的等待時間，除非許可可用 LockSupport.parkNanos(this, nanos); } }
說明：此函數表示當前線程自旋或阻塞，直到結點被匹配。
5.3 clean函數
void clean(QNode pred, QNode s) { // 設置等待線程爲null s.waiter = null; // forget thread /* * At any given time, exactly one node on list cannot be * deleted -- the last inserted node. To accommodate this, * if we cannot delete s, we save its predecessor as * "cleanMe", deleting the previously saved version * first. At least one of node s or the node previously * saved can always be deleted, so this always terminates. */ /* * 在任什麼時候候，最後插入的結點不能刪除，爲了知足這個條件 * 若是不能刪除s結點，咱們將s結點的前驅設置爲cleanMe結點 * 刪除以前保存的版本，至少s結點或者以前保存的結點可以被刪除 * 因此最後老是會結束 */ while (pred.next == s) { // pred的next域爲s // Return early if already unlinked // 獲取頭結點 QNode h = head; // 獲取頭結點的next域 QNode hn = h.next; // Absorb cancelled first node as head if (hn != null && hn.isCancelled()) { // hn不爲null而且hn被取消 // 設置新的頭結點 advanceHead(h, hn); // 跳事後面的部分，繼續 continue; } // 獲取尾結點，保證對尾結點的讀一致性 QNode t = tail; // Ensure consistent read for tail if (t == h) // 尾結點爲頭結點，表示隊列爲空 // 返回 return; // 獲取尾結點的next域 QNode tn = t.next; if (t != tail) // t不爲尾結點，不一致，重試 // 跳事後面的部分，繼續 continue; if (tn != null) { // tn不爲null // 設置新的尾結點 advanceTail(t, tn); // 跳事後面的部分，繼續 continue; } if (s != t) { // s不爲尾結點，移除s // If not tail, try to unsplice QNode sn = s.next; if (sn == s || pred.casNext(s, sn)) // return; } // 獲取cleanMe結點 QNode dp = cleanMe; if (dp != null) { // dp不爲null，斷開前面被取消的結點 // Try unlinking previous cancelled node // 獲取dp的next域 QNode d = dp.next; QNode dn; if (d == null || // d is gone or d == dp || // d is off list or !d.isCancelled() || // d not cancelled or (d != t && // d not tail and (dn = d.next) != null && // has successor dn != d && // that is on list dp.casNext(d, dn))) // d unspliced casCleanMe(dp, null); if (dp == pred) return; // s is already saved node } else if (casCleanMe(null, pred)) return; // Postpone cleaning s } }
說明：此函數用於移除已經被取消的結點。
3.3 類的屬性
public class SynchronousQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable { // 版本序列號 private static final long serialVersionUID = -3223113410248163686L; // 可用的處理器 static final int NCPUS = Runtime.getRuntime().availableProcessors(); // 最大空旋時間 static final int maxTimedSpins = (NCPUS < 2) ? 0 : 32; // 無限時的等待的最大空旋時間 static final int maxUntimedSpins = maxTimedSpins * 16; // 超時空旋等待閾值 static final long spinForTimeoutThreshold = 1000L; // 用於序列化 private ReentrantLock qlock; private WaitQueue waitingProducers; private WaitQueue waitingConsumers; }
說明：SynchronousQueue類的屬性包含了空旋等待時間相關的屬性。
3.4 類的構造函數
1. SynchronousQueue()型構造函數
public SynchronousQueue() { // 非公平策略（先進後出） this(false); }
說明：該構造函數用於建立一個具備非公平訪問策略的 SynchronousQueue。
2. SynchronousQueue(boolean)型構造函數
public SynchronousQueue(boolean fair) { // 根據指定的策略生成不一樣的結構 transferer = fair ? new TransferQueue<E>() : new TransferStack<E>(); }
說明：建立一個具備指定公平策略的 SynchronousQueue。
3.5 核心函數分析
在分析了TransferStack和TransferQueue的相關函數後，SynchronousQueue的函數的分析就很是簡單。
// 將指定元素添加到此隊列，若有必要則等待另外一個線程接收它 public void put(E e) throws InterruptedException { // e爲null則拋出異常 if (e == null) throw new NullPointerException(); if (transferer.transfer(e, false, 0) == null) { // 進行轉移操做 // 中斷當前線程 Thread.interrupted(); throw new InterruptedException(); } } // 將指定元素插入到此隊列，若有必要則等待指定的時間，以便另外一個線程接收它 public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException { // e爲null則拋出異常 if (e == null) throw new NullPointerException(); if (transferer.transfer(e, true, unit.toNanos(timeout)) != null) // 進行轉移操做 return true; if (!Thread.interrupted()) // 當前線程沒有被中斷 // 返回 return false; throw new InterruptedException(); } // 若是另外一個線程正在等待以便接收指定元素，則將指定元素插入到此隊列 public boolean offer(E e) { // e爲null則拋出異常 if (e == null) throw new NullPointerException(); return transferer.transfer(e, true, 0) != null; // 進行轉移操做 } // 獲取並移除此隊列的頭，若有必要則等待另外一個線程插入它 public E take() throws InterruptedException { // 進行轉移操做 E e = transferer.transfer(null, false, 0); if (e != null) return e; Thread.interrupted(); throw new InterruptedException(); } // 獲取並移除此隊列的頭，若有必要則等待指定的時間，以便另外一個線程插入它 public E poll(long timeout, TimeUnit unit) throws InterruptedException { E e = transferer.transfer(null, true, unit.toNanos(timeout)); if (e != null || !Thread.interrupted()) // 元素不爲null或者當前線程沒有被中斷 return e; throw new InterruptedException(); } // 若是另外一個線程當前正要使用某個元素，則獲取並移除此隊列的頭 public E poll() { return transferer.transfer(null, true, 0); } // 始終返回 true public boolean isEmpty() { return true; } // 始終返回 0 public int size() { return 0; } // 始終返回 0 public int remainingCapacity() { return 0; } // 不執行任何操做 public void clear() { } // 始終返回false public boolean contains(Object o) { return false; } // 始終返回false public boolean remove(Object o) { return false; } // 除非給定 collection 爲空，不然返回 false public boolean containsAll(Collection<?> c) { return c.isEmpty(); } // 始終返回 false public boolean removeAll(Collection<?> c) { return false; } // 始終返回 false public boolean retainAll(Collection<?> c) { return false; } // 始終返回 null public E peek() { return null; } // 返回一個空迭代器，其中 hasNext 始終返回 false public Iterator<E> iterator() { return Collections.emptyIterator(); } // public Spliterator<E> spliterator() { return Spliterators.emptySpliterator(); } // 返回一個 0 長度的數組 public Object[] toArray() { return new Object[0]; } // 將指定數組的第 0 個元素設置爲 null（若是該數組有非 0 的長度）並返回它 public <T> T[] toArray(T[] a) { if (a.length > 0) a[0] = null; return a; } // 移除此隊列中全部可用的元素，並將它們添加到給定 collection 中 public int drainTo(Collection<? super E> c) { if (c == null) throw new NullPointerException(); if (c == this) throw new IllegalArgumentException(); int n = 0; for (E e; (e = poll()) != null;) { c.add(e); ++n; } return n; } // 最多今後隊列中移除給定數量的可用元素，並將這些元素添加到給定 collection 中 public int drainTo(Collection<? super E> c, int maxElements) { if (c == null) throw new NullPointerException(); if (c == this) throw new IllegalArgumentException(); int n = 0; for (E e; n < maxElements && (e = poll()) != null;) { c.add(e); ++n; } return n; }
說明：SynchronousQueue的函數很大程度都是依託於TransferStack或TransferQueue的transfer函數，因此，瞭解transfer函數就能夠了解SynchronousQueue的原理。
4、示例
下面經過一個示例來詳細瞭解SynchronousQueue的使用。
package com.hust.grid.leesf.collections; import java.util.concurrent.SynchronousQueue; import java.util.concurrent.TimeUnit; public class SynchronousQueueDemo { public static void main(String[] args) { SynchronousQueue<Integer> queue = new SynchronousQueue<Integer>(); Producer p1 = new Producer("p1", queue, 10); Producer p2 = new Producer("p2", queue, 50); Consumer c1 = new Consumer("c1", queue); Consumer c2 = new Consumer("c2", queue); c1.start(); try { TimeUnit.MILLISECONDS.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } c2.start(); try { TimeUnit.MILLISECONDS.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } p1.start(); try { TimeUnit.MILLISECONDS.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } p2.start(); } static class Producer extends Thread { private SynchronousQueue<Integer> queue; private int n; public Producer(String name, SynchronousQueue<Integer> queue, int n) { super(name); this.queue = queue; this.n = n; } public void run() { System.out.println(getName() + " offer result " + queue.offer(n)); } } static class Consumer extends Thread { private SynchronousQueue<Integer> queue; public Consumer(String name, SynchronousQueue<Integer> queue) { super(name); this.queue = queue; } public void run() { try { System.out.println(getName() + " take result " + queue.take()); } catch (InterruptedException e) { e.printStackTrace(); } } } }
運行結果（某一次）
p1 offer result true c2 take result 10 p2 offer result true c1 take result 50
說明：該示例中，有兩個生產者p一、p2和兩個消費者c一、c2，按照c一、c二、p一、p2的順序啓動，而且每一個線程啓動後休眠100ms，則可能有以下的時序圖
說明：時序圖中，c1線程的take操做早於c2線程的take操做早於p1線程的offer操做早於p2線程的offer操做。
根據示例源碼可知，此SynchronousQueue採用非公平策略，即底層採用棧結構。
① c1執行take操做，主要的函數調用以下
說明：其中，c1線程進入awaitFulfill後，會空旋等待，直到空旋時間消逝，會調用LockSupport.park函數，會禁用當前線程（c1），直至許可可用。
② c1執行take操做，主要的函數調用以下
說明：其中，c2線程進入awaitFulfill後，會空旋等待，直到空旋時間消逝，會調用LockSupport.park函數，會禁用當前線程（c2），直至許可可用。而且此時棧中有兩個節點，c2線程所在的結點和c1線程所在的結點。
③ p1線程執行offer(10)操做，主要的函數調用以下
說明：在執行offer(10)操做後，c2線程所在的結點與頭結點進行了匹配（頭結點生產數據，c2線程所在的結點消費數據），c2線程被unpark，能夠繼續運行，而c1線程仍是被park中（非公平策略）。
③ c2線程被unpark後，繼續運行，主要函數調用以下（因爲c2線程是在awaitFulfill函數中被park的，因此，恢復也是在awaitFulfill函數中）
說明：c2線程從unpark恢復時，結構如上圖所示，先從awaitFulfill函數中返回，而後再從transfer函數中返回10，再從take函數中返回10。
④ p2線程執行offer(50)操做，主要的函數調用以下
說明：在執行offer(50)操做後，c1線程所在的結點與頭結點進行了匹配（頭結點生產數據，c1線程所在的結點消費數據），c1線程被unpark，能夠繼續運行。
⑤ c1線程被unpark後，繼續運行，主要函數調用以下（因爲c1線程是在awaitFulfill函數中被park的，因此，恢復也是在awaitFulfill函數中）
說明：c1線程從unpark恢復時，結構如上圖所示，先從awaitFulfill函數中返回，而後再從transfer函數中返回50，再從take函數中返回50。
上述是使用非公平策略的結果（首先匹配c2線程所在的結點，以後再匹配c1線程所在結點）。
修改示例，改用公平策略。
package com.hust.grid.leesf.collections; import java.util.concurrent.SynchronousQueue; import java.util.concurrent.TimeUnit; public class SynchronousQueueDemo { public static void main(String[] args) { SynchronousQueue<Integer> queue = new SynchronousQueue<Integer>(true); Producer p1 = new Producer("p1", queue, 10); Producer p2 = new Producer("p2", queue, 50); Consumer c1 = new Consumer("c1", queue); Consumer c2 = new Consumer("c2", queue); c1.start(); try { TimeUnit.MILLISECONDS.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } c2.start(); try { TimeUnit.MILLISECONDS.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } p1.start(); try { TimeUnit.MILLISECONDS.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } p2.start(); } static class Producer extends Thread { private SynchronousQueue<Integer> queue; private int n; public Producer(String name, SynchronousQueue<Integer> queue, int n) { super(name); this.queue = queue; this.n = n; } public void run() { System.out.println(getName() + " offer result " + queue.offer(n)); } } static class Consumer extends Thread { private SynchronousQueue<Integer> queue; public Consumer(String name, SynchronousQueue<Integer> queue) { super(name); this.queue = queue; } public void run() { try { System.out.println(getName() + " take result " + queue.take()); } catch (InterruptedException e) { e.printStackTrace(); } } } }
運行結果（某一次）
p1 offer result true c1 take result 10 p2 offer result true c2 take result 50
說明：從運行結果可知，c1線程會比c2線程先匹配（由於採用公平策略，先入隊列先匹配，因此c1先獲得匹配，而後再匹配c2）。具體的流程圖與非公平策略相似，在此再也不累贅。
當再次修改源碼，仍是使用非公平策略，只是改變c一、c二、p一、p2之間的啓動順序。更改成p1->c1->p2->c2。
package com.hust.grid.leesf.collections; import java.util.concurrent.SynchronousQueue; import java.util.concurrent.TimeUnit; public class SynchronousQueueDemo { public static void main(String[] args) { SynchronousQueue<Integer> queue = new SynchronousQueue<Integer>(); Producer p1 = new Producer("p1", queue, 10); Producer p2 = new Producer("p2", queue, 50); Consumer c1 = new Consumer("c1", queue); Consumer c2 = new Consumer("c2", queue); p1.start(); try { TimeUnit.MILLISECONDS.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } c1.start(); try { TimeUnit.MILLISECONDS.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } p2.start(); try { TimeUnit.MILLISECONDS.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } c2.start(); } static class Producer extends Thread { private SynchronousQueue<Integer> queue; private int n; public Producer(String name, SynchronousQueue<Integer> queue, int n) { super(name); this.queue = queue; this.n = n; } public void run() { System.out.println(getName() + " offer result " + queue.offer(n)); } } static class Consumer extends Thread { private SynchronousQueue<Integer> queue; public Consumer(String name, SynchronousQueue<Integer> queue) { super(name); this.queue = queue; } public void run() { try { System.out.println(getName() + " take result " + queue.take()); } catch (InterruptedException e) { e.printStackTrace(); } } } }
運行結果（某一次）
p1 offer result false p2 offer result true c1 take result 50
說明：此時，只有c1線程獲得了匹配，p1線程存放元素，直接返回的false，由於此時沒有消費者線程等待，而p2線程與c1線程進行了匹配，p2線程存放元素成功，c1線程獲取元素成功，而且此時，c2線程仍是處於park狀態，此時應用程序沒法正常結束。因此，可知，必需要先有取操做，而後存操做，二者才能正確的匹配，若先是存操做，而後再是取操做，此時沒法匹配成功，會阻塞，取操做期待下一個存操做進行匹配。
5、總結
SynchronousQueue的源碼就分析到這裏，SynchronousQueue適合一對一的匹配場景，沒有容量，沒法緩存。有了這個基礎，以後會方便分析線程池框架的源碼，謝謝各位園友的觀看~
相關文章
- 1. JUC源碼分析-集合篇（九）SynchronousQueue
- 2. 【JUC】JDK1.8源碼分析之LockSupport（一）
- 3. 【JUC】JDK1.8源碼分析之CountDownLatch（五）
- 4. 【JUC】JDK1.8源碼分析之ReentrantReadWriteLock（七）
- 5. 【JUC】JDK1.8源碼分析之CountDownLatc
- 6. 【JUC】JDK1.8源碼分析之LinkedBlockingQueue（四）
- 7. 【JUC】JDK1.8源碼分析之AbstractQueuedSynchronizer（二）
- 8. 【JUC】JDK1.8源碼分析之CopyOnWriteArrayList（六）
- 9. JUC--PriorityQueue源碼分析（基於JDK1.8）
- 10. JUC之ReentrantLock源碼分析
- 更多相關文章...
- • Docker 資源彙總 - Docker教程
- • Java操作Neo4j數據庫（附帶源碼） - NoSQL教程
- • 互聯網組織的未來：剖析GitHub員工的任性之源
- • Java Agent入門實戰（二）-Instrumentation源碼概述
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. JUC源碼分析-集合篇（九）SynchronousQueue
- 2. 【JUC】JDK1.8源碼分析之LockSupport（一）
- 3. 【JUC】JDK1.8源碼分析之CountDownLatch（五）
- 4. 【JUC】JDK1.8源碼分析之ReentrantReadWriteLock（七）
- 5. 【JUC】JDK1.8源碼分析之CountDownLatc
- 6. 【JUC】JDK1.8源碼分析之LinkedBlockingQueue（四）
- 7. 【JUC】JDK1.8源碼分析之AbstractQueuedSynchronizer（二）
- 8. 【JUC】JDK1.8源碼分析之CopyOnWriteArrayList（六）
- 9. JUC--PriorityQueue源碼分析（基於JDK1.8）
- 10. JUC之ReentrantLock源碼分析