【JUC】JDK1.8源码分析之SynchronousQueue（九）

太过爱你忘了你带给我的痛 2022-07-15 04:12 182阅读 0赞

http://www.cnblogs.com/leesf456/p/5560362.html

一、前言

本篇是在分析Executors源码时，发现JUC集合框架中的一个重要类没有分析，SynchronousQueue，该类在线程池中的作用是非常明显的，所以很有必要单独拿出来分析一番，这对于之后理解线程池有很有好处，SynchronousQueue是一种阻塞队列，其中每个插入操作必须等待另一个线程的对应移除操作 ，反之亦然。同步队列没有任何内部容量，甚至连一个队列的容量都没有。

二、SynchronousQueue数据结构

由于SynchronousQueue的支持公平策略和非公平策略，所以底层可能两种数据结构：队列（实现公平策略）和栈（实现非公平策略），队列与栈都是通过链表来实现的。具体的数据结构如下

![Image 1][]　　说明：数据结构有两种类型，栈和队列；栈有一个头结点，队列有一个头结点和尾结点；栈用于实现非公平策略，队列用于实现公平策略。

三、SynchronousQueue源码分析

3.1 类的继承关系

public class SynchronousQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {}

说明：SynchronousQueue继承了AbstractQueue抽象类，AbstractQueue定义了对队列的基本操作；同时实现了BlockingQueue接口，BlockingQueue表示阻塞型的队列，其对队列的操作可能会抛出异常；同时也实现了Searializable接口，表示可以被序列化。

3.2 类的内部类

SynchronousQueue的内部类框架图如下

![Image 1][]

说明：其中比较重要的类是左侧的三个类，Transferer是TransferStack栈和TransferQueue队列的公共类，定义了转移数据的公共操作，由TransferStack和TransferQueue具体实现，WaitQueue、LifoWaitQueue、FifoWaitQueue表示为了兼容JDK1.5版本中的SynchronousQueue的序列化策略所遗留的，这里不做具体的讲解。下面着重看左侧的三个类。

① Transferer

![ExpandedBlockStart.gif][]

![复制代码][copycode.gif]

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

![复制代码][copycode.gif]

说明：Transferer定义了transfer操作，用于take或者put数据。transfer方法由子类实现。

② TransfererStack

1. 类的继承关系

static final class TransferStack<E> extends Transferer<E> {}

说明：TransferStack继承Transferer抽象类，其实现了transfer方法。

2. 类的属性

![ExpandedBlockStart.gif][]

![复制代码][copycode.gif]

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

![复制代码][copycode.gif]

说明：TransferStack有三种不同的状态，REQUEST，表示消费数据的消费者；DATA，表示生产数据的生产者；FULFILLING，表示匹配另一个生产者或消费者。任何线程对TransferStack的操作都属于上述3种状态中的一种。同时还包含一个head域，表示头结点。

3. 类的内部类

SNode类

1. 类的属性

![ExpandedBlockStart.gif][]

![复制代码][copycode.gif]

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

![复制代码][copycode.gif]

说明：SNode类表示栈中的结点，使用了反射机制和CAS来保证原子性的改变相应的域值。

2. 类的构造函数

![ExpandedBlockStart.gif][]

SNode(Object item) {
                    this.item = item;
                }

说明：该构造函数仅仅设置了SNode的item域，其他域为默认值。

3. 核心函数分析

3.1. tryMatch函数

![ExpandedBlockStart.gif][]

![复制代码][copycode.gif]

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

![复制代码][copycode.gif]

说明：将s结点与本结点进行匹配，匹配成功，则unpark等待线程。具体流程如下

① 判断本结点的match域是否为null，若为null，则进入步骤②，否则，进入步骤⑤

② CAS设置本结点的match域为s结点，若成功，则进入步骤③，否则，进入步骤⑤

③ 判断本结点的waiter域是否为null，若不为null，则进入步骤④，否则，进入步骤⑤

④ 重新设置本结点的waiter域为null，并且unparkwaiter域所代表的等待线程。进入步骤⑥

⑤ 比较本结点的match域是否为本结点，若是，则进入步骤⑥，否则，进入步骤⑦

⑥ 返回true

⑦ 返回false

4. 核心函数分析

4.1 isFulfilling函数

![ExpandedBlockStart.gif][]

static boolean isFulfilling(int m) { return (m & FULFILLING) != 0; }

说明：表示是否包含FULFILLING标记。

4.2 transfer函数

![ExpandedBlockStart.gif][]

![复制代码][copycode.gif]

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

