【Java并发编程】17、SynchronousQueue源码分析

SynchronousQueue是一种特殊的阻塞队列，不同于LinkedBlockingQueue、ArrayBlockingQueue和PriorityBlockingQueue，其内部没有任何容量，任何的入队操作都需要等待其他线程的出队操作，反之亦然。如果将SynchronousQueue用于生产者/消费者模式，那么相当于生产者和消费者手递手交易，即生产者生产出一个货物，则必须等到消费者过来取货，方可完成交易。
SynchronousQueue有一个fair选项，如果fair为true，称为fair模式，否则就是unfair模式。fair模式使用一个先进先出的队列保存生产者或者消费者线程，unfair模式则使用一个后进先出的栈保存。

基本原理

SynchronousQueue通过将入队出队的线程绑定到队列的节点上，并借助LockSupport的park()和unpark()实现等待，先到达的线程A需调用LockSupport的park()方法将当前线程进入阻塞状态，知道另一个与之匹配的线程B调用LockSupport.unpark(Thread)来唤醒在该节点上等待的线程A。
基本逻辑：

初始状态队列为null
当一个线程到达，如果队列为null，无与之匹配的线程，则进入队列等待；队列不为null，参考3
当另一个线程到达，如果队列不为null，则判断队列中的第一个元素（针对fair和unfair不同）是否与其匹配，如果匹配则完成交易，不匹配则也入队；队列为null，参考2

常用方法解析

在深入分析其实现机制之前，我们先了解对于SynchronousQueue可执行哪些操作，由于SynchronousQueue的容量为0，所以一些针对集合的操作，如：isEmpty()/size()/clear()/remove(Object)/contains(Object)等操作都是无意义的，同样peek()也总是返回null。所以针对SynchronousQueue只有两类操作：

入队（put(E)/offer(E, long, TimeUnit)/offer(E)）
出队（take()/poll(long, TimeUnit)/poll()）

这两类操作内部都是调用Transferer的transfer(Object, boolean, long)方法，通过第一个参数是否为null，来区分是生产者还是消费者（生产者不为null）。
针对以上情况，我们将着重分析Transferer的transfer(Object, boolean, long)方法，这里由于两种不同的公平模式，会存在两个Transferer的派生类：

public SynchronousQueue(boolean fair) {

    transferer = (fair)? new TransferQueue() : new TransferStack();

}

可见fair模式使用TransferQueue，unfair模式使用TransferStack，下面我们将分别对这两种模式进行着重分析。

fair模式

fair模式使用一个FIFO的队列保存线程，TransferQueue的结构如下：

/** Dual Queue */

static final class TransferQueue extends Transferer {

    /** Node class for TransferQueue. */

    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) {

            this.item = item;

            this.isData = isData;

        }

        ...

    }

    /** 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 cancelled.

     */

    transient volatile QNode cleanMe;

    TransferQueue() {

        QNode h = new QNode(null, false); // initialize to dummy node.

        head = h;

        tail = h;

    }

    ...

}

以上是TransferQueue的大致结构，可以看到TransferQueue同一个普通的队列，同时存在一个指向队列头部的指针——head，和一个指向队列尾部的指针——tail；cleanMe的存在主要是解决不可清楚队列的尾节点的问题，后面会介绍到；队列的节点通过内部类QNode封装，QNode包含四个变量：

next：指向队列中的下一个节点
item：节点包含的数据
waiter：等待在该节点上的线程
isData：表示该节点由生产者创建还是由消费者创建，由于生产者是放入数据，所以isData==true，而消费者==false

其他的内容就是一些CAS变量以及操作，下面主要分析TransferQueue的三个重要方法：transfer(Object, boolean, long)、awaitFulfill(QNode, Object, boolean, long)、clean(QNode, QNode)。这三个方法是TransferQueue的核心，入口是transfer()，下面具体看代码。

transfer

/**

 * @By Vicky:交换数据，生产者和消费者通过e==null来区分

 */

