更新日志

SynchronousQueue类图

SynchronousQueue的实现细节

SynchronousQueue使用了一个非常关键的方法来转移数据（从生产者线程转移到消费者线程），下面是这个方法：

    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.
         */
        abstract E transfer(E e, boolean timed, long nanos);
    }

这里面需要说明一下的是，这个方法会根据参数e来区分调用方法的是一个生产者线程还是一个消费者线程，如果e为null，则说明这是一个消费者线程，比如一个take操作，如果e不为null，那么就是一个生产者线程，这个数据就是这个线程需要交付的数据，比如一个put操作。SynchronousQueue有两个版本的Transferer实现，一种为公平交易类型，一种为非公平交易类型，公平交易类型的实现类为TransferQueue，它使用队列来作为交易媒介，请求交易的线程总是先尝试跟队列头部（或者尾部）的线程进行交易，如果失败再将请求的线程添加到队列尾部，而非公平类型的实现类为TransferStack，它使用一个stack来作为交易媒介，请求交易的线程总是试图与栈顶线程进行交易，失败则添加到栈顶。所以SynchronousQueue就是使用队列和栈两种数据结构来模拟公平交易和非公平交易的。下面分别对两种交易类型进行分析。

非公平交易实现TransferStack

首先来分析一下公平交易的实现细节。每个请求节点都有一个状态，下面展示了提供的状态：

        /** 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;

REQUEST表示了一个请求交易但是没有得到匹配的消费者，DATA表示一个请求交易但是没有交付数据的生产者，而FULFILLING表示正在进行交易的生产者或者消费者。TransferStack使用了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;

next表示下一个节点，match表示的是与当前节点匹配的节点，而waiter表示该节点上的线程，item为交易的数据，mode表示了该节点的状态。TransferStack使用tryMatch这个方法来进行当前节点和s节点的匹配工作，如果匹配成功，则唤醒等待线程。下面展示了该方法的实现细节：

            boolean tryMatch(SNode s) {
                if (match == null &&
                    UNSAFE.compareAndSwapObject(this, matchOffset, null, s)) {
                    Thread w = waiter;
                    if (w != null) {    // waiters need at most one unpark
                        waiter = null;
                        LockSupport.unpark(w);
                    }
                    return true;
                }
                return match == s;
            }

如果match为null，说明还没有节点来匹配该节点，那么就设置s节点为该节点的匹配节点，然后唤醒waiter来进行数据交易，交易成功返回true，如果match不为null，说明当前节点已经被匹配走了，那么判断是否和s节点完成匹配，返回匹配结果。现在来看看TransferStack中的数据转移方法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
            int mode = (e == null) ? REQUEST : DATA;

            for (;;) {
                SNode h = head;
                if (h == null || h.mode == mode) {  // empty or same-mode
                    if (timed && nanos <= 0) {      // can't wait
                        if (h != null && h.isCancelled())
                            casHead(h, h.next);     // pop cancelled node
                        else
                            return null;
                    } else if (casHead(h, s = snode(s, e, h, mode))) {
                        SNode m = awaitFulfill(s, timed, nanos);
                        if (m == s) {               // wait was cancelled
                            clean(s);
                            return null;
                        }
                        if ((h = head) != null && h.next == s)
                            casHead(h, s.next);     // help s's fulfiller
                        return (E) ((mode == REQUEST) ? m.item : s.item);
                    }
                } else if (!isFulfilling(h.mode)) { // try to fulfill
                    if (h.isCancelled())            // already cancelled
                        casHead(h, h.next);         // pop and retry
                    else if (casHead(h, s=snode(s, e, h, FULFILLING|mode))) {
                        for (;;) { // loop until matched or waiters disappear
                            SNode m = s.next;       // m is s's match
                            if (m == null) {        // all waiters are gone
                                casHead(s, null);   // pop fulfill node
                                s = null;           // use new node next time
                                break;              // restart main loop
                            }
                            SNode mn = m.next;
                            if (m.tryMatch(s)) {
                                casHead(s, mn);     // pop both s and m
                                return (E) ((mode == REQUEST) ? m.item : s.item);
                            } else                  // lost match
                                s.casNext(m, mn);   // help unlink
                        }
                    }
                } else {                            // help a fulfiller
                    SNode m = h.next;               // m is h's match
                    if (m == null)                  // waiter is gone
                        casHead(h, null);           // pop fulfilling node
                    else {
                        SNode mn = m.next;
                        if (m.tryMatch(h))          // help match
                            casHead(h, mn);         // pop both h and m
                        else                        // lost match
                            h.casNext(m, mn);       // help unlink
                    }
                }
            }
        }

翻译一下方法的注释，这也是整个方法的运行算法：

1. 如果当前的交易栈是空的，或者包含与请求交易节点模式相同的节点，那么就将这个请求交易的节点作为新的栈顶节点，等待被下一个请求交易的节点匹配，最后会返回匹配节点的数据或者null，如果被取消则会返回null。
2. 如果当前交易栈不为空，并且请求交易的节点和当前栈顶节点模式互补，那么将这个请求交易的节点的模式变为FULFILLING，然后将其压入栈中，和互补的节点进行匹配，完成交易之后将两个节点一起弹出，并且返回交易的数据。
3. 如果栈顶已经存在一个模式为FULFILLING的节点，说明栈顶的节点正在进行匹配，那么就帮助这个栈顶节点快速完成交易，然后继续交易。

根据上面三个点，来对照了代码走一遍。主要分析一下方法中调用到的其他方法的实现细节，先来分析一下casHead这个方法，下面首先展示了这个方法：

        boolean casHead(SNode h, SNode nh) {
            return h == head &&
                UNSAFE.compareAndSwapObject(this, headOffset, h, nh);
        }

这个方法的功能是将nh设置为新的head，head表示的是当前的栈顶节点，h是旧的栈顶节点，该方法会通过CAS来安全完整的设置新的head，这个方法在交易节点被取消的时候会调用。接下来看一下awaitFulfill方法，这个方法实现的是等待其他的线程来匹配，下面展示了它的实现：

        SNode awaitFulfill(SNode s, boolean timed, long nanos) {
            final long deadline = timed ? System.nanoTime() + nanos : 0L;
            Thread w = Thread.currentThread();
            int spins = (shouldSpin(s) ?
                         (timed ? maxTimedSpins : maxUntimedSpins) : 0);
            for (;;) {
                if (w.isInterrupted())
                    s.tryCancel();
                SNode m = s.match;
                if (m != null)
                    return m;
                if (timed) {
                    nanos = deadline - System.nanoTime();
                    if (nanos <= 0L) {
                        s.tryCancel();
                        continue;
                    }
                }
                if (spins > 0)
                    spins = shouldSpin(s) ? (spins-1) : 0;
                else if (s.waiter == null)
                    s.waiter = w; // establish waiter so can park next iter
                else if (!timed)
                    LockSupport.park(this);
                else if (nanos > spinForTimeoutThreshold)
                    LockSupport.parkNanos(this, nanos);
            }
        }

这个线程一直阻塞直到被匹配，在阻塞之前首先会自旋，这个自旋会在阻塞之前进行，它会调用shouldSpin方法来进行判断是否需要自选，下面展示了shouldSpin这个方法：

        /**
         * Returns true if node s is at head or there is an active
         * fulfiller.
         */
        boolean shouldSpin(SNode s) {
            SNode h = head;
            return (h == s || h == null || isFulfilling(h.mode));
        }