![复制代码][copycode.gif]

说明：此函数用于生产或者消费一个元素，并且transfer函数调用了awaitFulfill函数，之后会通过一个例子给出流程。

4.3 awaitFulfill函数

![ExpandedBlockStart.gif][]

![复制代码][copycode.gif]

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

![复制代码][copycode.gif]

说明：此函数表示当前线程自旋或阻塞，直到结点被匹配。awaitFulfill函数调用了shouldSpin函数

4.4 shouldSpin函数

![ExpandedBlockStart.gif][]

![复制代码][copycode.gif]

boolean shouldSpin(SNode s) {
                // 获取头结点
                SNode h = head;
                // s为头结点或者头结点为null或者h包含FULFILLING标记,返回true
                return (h == s || h == null || isFulfilling(h.mode)); 
            }

![复制代码][copycode.gif]

说明：此函数表示是当前结点所包含的线程（当前线程）进行空旋等待，有如下情况需要进行空旋等待

① 当前结点为头结点

② 头结点为null

③ 头结点正在匹配中

4.5 clean函数

![ExpandedBlockStart.gif][]

![复制代码][copycode.gif]

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

![复制代码][copycode.gif]

说明：此函数用于移除从栈顶头结点开始到该结点（不包括）之间的所有已取消结点。

③ TransferQueue

1. 类的继承关系

static final class TransferQueue<E> extends Transferer<E> {}

说明：TransferQueue继承Transferer抽象类，其实现了transfer方法。

2. 类的属性

![ExpandedBlockStart.gif][]

![复制代码][copycode.gif]

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

![复制代码][copycode.gif]

说明：队列存在一个头结点和一个尾节点，分别指示队头和队尾，还包含了一个指示取消结点的域。

3. 类的内部类

QNode类

QNode的源码如下

![ExpandedBlockStart.gif][]

![复制代码][copycode.gif]

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

![复制代码][copycode.gif]

说明：QNode表示队列中的结点，并且通过反射和CAS原子性的修改对应的域值。

4. 类的构造函数

![ExpandedBlockStart.gif][]

![复制代码][copycode.gif]

TransferQueue() {
                // 初始化一个哨兵结点
                QNode h = new QNode(null, false); // initialize to dummy node.
                // 设置头结点
                head = h;
                // 设置尾结点
                tail = h;
            }

![复制代码][copycode.gif]

说明：该构造函数用于初始化一个队列，并且初始化了一个哨兵结点，头结点与尾节点均指向该哨兵结点。

5. 核心函数分析

5.1 transfer函数

![ExpandedBlockStart.gif][]

![复制代码][copycode.gif]

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

![复制代码][copycode.gif]

说明：此函数用于生产或者消费一个元素，并且transfer函数调用了awaitFulfill函数，之后会通过一个例子给出流程。

5.2 awaitFulfill函数

![ExpandedBlockStart.gif][]

![复制代码][copycode.gif]

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

![复制代码][copycode.gif]

说明：此函数表示当前线程自旋或阻塞，直到结点被匹配。

5.3 clean函数

![ExpandedBlockStart.gif][]

![复制代码][copycode.gif]

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

![复制代码][copycode.gif]

说明：此函数用于移除已经被取消的结点。

3.3 类的属性

![ExpandedBlockStart.gif][]

![复制代码][copycode.gif]

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

![复制代码][copycode.gif]

说明：SynchronousQueue类的属性包含了空旋等待时间相关的属性。

3.4 类的构造函数

1. SynchronousQueue()型构造函数

![ExpandedBlockStart.gif][]

public SynchronousQueue() {
            // 非公平策略（先进后出）
            this(false);
        }

说明：该构造函数用于创建一个具有非公平访问策略的 SynchronousQueue。

2. SynchronousQueue(boolean)型构造函数

![ExpandedBlockStart.gif][]

public SynchronousQueue(boolean fair) {
            // 根据指定的策略生成不同的结构
            transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
        }

说明：创建一个具有指定公平策略的 SynchronousQueue。

3.5 核心函数分析

在分析了TransferStack和TransferQueue的相关函数后，SynchronousQueue的函数的分析就非常简单。

![ExpandedBlockStart.gif][]

![复制代码][copycode.gif]

// 将指定元素添加到此队列，如有必要则等待另一个线程接收它
        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;
        }

![复制代码][copycode.gif]

说明：SynchronousQueue的函数很大程度都是依托于TransferStack或TransferQueue的transfer函数，所以，了解transfer函数就可以了解SynchronousQueue的原理。

四、示例

