【Java多线程】SynchronousQueue源码分析（二十四）

原文连接：SynchronousQueue原理解析

源码解析

1、SynchronousQueue.java

public class SynchronousQueue<E> extends AbstractQueue<E>
    implements BlockingQueue<E>, java.io.Serializable {
    
    //Transferer是一个抽象类，SynchronousQueue内部有2个Transferer的子类，分别是TransferQueue和TransferStack
    //
    private transient volatile Transferer<E> transferer;

    //默认构造方法的线程等待队列是不保证顺序的
    public SynchronousQueue() {
        this(false);
    }

    //如果fair为true，那SynchronousQueue所采用的是能保证先进先出的TransferQueue，也就是先被挂起的线程会先返回
    public SynchronousQueue(boolean fair) {
        transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
    }
    
    //向SynchronousQueue中添加数据，如果此时线程队列中没有获取数据的线程的话，当前的线程就会挂起等待
    public void put(E e) throws InterruptedException {
        //添加的数据不能是null
        if (e == null) throw new NullPointerException();
        //可以看到添加的方法调用的是transfer方法，如果添加失败会抛出InterruptedException异常
        //后面我们可以在transfer方法的源码中调用put方法添加数据在当前线程被中断时才会返回null
        //这里相当于继续把线程中断的InterruptedException向上抛出
        if (transferer.transfer(e, false, 0) == null) {
            Thread.interrupted();
            throw new InterruptedException();
        }
    }
    
    //不带超时时间的offer方法，如果此时没有线程正在等待获取数据的话transfer就会返回null,也就是添加数据失败
    public boolean offer(E e) {
        if (e == null) throw new NullPointerException();
        return transferer.transfer(e, true, 0) != null;
    }
    
    //带超时时间的offer方法，与上面的不同的是这个方法会等待一个超时时间，如果时间过了还没有线程来获取数据就会返回失败
    public boolean offer(E e, long timeout, TimeUnit unit)
        throws InterruptedException {
        if (e == null) throw new NullPointerException();
        //添加的数据被其他线程成功获取，返回成功
        if (transferer.transfer(e, true, unit.toNanos(timeout)) != null)
            return true;
        //如果添加数据失败了，有可能是线程被中断了，不是的话直接返回false
        if (!Thread.interrupted())
            return false;
        //是线程被中断的话就向上跑出InterruptedException异常
        throw new InterruptedException();
    }
    
    //take方法用于从队列中取数据，如果此时没有添加数据的线程被挂起，那当前线程就会被挂起等待
    public E take() throws InterruptedException {
        E e = transferer.transfer(null, false, 0);
        //成功获取数据
        if (e != null)
            return e;
        //没有获取到数据，同时又退出挂起状态了，那说明线程被中断了，向上抛出InterruptedException
        Thread.interrupted();
        throw new InterruptedException();
    }
    
    //poll方法同样用于获取数据
    public E poll() {
        return transferer.transfer(null, true, 0);
    }
    
    //带超时时间的poll方法，如果超时时间到了还没有线程插入数据，就会返回失败
    public E poll(long timeout, TimeUnit unit) throws InterruptedException {
        E e = transferer.transfer(null, true, unit.toNanos(timeout));
        //返回结果有2种情况
        //e != null表示成功取到数据了
        //!Thread.interrupted()表示返回失败了，且是因为超时失败的，此时e是null
        if (e != null || !Thread.interrupted())
            return e;
        //返回失败了，并且是因为当前线程被中断了
        throw new InterruptedException();
    }
    
    //可以看到SynchronousQueue的isEmpty方法一直返回的是true，因为SynchronousQueue没有任何容量
    public boolean isEmpty() {
        return true;
    }

    //同样的size方法也返回0
    public int size() {
        return 0;
    }
    
    <!--下面我们看看TransferQueue的具体实现，TransferQueue中的关键方法就是transfer方法了-->
    
    //先看看TransferQueue的父类Transferer，比较简单，就是提供了一个transfer方法，需要子类具体实现
    abstract static class Transferer<E> {
        abstract E transfer(E e, boolean timed, long nanos);
    }
    
    //TransferQueue
    static final class TransferQueue<E> extends Transferer<E> {
    
        //内部的节点类，用于表示一个请求
        //这里可以看出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
            //用于判断是入队还是出队，true表示的是入队操作，也就是添加数据
            final boolean isData;

            QNode(Object item, boolean isData) {
                this.item = item;
                this.isData = isData;
            }

            //可以看到QNode内部通过volatile关键字以及Unsafe类的CAS方法来实现线程安全
            //compareAndSwapObject方法第一个参数表示需要改变的对象，第二个参数表示偏移量
            //第三个参数表示参数期待的值，第四个参数表示更新后的值
            //下面的方法调用的意思是将当前的QNode对象(this)的next字段赋值为val，当目前的next的值是cmp时就会更新next字段成功
            boolean casNext(QNode cmp, QNode val) {
                return next == cmp &&
                    U.compareAndSwapObject(this, NEXT, cmp, val);
            }

            //方法的原理同上面的类似，这里就是更新item的值了
            boolean casItem(Object cmp, Object val) {
                return item == cmp &&
                    U.compareAndSwapObject(this, ITEM, cmp, val);
            }

            //方法的原理同上面的类似，这里把item赋值为自己，就表示取消当前节点表示的操作了
            void tryCancel(Object cmp) {
                U.compareAndSwapObject(this, ITEM, cmp, this);
            }

            //调用tryCancel方法后item就会是this，就表示当前任务被取消了
            boolean isCancelled() {
                return item == this;
            }

            //表示当前任务已经被返回了
            boolean isOffList() {
                return next == this;
            }

            // Unsafe mechanics
            private static final sun.misc.Unsafe U = sun.misc.Unsafe.getUnsafe();
            private static final long ITEM;
            private static final long NEXT;

            static {
                try {
                    ITEM = U.objectFieldOffset
                        (QNode.class.getDeclaredField("item"));
                    NEXT = U.objectFieldOffset
                        (QNode.class.getDeclaredField("next"));
                } catch (ReflectiveOperationException e) {
                    throw new Error(e);
                }
            }
        }
        
        //首节点
        transient volatile QNode head;
        //尾部节点
        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;

        //构造函数中会初始化一个出队的节点，并且首尾都指向这个节点
        TransferQueue() {
            QNode h = new QNode(null, false); // initialize to dummy node.
            head = h;
            tail = h;
        }
        
        //transfer方法用于提交数据或者是获取数据
        @SuppressWarnings("unchecked")
        E transfer(E e, boolean timed, long nanos) {
            QNode s = null; // constructed/reused as needed
            //如果e不为null，就说明是添加数据的入队操作
            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;
                    //这是一个检查，确保t指向队尾
                    if (t != tail)                  // inconsistent read
                        continue;
                    //tn不为null，说明t不是尾部节点，就执行advanceTail操作，将tn作为尾部节点，继续循环
                    if (tn != null) {               // lagging tail
                        advanceTail(t, tn);
                        continue;
                    }
                    //如果timed为true，表示带有超时参数，等待超时期间没有其他相反操作的线程提交就会直接返回null
                    //这里如果nanos初始值就是0，比如不带超时时间的offer和poll方法，当队尾的节点不是相反操作时就会直接返回null
                    if (timed && nanos <= 0L)       // can't wait
                        return null;
                    //如果没有超时时间或者超时时间不为0的话就创建新的节点
                    if (s == null)
                        s = new QNode(e, isData);
                    //使tail的next指向新的节点
                    if (!t.casNext(null, s))        // failed to link in
                        continue;
                    //更新TransferQueue的tail指向新的节点，这样tail节点就始终是尾部节点
                    advanceTail(t, s);              // swing tail and wait
                    //如果当前操作是带超时时间的，则进行超时等待，否则就挂起线程，直到有新的反向操作提交
                    Object x = awaitFulfill(s, e, timed, nanos);
                    //当挂起的线程被中断或是超时时间已经过了，awaitFulfill方法就会返回当前节点，这样就会有x == s为true
                    if (x == s) {                   // wait was cancelled
                        //将队尾节点移出，并重新更新尾部节点，返回null，就是入队或是出队操作失败了
                        clean(t, s);
                        return null;
                    }
                    
                    //如果s还没有被
                    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;
                    //这里先判断m是否是有效的操作
                    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
                    //唤醒m节点的被挂起的线程
                    LockSupport.unpark(m.waiter);
                    //返回的结果用于给对应的操作，如take、offer等判断是否执行操作成功
                    return (x != null) ? (E)x : e;
                }
            }
        }
        
        <!--下面看看执行挂起线程的方法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 ? MAX_TIMED_SPINS : MAX_UNTIMED_SPINS)
                : 0;
            for (;;) {
                //这里首先判断线程是否被中断了，如果被中断了就取消等待，并设置s的item指向s本身作为标记
                if (w.isInterrupted())
                    s.tryCancel(e);
                Object x = s.item;
                //x != e就表示超时时间到了或是线程被中断了，也就是执行了tryCancel方法
                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)
                    //park操作会让线程挂起进入等待状态(Waiting)，需要其他线程调用unpark方法唤醒
                    LockSupport.park(this);
                else if (nanos > SPIN_FOR_TIMEOUT_THRESHOLD)
                    //parkNanos操作会让线程挂起进入限期等待(Timed Waiting)，不用其他线程唤醒，时间到了会被系统唤醒
                    LockSupport.parkNanos(this, nanos);
            }
        }
        
    }
}