如果当前节点在栈顶，并且正在请求交易，那么就应该自旋。在多CPU的环境下，这种情况下的自旋是有必要的，因为很可能立刻就会有新的线程到来，那么就会立刻进行交易而不需要进行阻塞，然后被唤醒，这是需要过程的，所以这样的自旋等待是值得的。回到awaitFulfill方法，接着看，如果被中断了，那么就取消交易，如果匹配成功了，那么就返回匹配的节点，否则就旋或者进行阻塞。等待被请求交易的线程唤醒。再来看transfer方法，当一个节点被取消交易了，那么就要进行清理，而判断是否被取消交易的依据是什么呢？首选看一下取消交易的方法：

            /**
             * Tries to cancel a wait by matching node to itself.
             */
            void tryCancel() {
                UNSAFE.compareAndSwapObject(this, matchOffset, null, this);
            }

可以看到，取消交易就是将match指向自己，而在awaitFulfill方法中返回的就是match，那么awaitFulfill方法返回之后做一下判断，如果和自己相等，那么就是被取消交易了，那么就需要调用方法clean来清理一下，下面是clean方法的细节：

        /**
         * Unlinks s from the stack.
         */
        void clean(SNode s) {
            s.item = null;   // forget item
            s.waiter = null; // forget thread

            SNode past = s.next;
            if (past != null && past.isCancelled())
                past = past.next;

            // Absorb cancelled nodes at head
            SNode p;
            while ((p = head) != null && p != past && p.isCancelled())
                casHead(p, p.next);

            // Unsplice embedded nodes
            while (p != null && p != past) {
                SNode n = p.next;
                if (n != null && n.isCancelled())
                    p.casNext(n, n.next);
                else
                    p = n;
            }
        }