Object transfer(Object e, boolean timed, long nanos) {

    QNode s = null; // constructed/reused as needed

    boolean isData = (e != null);// e==null，则isData==false,else idData==true

    for (;;) {// 循环

        QNode t = tail;

        QNode h = head;

        if (t == null || h == null)         // 无视即可，具体信息在方法开始的注释中有提到

            continue;                       // spin

        // h==t队列为null,tail的isData==isData表示该队列中的等待的线程与当前线程是相同模式

        //（同为生产者，或者同为消费者）（队列中只存在一种模式的线程）

        // 此时需要将该线程插入到队列中进行等待

        if (h == t || t.isData == isData) {

            QNode tn = t.next;

            if (t != tail)                  // inconsistent read

                continue;

            // 这里的目的是为了帮助其他线程完成入队操作

            if (tn != null) {               // lagging tail

                // 原子性将tail从t更新为tn，即将tail往后移动，直到队列的最后一个元素

                advanceTail(t, tn);

                continue;

            }

            // 如果nanos<=0则说明不等待，那么到这里已经说到队列没有可匹配的线程，所以直接返回null即可

            if (timed && nanos <= 0)        // can't wait

                return null;

            // 仅初始化一次s，节点s会保存isData信息作为生产者和消费者的区分

            if (s == null)

                s = new QNode(e, isData);

            // 原子性的更新t的next指针指向s，上面将tail从t更新为tn就是为了处理此处剩下的操作

            // 由于此处插入一个节点分成了两个步骤，所以过程中会插入其他线程，导致看到不一致状态

            // 所以其他线程会执行剩下的步骤帮助其完成入队操作

            if (!t.casNext(null, s))        // failed to link in

                continue;

            // 如果自己执行失败没有关系，会有其他线程帮忙执行完成的，所以才无需锁，类似ConcurrentLinkedQueue

            advanceTail(t, s);              // swing tail and wait

            // 等待匹配

            Object x = awaitFulfill(s, e, timed, nanos);

            // 这里有两种情况：

            //  A:匹配完成，返回数据

            //  B:等待超时/取消，返回原节点s

            if (x == s) {                   // wait was cancelled

                // 情况B则需要清除掉节点s

                clean(t, s);

                return null;

            }

            // 情况A，则匹配成功了，但是还需要将该节点从队列中移除

            //  由于FIFO原则，所以匹配上的元素必然是队列的第一个元素，所以只需要移动head即可

            if (!s.isOffList()) {           // not already unlinked

                // 移动head指向s，则下次匹配从s.next开始

                advanceHead(t, s);          // unlink if head

                // 清除对节点中保存的数据的引用，GC友好

                if (x != null)              // and forget fields

                    s.item = s;

                s.waiter = null;

            }

            return (x != null)? x : e;

        } else {                            // complementary-mode

            // 进行匹配，从队列的头部开始，即head.next，非head

            QNode m = h.next;               // node to fulfill

            if (t != tail || m == null || h != head)

                continue;                   // inconsistent read

            // 判断该节点的isData是否与当前线程的isData匹配

            // 相等则说明m已经匹配过了，因为正常情况是不相等才对

            // x==m说明m被取消了，见QNode的tryCancel()方法

            // CAS设置m.item为e，这里的e，如果是生产者则是数据，消费者则是null，

            // 所以m如果是生产者，则item变为null，消费者则变为生产者的数据

            // CAS操作失败，则直接将m出队，CAS失败说明m已经被其他线程匹配了，所以将其出队，然后retry

            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;

            }

            // 与m匹配成功，将m出队，并唤醒等待在m上的线程m.waiter

            advanceHead(h, m);              // successfully fulfilled

            LockSupport.unpark(m.waiter);

            return (x != null)? x : e;

        }

    }

}

从上面的代码可以看出TransferQueue.transfer()的整体流程：

判断当前队列是否为null或者队尾线程是否与当前线程匹配，为null或者不匹配都将进行入队操作
入队主要很简单，分成两步：修改tail的next为新的节点，修改tail为新的节点，这两步操作有可能分在两个不同的线程执行，不过不影响执行结果
入队之后需要将当前线程阻塞，调用LockSupport.park()方法，直到打断/超时/被匹配的线程唤醒
如果被取消，则需要调用clean()方法进行清除
由于FIFO，所以匹配总是发生在队列的头部，匹配将修改等待节点的item属性传递数据，同时唤醒等待在节点上的线程

awaitFulfill

下面看看具体如何让一个线程进入阻塞。

