SynchronousQueue1.8源码解析

[TOC]

成都创新互联主要从事网页设计、PC网站建设(电脑版网站建设)、wap网站建设(手机版网站建设)、响应式网站开发、程序开发、网站优化、微网站、微信小程序定制开发等,凭借多年来在互联网的打拼,我们在互联网网站建设行业积累了丰富的成都网站建设、网站设计、网站设计、网络营销经验,集策划、开发、设计、营销、管理等多方位专业化运作于一体。

SynchronousQueue 1.8 源码解析

一,简介

SynchronousQueue 是一个很奇怪的队列,感觉都不能叫队列,因为内部没有数据的存储空间,队列不能peek,因为不存在元素,任何入队的线程都会阻塞,直到有线程来出队,也就是这个队列是一组操作,入队和出队要一起离开,出队也是一样,必须等入队,必须结伴而行;队列支持公平和非公平的模式(指的是队列匹配线程的顺序),公平模式的数据结构是队列(FIFO),非公平模式使用的是栈(LIFO)。

二,UML 图

SynchronousQueue 1.8 源码解析

三,基本成员
abstract static class Transferer {
    // 出队入队都是这一个方法
    abstract E transfer(E e, boolean timed, long nanos);
}

    // npu数
    static final int NCPUS = Runtime.getRuntime().availableProcessors();

    // 带超时时间的自旋次数
    static final int maxTimedSpins = (NCPUS < 2) ? 0 : 32;

    // 没有超时的自旋次数
    static final int maxUntimedSpins = maxTimedSpins * 16;
TransferStack 非公平的实现,主要成员

TransferStack 继承 Transferer

注意:这几个状态很重要,因为继承了Transferer,所以出队和入队都是使用的transfer方法,状态是用来区分的,后面方法部分会详解

        /** 0表示消费者 */
        static final int REQUEST    = 0;
        /** 1表示数据的生产者 */
        static final int DATA       = 1;
        /** 2 表示数据正在匹配 */
        static final int FULFILLING = 2;

        static final class SNode {
            volatile SNode next;        // 下一个节点
            volatile SNode match;       // 匹配的节点
            volatile Thread waiter;     // 等待的线程
            Object item;                // 数据
            int mode;                   // 模式 0 , 1 , 2
        }
    /** 头结点 */
    volatile SNode head;
TransferQueue 公平实现,主要成员

TransferQueue 继承 Transferer

    static final class QNode {
            volatile QNode next;          // next 节点
            volatile Object item;         // 数据项
            volatile Thread waiter;       // 等待线程
            final boolean isData;         // 区分生产和消费
    }

        /** 头结点 */
        transient volatile QNode head;
        /** 尾节点 */
        transient volatile QNode tail;
四,常用方法
构造方法
    public SynchronousQueue() {
        this(false);
    }

    // 构造方法,fair表示公平或者非公平
    public SynchronousQueue(boolean fair) {
        transferer = fair ? new TransferQueue() : new TransferStack();
    }
TransferStack 非公平常用方法
offer 方法
    public boolean offer(E e) {
        // e 不能为null
        if (e == null) throw new NullPointerException();
        return transferer.transfer(e, true, 0) != null;
    }

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

    }
poll 方法
    public E poll() {
        return transferer.transfer(null, true, 0);
    }
take 方法
public E take() throws InterruptedException {
        E e = transferer.transfer(null, false, 0);
        if (e != null)
            return e;
        Thread.interrupted();
        throw new InterruptedException();
    }
transfer 方法
    E transfer(E e, boolean timed, long nanos) {

            SNode s = null; // constructed/reused as needed
            //  根据所传元素判断为生产or消费
            int mode = (e == null) ? REQUEST : DATA;

            for (;;) { // 无限循环
                SNode h = head; // 获取头结点
                if (h == null || h.mode == mode) {  // 头结点为空或者当前节点状态(0,1,2)和头结点相同
                    if (timed && nanos <= 0) {      // can't wait 设置有时间
                        // 节点不为null并且为取消状态
                        if (h != null && h.isCancelled())
                            // 弹出取消的节点
                            casHead(h, h.next);     // pop cancelled node
                        else
                            // 超时直接返回null
                            return null;
                     // 没有设置超时
                    } else if (casHead(h, s = snode(s, e, h, mode))) { // 将h设为自己的next节点
                        // 空旋或者阻塞直到s结点被FulFill操作所匹配
                        SNode m = awaitFulfill(s, timed, nanos);
                        if (m == s) {                // wait was cancelled 节点被取消了
                            clean(s);
                            return null;
                        }
                        // 找到匹配的线程了
                        // h == head 可能已经已经被匹配
                        // h.next 等于s 不同类型
                        if ((h = head) != null && h.next == s)
                            // 弹出h 和 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 // m等于null
                                casHead(s, null);   // pop fulfill node // cas 设置head
                                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 // 失败,说明m背的线程匹配了,或者取消了
                                s.casNext(m, mn);   // help unlink //  修改next节点
                        }
                    }
                } 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
                    }
                }
            }
        }