2、构造函数

 public SynchronousQueue() {
        this(false);
    }

    public SynchronousQueue(boolean fair) {
        // 通过 fair 值来决定公平性和非公平性
        // 公平性使用TransferQueue，非公平性采用TransferStack
        transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
    }

3、公平模式 TransferQueue

/**
 *  这是一个非常典型的 queue , 它有如下的特点
 *  1. 整个队列有 head, tail 两个节点
 *  2. 队列初始化时会有个 dummy 节点
 *  3. 这个队列的头节点是个 dummy 节点/ 或 哨兵节点, 所以操作的总是队列中的第二个节点(AQS的设计中也是这也)
 */

/** 头节点 */
transient volatile QNode head;
/** 尾节点 */
transient volatile QNode tail;
/**
 * Reference to a cancelled node that might not yet have been
 * unlinked from queue because it was last inserted node
 * when it was cancelled
 */
/**
 * 对应 中断或超时的 前继节点,这个节点存在的意义是标记, 它的下个节点要删除
 * 何时使用:
 *      当你要删除 节点 node, 若节点 node 是队列的末尾, 则开始用这个节点,
 * 为什么呢？
 *      大家知道 删除一个节点 直接 A.CASNext(B, B.next) 就可以,但是当  节点 B 是整个队列中的末尾元素时,
 *      一个线程删除节点B, 一个线程在节点B之后插入节点 这样操作容易致使插入的节点丢失, 这个cleanMe很像
 *      ConcurrentSkipListMap 中的 删除添加的 marker 节点, 他们都是起着相同的作用
 */