/**

 *@ By Vicky:等待匹配，该方法会进入阻塞，直到三种情况下才返回：

 *  a.等待被取消了，返回值为s

 *  b.匹配上了，返回另一个线程传过来的值

 *  c.线程被打断，会取消，返回值为s

 */

Object awaitFulfill(QNode s, Object e, boolean timed, long nanos) {

    // timed==false，则不等待，lastTime==0即可

    long lastTime = (timed)? System.nanoTime() : 0;

    // 当前线程

    Thread w = Thread.currentThread();

    // 循环次数，原理同自旋锁，如果不是队列的第一个元素则不自旋，因为压根轮不上他，自旋只是浪费CPU

    // 如果等待的话则自旋的次数少些，不等待就多些

    int spins = ((head.next == s) ?

                 (timed? maxTimedSpins : maxUntimedSpins) : 0);

    for (;;) {

        if (w.isInterrupted())// 支持打断

            s.tryCancel(e);

        // 如果s的item不等于e，有三种情况：

        // a.等待被取消了，此时x==s

        // b.匹配上了，此时x==另一个线程传过来的值

        // c.线程被打断，会取消，此时x==s

        // 不管是哪种情况都不要再等待了，返回即可

        Object x = s.item;

        if (x != e)

            return x;

        // 等到，直接超时取消

        if (timed) {

            long now = System.nanoTime();

            nanos -= now - lastTime;

            lastTime = now;

            if (nanos <= 0) {

                s.tryCancel(e);

                continue;

            }

        }

        // 自旋，直到spins==0，进入等待

        if (spins > 0)

            --spins;

        // 设置等待线程

        else if (s.waiter == null)

            s.waiter = w;

        // 调用LockSupport.park进入等待

        else if (!timed)

            LockSupport.park(this);

        else if (nanos > spinForTimeoutThreshold)

            LockSupport.parkNanos(this, nanos);

    }

}

awaitFulfill()主要涉及自旋以及LockSupport.park()两个关键点，自旋可去了解自旋锁的原理。

自旋锁原理：通过空循环则霸占着CPU，避免当前线程进入睡眠，因为睡眠/唤醒是需要进行线程上下文切换的，所以如果线程睡眠的时间很段，那么使用空循环能够避免线程进入睡眠的耗时，从而快速响应。但是由于空循环会浪费CPU，所以也不能一直循环。自旋锁一般适合同步快很小，竞争不是很激烈的场景。

LockSupport.park()可到API文档进行了解。

clean

下面再看看如何清除被取消的节点。

/**

 *@By Vicky:清除节点被取消的节点

 */

void clean(QNode pred, QNode s) {

    s.waiter = null; // forget thread

    // 如果pred.next!=s则说明s已经出队了

    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

        // t==h则队列为null

        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

            // 出队方式很简单，将pred.next指向s.next即可

            QNode sn = s.next;

            if (sn == s || pred.casNext(s, sn))

                return;

        }

        // 如果s是队尾元素，那么就需要cleanMe出场了，如果cleanMe==null，则只需将pred赋值给cleanMe即可，

        // 赋值cleanMe的意思是等到s不是队尾时再进行清除，毕竟队尾只有一个

        // 同时将上次的cleanMe清除掉，正常情况下此时的cleanMe已经不是队尾了，因为当前需要清除的节点是队尾

        // (上面说的cleanMe其实是需要清除的节点的前继节点)

        QNode dp = cleanMe;

        if (dp != null) {    // Try unlinking previous cancelled node

            QNode d = dp.next;

            QNode dn;

            // d==null说明需要清除的节点已经没了

            // d==dp说明dp已经被清除了，那么dp.next也一并被清除了

            // 如果d未被取消，说明哪里出错了，将cleanMe清除，不清除这个节点了

            // 后面括号将清除cleanMe的next出局，前提是cleanMe.next没有已经被出局

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

            // dp==pred说明cleanMe.next已经其他线程被更新了

            if (dp == pred)

                return;      // s is already saved node

        } else if (casCleanMe(null, pred))

            return;          // Postpone cleaning s

    }

}

清除节点时有个原则：不能清除队尾节点。所以如果对尾节点需要被清除，则将其保存到cleanMe变量，等待下次进行清除。在清除cleanMe时可能说的有点模糊，因为涉及到太多的并发会出现很多情况，所以if条件太多，导致难以分析全部情况。