这个方法首先找到接下来的第一个不为null并且没有被取消交易的节点past，然后设置新的head节点，并把已经取消交易的节点移除掉。回到transfer方法中，有下面一句代码：

 int mode = (e == null) ? REQUEST : DATA;
 return (E) ((mode == REQUEST) ? m.item : s.item);

这句话的意思是什么呢，如果mode为REQUEST，这个也就是说这是一个请求交易的生产者节点，那么返回的数据就是m节点的数据，否则是s节点的数据，那么m节点和s节点分别是什么呢？m节点是匹配到的节点，也就是和s匹配的线程，而s为请求匹配的线程，所以，当请求的是消费者线程的话，那么就会返回与之匹配的生产者线程的数据，否则返回自身携带的数据。

在第二种情况下，如果请求匹配的节点发现当前栈顶和自己互补，那么就会开始和栈顶节点进行交易，主要的代码如下：

         SNode mn = m.next;
         if (m.tryMatch(s)) {
             casHead(s, mn);// pop both s and m
             return (E) ((mode == REQUEST) ? m.item : s.item);
          } else // lost match
             s.casNext(m, mn);// help unlink

m节点试图使用上文中分析过的tryMatch方法来进行匹配，如果匹配成功则将m节点和s节点都弹出栈，设置新的栈顶节点。否则没有匹配成功，可以解释为m已经被其他线程匹配走了，那么就需要把m从栈中移除。然后是第三种情况，如果不满足上面两种情况的话，那么说明栈顶节点正在进行匹配过程，那么就帮助栈顶节点进行匹配，然后再进行自身匹配，而且这种付出是值得的，具体的实现细节参照代码和注释。

公平交易实现TransferQueue

现在来分析一下公平交易TransferQueue的实现，在分析了非公平交易的实现之后，对公平交易的分析就会比较粗略，首先来看一下TransferQueue的结构，它使用了队列这种数据结构来实现公平交易，这种公平是对多线程来说的，等待得越久的线程越快被互补节点匹配，而在非公平交易中不是这样的。下面展示了TransferQueue使用的队列节点的结构：

            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;

接下来就是公平交易的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
            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
                    QNode tn = t.next;
                    if (t != tail)                  // inconsistent read
                        continue;
                    if (tn != null) {               // lagging tail
                        advanceTail(t, tn);
                        continue;
                    }
                    if (timed && nanos <= 0)        // can't wait
                        return null;
                    if (s == null)
                        s = new QNode(e, isData);
                    if (!t.casNext(null, s))        // failed to link in
                        continue;

                    advanceTail(t, s);              // swing tail and wait
                    Object x = awaitFulfill(s, e, timed, nanos);
                    if (x == s) {                   // wait was cancelled
                        clean(t, s);
                        return null;
                    }

                    if (!s.isOffList()) {           // not already unlinked
                        advanceHead(t, s);          // unlink if head
                        if (x != null)              // and forget fields
                            s.item = s;
                        s.waiter = null;
                    }
                    return (x != null) ? (E)x : e;

                } else {                            // complementary-mode
                    QNode m = h.next;               // node to fulfill
                    if (t != tail || m == null || h != head)
                        continue;                   // inconsistent read

                    Object x = m.item;
                    if (isData == (x != null) ||    // m already fulfilled
                        x == m ||                   // m cancelled
                        !m.casItem(x, e)) {         // lost CAS
                        advanceHead(h, m);          // dequeue and retry
                        continue;
                    }

                    advanceHead(h, m);              // successfully fulfilled
                    LockSupport.unpark(m.waiter);
                    return (x != null) ? (E)x : e;
                }
            }
        }