transient volatile QNode cleanMe;

TransferQueue(){
    /**
     * 构造一个 dummy node, 而整个 queue 中永远会存在这样一个 dummy node
     * dummy node 的存在使得 代码中不存在复杂的 if 条件判断
     */
    QNode h = new QNode(null, false);
    head = h;
    tail = h;
}

/**
 * 推进 head 节点,将 老节点的 oldNode.next = this, help gc,
 * 这种和 ConcurrentLinkedQueue 中一样
 */
void advanceHead(QNode h, QNode nh){
    if(h == head && unsafe.compareAndSwapObject(this, headOffset, h, nh)){
        h.next = h; // forget old next help gc
    }
}

/** 更新新的 tail 节点 */
void advanceTail(QNode t, QNode nt){
    if(tail == t){
        unsafe.compareAndSwapObject(this, tailOffset, t, nt);
    }
}

/** CAS 设置 cleamMe 节点 */
boolean casCleanMe(QNode cmp, QNode val){
    return cleanMe == cmp && unsafe.compareAndSwapObject(this, cleanMeOffset, cmp, val);
}

4、公平模式 TransferQueue中定义了QNode类

static final class QNode {
        // next 域
        volatile QNode next;
        // item数据项
        volatile Object item;
        //  等待线程，用于park/unpark
        volatile Thread waiter;       // to control park/unpark
        //模式，表示当前是数据还是请求，只有当匹配的模式相匹配时才会交换
        final boolean isData;

