【JUC源码】阻塞队列：SynchronousQueue 源码分析（公平模式）

╰+攻爆jí腚メ 2022-10-31 13:48 230阅读 0赞

> 阻塞队列系列：
> 
>  *  [【JUC源码】阻塞队列：继承关系及接口分析][JUC]
>  *  [【JUC源码】阻塞队列：ArrayBlockingQueue 源码分析][JUC_ArrayBlockingQueue]
>  *  [【JUC源码】阻塞队列：LinkedBlockingQueue 源码分析][JUC_LinkedBlockingQueue]
>  *  [【JUC源码】阻塞队列：DelayQueue 源码分析][JUC_DelayQueue]
>  *  [【JUC源码】阻塞队列：SynchronousQueue 源码分析（非公平模式）][JUC_SynchronousQueue]
>  *  [【JUC源码】阻塞队列：SynchronousQueue 源码分析（公平模式）][JUC_SynchronousQueue 1]
>  *  [【JUC源码】阻塞队列：总结 --特性对比&应用场景举例][JUC_ --]
>  *  [【JUC源码】阻塞队列：各种阻塞队列面试题合集][JUC 1]

在[上一篇][JUC_SynchronousQueue]文章，我们说了基于栈（FILO）的 TransferStack 的非公平线程控制逻辑，本篇我们就来看看基于队列（FIFO）的 TransferQueue 的公平的线程控制逻辑。

先来回顾一个问题，当前线程是如何把自己的数据传给阻塞线程的？

为了方便说明，我们假设线程 1 往队列中 take 数据 ，被阻塞住了，变成阻塞线程 A ，然后线程 2 开始往队列中 put 数据 B，大致的流程是这样的：

1.  线程 1 从队列中拿数据，发现队列中没有数据，于是被阻塞，成为 A ；
2.  线程 2 往队尾 put 数据，会从**队尾往前找到第一个被阻塞的节点**，假设此时能找到的就是节点 A，然后线程 B 把将 put 的数据放到节点 A 的 item 属性里面，并唤醒线程 1；
3.  线程 1 被唤醒后，就能从 A.item 里面拿到线程 2 put 的数据了，线程 1 成功返回。

从这个过程中，我们能看出公平主要体现在，每次 put 数据的时候，都 put 到队尾上，而每次拿数据时，并不是直接从队头拿数据，而是从队尾往前寻找第一个被阻塞的线程，这样就会按照顺序释放被阻塞的线程。

> PS：TransferStack 的线程匹配是基于栈的匹配离栈顶近的，所以不公平。

### TransferQueue(公平=>FIFO) ###

static final class TransferQueue<E> extends Transferer<E>{
        
        // 队列头 
        transient volatile QNode head;
        // 队列尾 
        transient volatile QNode tail;
    
        // 队列的元素
        static final class QNode {
        ...}
        
        //...
    }

### QNode ###

static final class QNode {
        
    
    	// 当前元素的值，如果当前元素被阻塞住了，等其他线程来唤醒自己时，其他线程会把自己 set 到 item 里面
        volatile Object item;         // CAS'ed to or from null
        // 可以阻塞住的当前线程
        volatile Thread waiter;       // to control park/unpark
         // true 是 put，false 是 take
        final boolean isData;
        
        // 当前元素的下一个元素
        volatile QNode next;         
    	
    	// 构造时传入数据和节点类型
    	QNode(Object item, boolean isData) {
        
    	    this.item = item;
    	    this.isData = isData;
    	}
    	
    	// 将节点val通过cas连接在cmp后面
    	boolean casNext(QNode cmp, QNode val) {
        
    	    return next == cmp &&
    	        UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
    	}
    	
    	// 将item的值通过cas变为val
    	boolean casItem(Object cmp, Object val) {
        
    	    return item == cmp &&
    	        UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
    	}
    	