下面通过一个示例来详细了解SynchronousQueue的使用。

![ExpandedBlockStart.gif][]

![复制代码][copycode.gif]

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

![复制代码][copycode.gif]

运行结果（某一次）

p1 offer result true
    c2 take result 10
    p2 offer result true
    c1 take result 50

说明：该示例中，有两个生产者p1、p2和两个消费者c1、c2，按照c1、c2、p1、p2的顺序启动，并且每个线程启动后休眠100ms，则可能有如下的时序图

![Image 1][]

说明：时序图中，c1线程的take操作早于c2线程的take操作早于p1线程的offer操作早于p2线程的offer操作。

根据示例源码可知，此SynchronousQueue采用非公平策略，即底层采用栈结构。

① c1执行take操作，主要的函数调用如下

![Image 1][]

说明：其中，c1线程进入awaitFulfill后，会空旋等待，直到空旋时间消逝，会调用LockSupport.park函数，会禁用当前线程（c1），直至许可可用。

② c1执行take操作，主要的函数调用如下

![Image 1][]

说明：其中，c2线程进入awaitFulfill后，会空旋等待，直到空旋时间消逝，会调用LockSupport.park函数，会禁用当前线程（c2），直至许可可用。并且此时栈中有两个节点，c2线程所在的结点和c1线程所在的结点。

③ p1线程执行offer(10)操作，主要的函数调用如下

![Image 1][]

说明：在执行offer(10)操作后，c2线程所在的结点与头结点进行了匹配（头结点生产数据，c2线程所在的结点消费数据），c2线程被unpark，可以继续运行，而c1线程还是被park中（非公平策略）。

③ c2线程被unpark后，继续运行，主要函数调用如下（由于c2线程是在awaitFulfill函数中被park的，所以，恢复也是在awaitFulfill函数中）

![Image 1][]

说明：c2线程从unpark恢复时，结构如上图所示，先从awaitFulfill函数中返回，然后再从transfer函数中返回10，再从take函数中返回10。

④ p2线程执行offer(50)操作，主要的函数调用如下

![Image 1][]

说明：在执行offer(50)操作后，c1线程所在的结点与头结点进行了匹配（头结点生产数据，c1线程所在的结点消费数据），c1线程被unpark，可以继续运行。

⑤ c1线程被unpark后，继续运行，主要函数调用如下（由于c1线程是在awaitFulfill函数中被park的，所以，恢复也是在awaitFulfill函数中）

![Image 1][]

说明：c1线程从unpark恢复时，结构如上图所示，先从awaitFulfill函数中返回，然后再从transfer函数中返回50，再从take函数中返回50。

上述是使用非公平策略的结果（首先匹配c2线程所在的结点，之后再匹配c1线程所在结点）。

修改示例，改用公平策略。

![ExpandedBlockStart.gif][]

![复制代码][copycode.gif]

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

![复制代码][copycode.gif]

运行结果（某一次）

p1 offer result true
    c1 take result 10
    p2 offer result true
    c2 take result 50

说明：从运行结果可知，c1线程会比c2线程先匹配（因为采用公平策略，先入队列先匹配，所以c1先得到匹配，然后再匹配c2）。具体的流程图与非公平策略类似，在此不再累赘。

当再次修改源码，还是使用非公平策略，只是改变c1、c2、p1、p2之间的启动顺序。更改为p1->c1->p2->c2。

![ExpandedBlockStart.gif][]

![复制代码][copycode.gif]

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

![复制代码][copycode.gif]

运行结果（某一次）

p1 offer result false
    p2 offer result true
    c1 take result 50

说明：此时，只有c1线程得到了匹配，p1线程存放元素，直接返回的false，因为此时没有消费者线程等待，而p2线程与c1线程进行了匹配，p2线程存放元素成功，c1线程获取元素成功，并且此时，c2线程还是处于park状态，此时应用程序无法正常结束。所以，可知，必须要先有取操作，然后存操作，两者才能正确的匹配，若先是存操作，然后再是取操作，此时无法匹配成功，会阻塞，取操作期待下一个存操作进行匹配。

五、总结

SynchronousQueue的源码就分析到这里，SynchronousQueue适合一对一的匹配场景，没有容量，无法缓存。有了这个基础，之后会方便分析线程池框架的源码，谢谢各位园友的观看~

[Image 1]: 
[ExpandedBlockStart.gif]: /images/20220715/a8c7a6892ca54474a0ff7fe94a12672c.png
[copycode.gif]: /images/20220715/4e6f67132d224dd7b06fbffd812d62d7.png