        QNode(Object item, boolean isData) {
            this.item = item;
            this.isData = isData;
        }
...

5、公平模式 TransferQueue transfer() 方法

 /**
 * Puts or takes an item
 * 主方法
 *
 * @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 nanosecond
 * @return
 */
@Override
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 gurading against
     * seeing uninitialized head or tail value. 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 implicity interspersed
     *
     * 这个 producer / consumer 的主方法, 主要分为两种情况
     *
     * 1. 若队列为空 / 队列中的尾节点和自己的 类型相同, 则添加 node
     *      到队列中, 直到 timeout/interrupt/其他线程和这个线程匹配
     *      timeout/interrupt awaitFulfill方法返回的是 node 本身
     *      匹配成功的话, 要么返回 null (producer返回的), 或正真的传递值 (consumer 返回的)
     *
     * 2. 队列不为空, 且队列的 head.next 节点是当前节点匹配的节点,
     *      进行数据的传递匹配, 并且通过 advanceHead 方法帮助 先前 block 的节点 dequeue
     */
    QNode s = null; // constrcuted/reused as needed
    boolean isData = (e != null); // 1.判断 e != null 用于区分 producer 与 consumer

    for(;;){
        QNode t = tail;
        QNode h = head;
        if(t == null || h == null){         // 2. 数据未初始化, continue 重来
            continue;                       // spin
        }
        if(h == t || t.isData == isData){   // 3. 队列为空, 或队列尾节点和自己相同 (注意这里是和尾节点比价, 下面进行匹配时是和 head.next 进行比较)
            QNode tn = t.next;
            if(t != tail){                  // 4. tail 改变了, 重新再来
                continue;
            }
            if(tn != null){                 // 5. 其他线程添加了 tail.next, 所以帮助推进 tail
                advanceTail(t, tn);
                continue;
            }
            if(timed && nanos <= 0){        // 6. 调用的方法的 wait 类型的, 并且 超时了, 直接返回 null, 直接见 SynchronousQueue.poll() 方法,说明此 poll 的调用只有当前队列中正好有一个与之匹配的线程在等待被【匹配才有返回值
                return null;
            }
            if(s == null){
                s = new QNode(e, isData);  // 7. 构建节点 QNode
            }
            if(!t.casNext(null, s)){      // 8. 将 新建的节点加入到 队列中
                continue;
            }

            advanceTail(t, s);             // 9. 帮助推进 tail 节点
            Object x = awaitFulfill(s, e, timed, nanos); // 10. 调用awaitFulfill, 若节点是 head.next, 则进行一些自旋, 若不是的话, 直接 block, 知道有其他线程 与之匹配, 或它自己进行线程的中断
            if(x == s){                   // 11. 若 (x == s)节点s 对应额线程 wait 超时 或线程中断, 不然的话 x == null (s 是 producer) 或 是正真的传递值(s 是 consumer)
                clean(t, s);              // 12. 对接点 s 进行清除, 若 s 不是链表的最后一个节点, 则直接 CAS 进行 节点的删除, 若 s 是链表的最后一个节点, 则 要么清除以前的 cleamMe 节点(cleamMe != null), 然后将 s.prev 设置为 cleanMe 节点, 下次进行删除 或直接将 s.prev 设置为cleanMe
                return null;
            }

            if(!s.isOffList()){          // 13. 节点 s 没有 offlist
                advanceHead(t, s);       // 14. 推进head 节点, 下次就调用 s.next 节点进行匹配(这里调用的是 advanceHead, 因为代码能执行到这边说明s已经是 head.next 节点了)
                if(x != null){          // and forget fields
                    s.item = s;
                }
                s.waiter = null;       // 15. 释放线程 ref
            }

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

        }else{                              // 16. 进行线程的匹配操作, 匹配操作是从 head.next 开始匹配 (注意 队列刚开始构建时 有个 dummy node, 而且 head 节点永远是个 dummy node 这个和 AQS 中一样的)
            QNode m = h.next;               // 17. 获取 head.next 准备开始匹配
            if(t != tail || m == null || h != head){
                continue;                  // 18. 不一致读取, 有其他线程改变了队列的结构inconsistent read
            }

            /** producer 和 consumer 匹配操作
             *  1. 获取 m的 item (注意这里的m是head的next节点
             *  2. 判断 isData 与x的模式是否匹配, 只有produce与consumer才能配成一对
             *  3. x == m 判断是否 节点m 是否已经进行取消了, 具体看(QNOde#tryCancel)
             *  4. m.casItem 将producer与consumer的数据进行交换 (这里存在并发时可能cas操作失败的情况)
             *  5. 若 cas操作成功则将h节点dequeue
             *
             *  疑惑: 为什么将h进行 dequeue, 而不是 m节点
             *  答案: 因为每次进行配对时, 都是将 h 是个 dummy node, 正真的数据节点 是 head.next
             */
            Object x = m.item;
            if(isData == (x != null) ||    // 19. 两者的模式是否匹配 (因为并发环境下 有可能其他的线程强走了匹配的节点)
                    x == m ||               // 20. m 节点 线程中断或者 wait 超时了
                    !m.casItem(x, e)        // 21. 进行 CAS 操作 更改等待线程的 item 值(等待的有可能是 concumer / producer)
                    ){
                advanceHead(h, m);          // 22.推进 head 节点 重试 (尤其 21 操作失败)
                continue;
            }

            advanceHead(h, m);             // 23. producer consumer 交换数据成功, 推进 head 节点
            LockSupport.unpark(m.waiter); // 24. 换线等待中的 m 节点, 而在 awaitFulfill 方法中 因为 item 改变了,  所以 x != e 成立, 返回
            return (x != null) ? (E)x : e; // 25. 操作到这里若是 producer, 则 x != null, 返回 x, 若是consumer， 则 x == null,.返回 producer(其实就是 节点m) 的 e
        }
    }

}

