SynchronousQueue 1.8 源码解析

不念不忘少年蓝@ 2023-06-06 08:06 36阅读 0赞

#### 1、简介： ####

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

#### 2、UML ： ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5neW9uZzAxMjQ1_size_16_color_FFFFFF_t_70]

#### 2、基本成员： ####

abstract static class Transferer<E> {
        // 出队入队都是这一个方法
        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;

#### 4，常用方法 ####

构造方法

public SynchronousQueue() {
            this(false);
        }
    
        // 构造方法，fair表示公平或者非公平
        public SynchronousQueue(boolean fair) {
            transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
        }

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方法：

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

#### 5、总结 ####

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

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5neW9uZzAxMjQ1_size_16_color_FFFFFF_t_70]: /images/20230601/f1843f9b2106481ba65703ea18645120.png