以上就是TransferQueue的操作逻辑，下面看看后进先出的TransferStack。

unfair模式

unfair模式使用一个LIFO的队列保存线程，TransferStack的结构如下：

/** Dual stack */

static final class TransferStack extends Transferer {

    /* 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;// 正在匹配中...

    /** 只需要判断mode的第二位是否==1即可，==1则正在匹配中...*/

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

    /** Node class for TransferStacks. */

    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.

        SNode(Object item) {

            this.item = item;

        }

    }

    /** The head (top) of the stack */

    volatile SNode head;

    static SNode snode(SNode s, Object e, SNode next, int mode) {

        if (s == null) s = new SNode(e);

        s.mode = mode;

        s.next = next;

        return s;

    }

}

TransferStacks比TransferQueue的结构复杂些。使用一个head指向栈顶元素，使用内部类SNode封装栈中的节点信息，SNode包含5个变量：

next：指向栈中下一个节点
match：与之匹配的节点
waiter：等待的线程
item：数据
mode：模式，对应REQUEST/DATA/FULFILLING（第三个并不是FULFILLING，而是FULFILLING | REQUEST或者FULFILLING | DATA）

SNode的5个变量，三个是volatile的，另外两个item和mode没有volatile修饰，代码注释给出的解释是：对这两个变量的写总是发生在volatile/原子操作的之前，读总是发生在volatile/原子操作的之后。

上面提到SNode.mode的三个常量表示栈中节点的状态,f分别为：

REQUEST：0，消费者的请求生成的节点
DATA：1，生产者的请求生成的节点
FULFILLING：2，正在匹配中的节点，具体对应的mode值是FULFILLING | REQUEST和FULFILLING | DATA

其他内部基本同TransferQueue，不同之处是当匹配到一个节点时并非是将被匹配的节点出栈，而是将匹配的节点入栈，然后同时将匹配上的两个节点一起出栈。下面我们参照TransferQueue来看看TransferStacks的三个方法：transfer(Object, boolean, long)、awaitFulfill(QNode, Object, boolean, long)、clean(QNode, QNode)。

transfer

/**

 * @By Vicky:交换数据，生产者和消费者通过e==null来区分

 */

Object transfer(Object e, boolean timed, long nanos) {

    SNode s = null; // constructed/reused as needed

    int mode = (e == null)? REQUEST : DATA;// 根据e==null判断生产者还是消费者，对应不同的mode值

    for (;;) {

        SNode h = head;

        // 栈为null或者栈顶元素的模式同当前模式，则进行入栈操作

        if (h == null || h.mode == mode) {  // empty or same-mode

            // 不等待，则直接返回null，返回之前顺带清理下被取消的元素

            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))) {// 入栈，更新栈顶为新节点

                // 等待，返回值m==s，则被取消，需清除

                SNode m = awaitFulfill(s, timed, nanos);

                // m==s说明s被取消了，清除

                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 mode == REQUEST? m.item : s.item;

            }

        } else if (!isFulfilling(h.mode)) { // try to fulfill   // 栈顶未开始匹配，则开始匹配

            // h被取消，则出栈

            if (h.isCancelled())            // already cancelled

                casHead(h, h.next);         // pop and retry

            // 更新栈顶为新插入的节点，并更新节点的mode为FULFILLING，对应判断是否正在出栈的方法

            // 匹配需要先将待匹配的节点入栈，所以不管是匹配还是不匹配都需要创建一个节点入栈

            else if (casHead(h, s=snode(s, e, h, FULFILLING|mode))) {

                // 循环直到找到一个可以匹配的节点

                for (;;) { // loop until matched or waiters disappear

                    // m即与s匹配的节点

                    SNode m = s.next;       // m is s's match

                    // m==null说明栈s之后无元素了，直接将栈顶设置为null，并重新进行最外层的循环

                    if (m == null) {        // all waiters are gone

                        casHead(s, null);   // pop fulfill node

                        s = null;           // use new node next time

                        break;              // restart main loop

                    }

                    // 将s设置为m的匹配节点，并更新栈顶为m.next，即将s和m同时出栈

                    SNode mn = m.next;

                    if (m.tryMatch(s)) {

                        casHead(s, mn);     // pop both s and m

                        return (mode == REQUEST)? m.item : s.item;

                    } else                  // lost match

                        // 设置匹配失败，则说明m正准备出栈，帮助出栈

                        s.casNext(m, mn);   // help unlink

                }

            }

        } else {                            // help a fulfiller // 栈顶已开始匹配，帮助匹配

            // 此处的操作逻辑同上面的操作逻辑一致，目的就是帮助上面进行操作，因为此处完成匹配需要分成两步：

            // a.m.tryMatch(s)和b.casHead(s, mn)

            // 所以必然会插入其他线程，只要插入的线程也按照这个步骤执行那么就避免了不一致问题

            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

            }

        }

    }

}

