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

阻塞队列系列：

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

在上一篇文章，我们说了基于栈（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;
        }
    }
}