6、公平模式 TransferQueueawaitFulfill() 方法

// 自旋或者等待，直到填充完毕
// 这里的策略是什么呢？如果自旋次数不够了，通常是 16 次，但还有超过 1 秒的时间，就阻塞等待被唤醒。
// 如果时间到了，就取消这次的入队行为。
// 返回的是 Node 本身
// s.item 就是 e 
Object awaitFulfill(QNode s, E e, boolean timed, long nanos) {
    final long deadline = timed ? System.nanoTime() + nanos : 0L;
    Thread w = Thread.currentThread();
    int spins = ((head.next == s) ?// 如果成功将 tail.next 覆盖了 tail，如果有超时机制，则自旋 32 次，如果没有超时机制，则自旋 32 *16 = 512次
                 (timed ? maxTimedSpins : maxUntimedSpins) : 0);
    for (;;) {
        if (w.isInterrupted())// 当前线程被中断
            s.tryCancel(e);// 尝试取消这个 item
        Object x = s.item;// 获取到这个 tail 的 item
        if (x != e) // 如果不相等，说明 node 中的 item 取消了，返回这个 item。
            // 这里是唯一停止循环的地方。当 s.item 已经不是当初的哪个 e 了，说明要么是时间到了被取消了，要么是线程中断被取消了。
            // 当然，不仅仅只有这2种 “意外” 情况，还有一种情况是：当另一个线程拿走了这个数据，并修改了 item，也会通过这个判断，返回被“修改”过的 item。
            return x;
        if (timed) {// 如果有时间限制
            nanos = deadline - System.nanoTime();
            if (nanos <= 0L) {// 如果时间到了
                s.tryCancel(e);// 尝试取消 item，供上面的 x != e 判断
                continue;// 重来
            }
        }
        if (spins > 0)// 如果还有自旋次数
            --spins;// 减一
        else if (s.waiter == null)// 如果自旋不够，且 tail 的等待线程还没有赋值
            s.waiter = w;// 当前线程赋值给 tail 的等待线程
        else if (!timed)// 如果自旋不够，且如果线程赋值过了，且没有限制时间，则 wait，（危险操作）
            LockSupport.park(this);
        else if (nanos > spinForTimeoutThreshold)// 如果自旋不够，且如果限制了时间，且时间还剩余超过 1 秒，则 wait 剩余时间。
            // 主要目的就是等待，等待其他线程唤醒这个节点所在的线程。
            LockSupport.parkNanos(this, nanos);
    }
}

====

链接：https://www.jianshu.com/p/af6f83c78506