从上面的代码可以看出TransferStack.transfer()的整体流程：

判断当前栈是否为null或者栈顶线程是否与当前线程匹配，为null或者不匹配都将进行入栈操作
入栈主要很简单，分成两步：插入一个节点入栈，该步无需同步，第二步需要head指针指向新节点，该步通过CAS保证安全
入栈之后需要将当前线程阻塞，调用LockSupport.park()方法，直到打断/超时/被匹配的线程唤醒
如果被取消，则需要调用clean()方法进行清除
由于LIFO，所以匹配的节点总是栈顶的两个节点，分成两步：原子性更新节点的match变量，更新head。由于两步无法保证原子性，所以通过将栈顶元素的mode更新为FULFILLING，阻止其他线程在栈顶发生匹配时进行其他操作，同时其他线程需帮助栈顶进行的匹配操作

awaitFulfill

下面看看TransferStack是如何让一个线程进入阻塞。

/**

 *@ By Vicky:等待匹配，逻辑大致同TransferQueue可参考阅读

 */

SNode awaitFulfill(SNode s, boolean timed, long nanos) {

    long lastTime = (timed)? System.nanoTime() : 0;

    Thread w = Thread.currentThread();

    SNode h = head;

    // 计算自旋的次数，逻辑大致同TransferQueue

    int spins = (shouldSpin(s)?

                 (timed? maxTimedSpins : maxUntimedSpins) : 0);

    for (;;) {

        if (w.isInterrupted())

            s.tryCancel();

        // 如果s的match不等于null，有三种情况：

        // a.等待被取消了，此时x==s

        // b.匹配上了，此时match==另一个节点

        // c.线程被打断，会取消，此时x==s

        // 不管是哪种情况都不要再等待了，返回即可

        SNode m = s.match;

        if (m != null)

            return m;

        if (timed) {

            // 等待

            long now = System.nanoTime();

            nanos -= now - lastTime;

            lastTime = now;

            if (nanos <= 0) {

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

    }

}

逻辑基本同TransferQueue，不同之处是通过修改SNode的match变量标示匹配，以及取消。

clean

下面再看看如何清除被取消的节点。

/**

 * @By Vicky:清除节点

 */

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

    // 从栈顶节点开始清除，一直到遇到未被取消的节点，或者直到s.next

    SNode p;

    while ((p = head) != null && p != past && p.isCancelled())

        casHead(p, p.next);

    // Unsplice embedded nodes

    // 如果p本身未取消(上面的while碰到一个未取消的节点就会退出，但这个节点和past节点之间可能还有取消节点)，

    // 再把p到past之间的取消节点都移除。

    while (p != null && p != past) {

        SNode n = p.next;

        if (n != null && n.isCancelled())

            p.casNext(n, n.next);

        else

            p = n;

    }

}

以上即全部的TransferStack的操作逻辑。

看完了TransferQueue和TransferStack的逻辑，SynchronousQueue的逻辑基本清楚了。

应用场景

SynchronousQueue的应用场景得看具体业务需求，J.U.C下有一个应用案例：Executors.newCachedThreadPool()就是使用SynchronousQueue作为任务队列。

出处：http://blog.csdn.net/vickyway/article/details/50113429

巴特西

【Java并发编程】17、SynchronousQueue源码分析

基本原理

常用方法解析

fair模式

transfer

awaitFulfill

clean

unfair模式

transfer

awaitFulfill

clean

应用场景

最新文章

热门文章