根据注释，下面来说明一下整个方法的运行流程：

1. 如果队列为空，或者请求交易的节点和队列中的节点具有相同的交易类型，那么就将该请求交易的节点添加到队列尾部等待交易，直到被匹配或者被取消
2. 如果队列中包含了等待的节点，并且请求的节点和等待的节点是互补的，那么进行匹配并且进行交易

第一种情况的判断条件为：

 QNode t = tail;
 QNode h = head;
if (h == t || t.isData == isData)

也就是说，如果头为节点相等（队列为空），或者队列尾部节点和请求的节点具有相同的交易类型，那么就将节点添加到队列尾部，并且等待匹配。否则进行匹配。下面来看一下awaitFulfill方法的实现细节：

 Object awaitFulfill(QNode s, E e, boolean timed, long nanos) {
            /* Same idea as TransferStack.awaitFulfill */
            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);
                Object x = s.item;
                if (x != e)
                    return x;
                if (timed) {
                    nanos = deadline - System.nanoTime();
                    if (nanos <= 0L) {
                        s.tryCancel(e);
                        continue;
                    }
                }
                if (spins > 0)
                    --spins;
                else if (s.waiter == null)
                    s.waiter = w;
                else if (!timed)
                    LockSupport.park(this);
                else if (nanos > spinForTimeoutThreshold)
                    LockSupport.parkNanos(this, nanos);
            }
        }

这个方法的等待节点被匹配，如果匹配成功则返回匹配到的节点，否则阻塞等待被匹配的线程唤醒。下面再来看一下呗取消之后的clean方法的实现细节：

 void clean(QNode pred, QNode s) {
            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.
             */
            while (pred.next == s) { // Return early if already unlinked
                QNode h = head;
                QNode hn = h.next;   // Absorb cancelled first node as head
                if (hn != null && hn.isCancelled()) {
                    advanceHead(h, hn);
                    continue;
                }
                QNode t = tail;      // Ensure consistent read for tail
                if (t == h)
                    return;
                QNode tn = t.next;
                if (t != tail)
                    continue;
                if (tn != null) {
                    advanceTail(t, tn);
                    continue;
                }
                if (s != t) {        // If not tail, try to unsplice
                    QNode sn = s.next;
                    if (sn == s || pred.casNext(s, sn))
                        return;
                }
                QNode dp = cleanMe;
                if (dp != null) {    // Try unlinking previous cancelled node 【主要的部分】
                    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
            }
        }

可以发现这个方法较为复杂，现在要提到一个成员变量：cleanMe，这个变量保存的是一个被取消交易但是没有被移除队列的节点，这个节点总是最后被添加到队列的。clean方法的实现目前还没有搞明白，以后补上吧。

SynchronousQueue的交易实现

SynchronousQueue使用了transfer来交易数据，上文分析了公平交易和非公平交易，理解了上面的两个交易类之后，就很好理解队列的具体交易了，其实SynchronousQueue的交易也就仅仅是调用transfer来进行而已，下面看两个交易方法的实现，一个是生产者交付数据的方法put，一个是消费者获取数据的take，首先看代码：

    public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        if (transferer.transfer(e, false, 0) == null) {
            Thread.interrupted();
            throw new InterruptedException();
        }
    }

    public E take() throws InterruptedException {
        E e = transferer.transfer(null, false, 0);
        if (e != null)
            return e;
        Thread.interrupted();
        throw new InterruptedException();
    }
    

虽然实看起来非常简洁，但是transfer的实现却是很复杂的，需要去理解上文中提到的两个交易类，才能很好的理解这些交易方法。最后需要注意的是，因为SynchronousQueue队列的特殊性，你不能使用和put队列提供的方法来访问SynchronousQueue，比如peek，因为任意的线程都是带有交易属性的，一个peek的线程只是想剽窃一眼是否有数据，不希望消费数据，但是SynchronousQueue不提供类似的剽窃能力。而size方法也是没有意义的，因为任意线程都是交付完成再返回的，所以没有容量的概念。具体的哪些方法是不支持的可以参考源码。还需要说明的一点是，本文仅仅是对文章Java阻塞队列详解的补充，未来还会持续进行。

扫码入群