    	/**
    	 * Tries to cancel by CAS'ing ref to this as item.
    	 */
    	void tryCancel(Object cmp) {
        
    	    UNSAFE.compareAndSwapObject(this, itemOffset, cmp, this);
    	}
    	
    	// 被取消的节点就是item=this
    	boolean isCancelled() {
        
    	    return item == this;
    	}
    	
    	// unsafe相关代码...
    }

### transfer()：入队&出队 ###

E transfer(E e, boolean timed, long nanos) {
        
    
        QNode s = null; // constructed/reused as needed
        // true 是 put，false 是 get
        boolean isData = (e != null);
    
        for (;;) {
        
            // 队列头和尾的临时变量,队列是空的时候，t=h
            QNode t = tail;
            QNode h = head;
            // tail 和 head 没有初始化时，无限循环
            // 虽然这种continue非常耗cpu，但一般碰不到这种情况，因为tail和head 在 TransferQueue 初始化时就已经被赋值空节点了
            if (t == null || h == null)
                continue;
    -----------------------------------------------------------------------------------------------------------------                        
            // 情况一：首尾节点相同（空队列）|| 尾节点的操作和当前节点操作一致（比如队尾是take时阻塞，当前线程也是take）
            if (h == t || t.isData == isData) {
        
                QNode tn = t.next;
                // 当 t 不是 tail 时。即 tail 已经被修改过了，因为 tail 没有被修改的情况下，t 和 tail 必然相等
                if (t != tail)
                    continue;
                // 队尾后面的值还不为空，t 还不是队尾，直接把 tn 赋值给 t，这是一步加强校验。
                if (tn != null) {
        
                	// CAS修改tail为tn
                    advanceTail(t, tn);
                    continue;
                }
                // 超时直接返回 null
                if (timed && nanos <= 0)        // can't wait
                    return null;
                // 构造node节点
                if (s == null)
                    s = new QNode(e, isData);
                // 如果把 e 放到队尾失败，继续递归放进去
                if (!t.casNext(null, s))        // failed to link in
                    continue;
    
                advanceTail(t, s);              // swing tail and wait
                // awaitFulfill 同 TransferStack，阻塞住自己，等待配对节点x
                Object x = awaitFulfill(s, e, timed, nanos);
                if (x == s) {
                           // wait was cancelled
                    clean(t, s);
                    return null;
                }
    
                if (!s.isOffList()) {
           // not already unlinked
                	// CAS修改head为s     
                    advanceHead(t, s);  // unlink if head
                    if (x != null)      // and forget fields
                        s.item = s;
                    s.waiter = null;
                }
                return (x != null) ? (E)x : e;
    -----------------------------------------------------------------------------------------------------------------            
            // 情况二：队列不为空，并且当前操作和队尾不一致（比如队尾是因为 take 被阻塞的，那么当前操作必然是 put）
            } else {
                                    // complementary-mode
                // 如果是第一次执行，此处的 m 代表就是 tail
                // 也就是这行代码体现出队列的公平，每次操作时，从头开始按照顺序进行操作
                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 代表栈头
                    // 这里把当前的操作值赋值给阻塞住的 m 的 item 属性，所以 m 被释放时，就可得到此次操作的值
                    !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;
            }
        }
    }

[JUC]: https://blog.csdn.net/weixin_43935927/article/details/113952868
[JUC_ArrayBlockingQueue]: https://blog.csdn.net/weixin_43935927/article/details/108833582
[JUC_LinkedBlockingQueue]: https://blog.csdn.net/weixin_43935927/article/details/108853656
[JUC_DelayQueue]: https://blog.csdn.net/weixin_43935927/article/details/108858129
[JUC_SynchronousQueue]: https://blog.csdn.net/weixin_43935927/article/details/108866902
[JUC_SynchronousQueue 1]: https://blog.csdn.net/weixin_43935927/article/details/113964294
[JUC_ --]: https://blog.csdn.net/weixin_43935927/article/details/108888294
[JUC 1]: https://blog.csdn.net/weixin_43935927/article/details/108857735