awaitFulfill 方法
    SNode awaitFulfill(SNode s, boolean timed, long nanos) {

            // 计算时间
            final long deadline = timed ? System.nanoTime() + nanos : 0L;
            // 获取当前线程
            Thread w = Thread.currentThread();
            // shouldSpin控制自旋
            // shouldSpin 用于检测当前节点是否需要自旋
            // 如果栈为空、该节点是首节点或者该节点是匹配节点,则先采用自旋,否则阻塞
            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;
                    }
                }
                // 自旋;每次自旋的时候都需要检查自身是否满足自旋条件,满足就 - 1,否则为0
                if (spins > 0)
                    spins = shouldSpin(s) ? (spins-1) : 0;
                // 设置node的线程
                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);
            }
        }
clean 方法
    void clean(SNode s) {
            s.item = null;   // forget item
            s.waiter = null; // forget thread

            // next节点
            SNode past = s.next;
            // 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
            // 从有效的头节点开始p ,到s的后继节点,继续清除
            while (p != null && p != past) {
                SNode n = p.next;
                if (n != null && n.isCancelled())
                    p.casNext(n, n.next);
                else
                    p = n;
            }
        }

分析transfer 方法:

  • 我们可以发现transfer 是通过e是空来判断是offer方法还是poll方法的,也就是入队者和出对者的区分。

  • 第一种情况,如果队列为空head,或者队列存在的head节点和自己的模式相同,首先判断有没有超时或者取消,有就执行这些操作,没有就执行入队操作,然后把新加入的节点加入栈顶,然后调用awaitFulfill方法阻塞线程,直到被中断,超时或者匹配成功,为什么要阻塞了因为大家的模式都相同没法匹配,所以只能阻塞线程,直到一个不同模式的线程匹配成功,唤醒自己,这也是awaitFulfill方法的结束流程。

  • 第二种情况,如果入队的模式不同,通过isFulfilling方法判断head节点有没有在匹配,没有就执行匹配流程,

    首先判断节点是否被取消了,没有在判断自己是不是唯一的一个节点,如果是循环,重新开始流程,如果不是上面的这些情况,就可以开始匹配节点了,调用tryMatch方法,成功唤醒另一个节点,然后一起出栈,返回结果,失败就向后推进,找下一个节点,这里可能别的线程会竞争的匹配。

  • 第三种情况,入队的模式不同,但是head节点正在匹配,那就帮助它匹配完成,然后重新走整个循环。
TransferQueue 公平常用方法

入队和出队的方法是一样的,我们主要看下transfer 方法吧。

transfer 方法
    E transfer(E e, boolean timed, long nanos) {

            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
                // h==t 刚刚初始化 t.isData == isData,尾节点和当前节点的类型一样
                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;
                    // isData == (x != null) 判断模式是否相同,不相同才能匹配
                    // x == m 说明已经被中断或者超时了
                    // m.casItem(x, e) 匹配
                    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;
                }
            }
        }

分析transfer方法:

  • 也是通过e来判断是入队还是出队的,都是调用transfer方法,transfer方法可以看出两部分,入队和匹配。
  • 第一部分入队,入队的模式也是相同的,入队是从尾节点开始,获取尾节点,判断尾节点有没有发生变化,可能存在多线程的情况,发生改变就重新遍历,没有就判断尾节点有没有next节点,有就说明别的线程添加了新的节点,需要更新尾节点,然后构造新的节点加入当前尾节点的next节点,更新尾节点,然后调用awaitFulfill阻塞当前节点,直到中断,超时或者匹配,然后清除匹配成功的节点,调用clean方法。
  • 第二部分匹配(出队),出队是从头节点开始,然后判断模式是否不同,是否被取消,cas设置item,其实也就是数据的传递,如果匹配成功,唤醒等待在m的线程,这里注意把m设置成了头结点,其实就是把m节点弹出了,因为我们匹配取得头结点的next节点。
五,总结

SynchronousQueue 的实现还是很复杂的,我们可以发现虽然是个阻塞队列,可是没有使用锁;这个队列适合传递的场景,队列没有存储元素的队列,出队和入队必须结伴而行。

参考 《Java 并发编程的艺术》


分享名称:SynchronousQueue1.8源码解析
本文来源:http://myzitong.com/article/jcgeic.html