一、前言
本篇是在分析Executors源码时，发现JUC集合框架中的一个重要类没有分析，SynchronousQueue，该类在线程池中的作用是非常明显的，所以很有必要单独拿出来分析一番，这对于之后理解线程池有很有好处，SynchronousQueue是一种阻塞队列，其中每个插入操作必须等待另一个线程的对应移除操作 ，反之亦然。同步队列没有任何内部容量，甚至连一个队列的容量都没有。
二、SynchronousQueue数据结构
由于SynchronousQueue的支持公平策略和非公平策略，所以底层可能两种数据结构：队列（实现公平策略）和栈（实现非公平策略），队列与栈都是通过链表来实现的。具体的数据结构如下
说明：数据结构有两种类型，栈和队列；栈有一个头结点，队列有一个头结点和尾结点；栈用于实现非公平策略，队列用于实现公平策略。
三、SynchronousQueue源码分析
3.1 类的继承关系
public class SynchronousQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {}
说明：SynchronousQueue继承了AbstractQueue抽象类，AbstractQueue定义了对队列的基本操作；同时实现了BlockingQueue接口，BlockingQueue表示阻塞型的队列，其对队列的操作可能会抛出异常；同时也实现了Searializable接口，表示可以被序列化。
3.2 类的内部类
SynchronousQueue的内部类框架图如下
说明：其中比较重要的类是左侧的三个类，Transferer是TransferStack栈和TransferQueue队列的公共类，定义了转移数据的公共操作，由TransferStack和TransferQueue具体实现，WaitQueue、LifoWaitQueue、FifoWaitQueue表示为了兼容JDK1.5版本中的SynchronousQueue的序列化策略所遗留的，这里不做具体的讲解。下面着重看左侧的三个类。
① Transferer
abstract static class Transferer<E> {
/**
* Performs a put or take.
*
* @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 nanoseconds
* @return if non-null, the item provided or received; if null,
* the operation failed due to timeout or interrupt --
* the caller can distinguish which of these occurred
* by checking Thread.interrupted.
*/
// 转移数据，put或者take操作
abstract E transfer(E e, boolean timed, long nanos);
}
说明：Transferer定义了transfer操作，用于take或者put数据。transfer方法由子类实现。
② TransfererStack
1. 类的继承关系
static final class TransferStack<E> extends Transferer<E> {}
说明：TransferStack继承Transferer抽象类，其实现了transfer方法。
2. 类的属性
static final class TransferStack<E> extends Transferer<E> {
/*
* This extends Scherer-Scott dual stack algorithm, differing,
* among other ways, by using "covering" nodes rather than
* bit-marked pointers: Fulfilling operations push on marker
* nodes (with FULFILLING bit set in mode) to reserve a spot
* to match a waiting node.
*/
/* Modes for SNodes, ORed together in node fields */
/** Node represents an unfulfilled consumer */
// 表示消费数据的消费者
static final int REQUEST = 0;
/** Node represents an unfulfilled producer */
// 表示生产数据的生产者
static final int DATA = 1;
/** Node is fulfilling another unfulfilled DATA or REQUEST */
// 表示匹配另一个生产者或消费者
static final int FULFILLING = 2;
/** The head (top) of the stack */
// 头结点
volatile SNode head;
}
说明：TransferStack有三种不同的状态，REQUEST，表示消费数据的消费者；DATA，表示生产数据的生产者；FULFILLING，表示匹配另一个生产者或消费者。任何线程对TransferStack的操作都属于上述3种状态中的一种。同时还包含一个head域，表示头结点。
3. 类的内部类
SNode类
1. 类的属性
static final class SNode {
// 下一个结点
volatile SNode next; // next node in stack
// 相匹配的结点
volatile SNode match; // the node matched to this
// 等待的线程
volatile Thread waiter; // to control park/unpark
// 元素项
Object item; // data; or null for REQUESTs
// 模式
int mode;
// Note: item and mode fields don‘t need to be volatile
// since they are always written before, and read after,
// other volatile/atomic operations.
// item域和mode域不需要使用volatile修饰，因为它们在volatile/atomic操作之前写，之后读
// Unsafe mechanics
// 反射机制
private static final sun.misc.Unsafe UNSAFE;
// match域的内存偏移地址
private static final long matchOffset;
// next域的偏移地址
private static final long nextOffset;
static {
try {
UNSAFE = sun.misc.Unsafe.getUnsafe();
Class<?> k = SNode.class;
matchOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("match"));
nextOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("next"));
} catch (Exception e) {
throw new Error(e);
}
}
}
说明：SNode类表示栈中的结点，使用了反射机制和CAS来保证原子性的改变相应的域值。
2. 类的构造函数
SNode(Object item) {
this.item = item;
}
说明：该构造函数仅仅设置了SNode的item域，其他域为默认值。
3. 核心函数分析
3.1. tryMatch函数
boolean tryMatch(SNode s) {
if (match == null &&
UNSAFE.compareAndSwapObject(this, matchOffset, null, s)) { // 本结点的match域为null并且比较并替换match域成功
// 获取本节点的等待线程
Thread w = waiter;
if (w != null) { // 存在等待的线程 // waiters need at most one unpark
// 将本结点的等待线程重新置为null
waiter = null;
// unpark等待线程
LockSupport.unpark(w);
}
return true;
}
// 如果match不为null或者CAS设置失败，则比较match域是否等于s结点，若相等，则表示已经完成匹配，匹配成功
return match == s;
}
说明：将s结点与本结点进行匹配，匹配成功，则unpark等待线程。具体流程如下
① 判断本结点的match域是否为null，若为null，则进入步骤②，否则，进入步骤⑤
② CAS设置本结点的match域为s结点，若成功，则进入步骤③，否则，进入步骤⑤
③ 判断本结点的waiter域是否为null，若不为null，则进入步骤④，否则，进入步骤⑤
④ 重新设置本结点的waiter域为null，并且unparkwaiter域所代表的等待线程。进入步骤⑥
⑤ 比较本结点的match域是否为本结点，若是，则进入步骤⑥，否则，进入步骤⑦
⑥ 返回true
⑦ 返回false
4. 核心函数分析
4.1 isFulfilling函数
static boolean isFulfilling(int m) { return (m & FULFILLING) != 0; }
说明：表示是否包含FULFILLING标记。
4.2 transfer函数
E transfer(E e, boolean timed, long nanos) {
/*
* Basic algorithm is to loop trying one of three actions:
*
* 1. If apparently empty or already containing nodes of same
* mode, try to push node on stack and wait for a match,
* returning it, or null if cancelled.
*
* 2. If apparently containing node of complementary mode,
* try to push a fulfilling node on to stack, match
* with corresponding waiting node, pop both from
* stack, and return matched item. The matching or
* unlinking might not actually be necessary because of
* other threads performing action 3:
*
* 3. If top of stack already holds another fulfilling node,
* help it out by doing its match and/or pop
* operations, and then continue. The code for helping
* is essentially the same as for fulfilling, except
* that it doesn‘t return the item.
*/
SNode s = null; // constructed/reused as needed
// 根据e确定此次转移的模式（是put or take）
int mode = (e == null) ? REQUEST : DATA;
for (;;) { // 无限循环
// 保存头结点
SNode h = head;
if (h == null || h.mode == mode) { // 头结点为null或者头结点的模式与此次转移的模式相同 // empty or same-mode
if (timed && nanos <= 0) { // 设置了timed并且等待时间小于等于0，表示不能等待，需要立即操作 // can‘t wait
if (h != null && h.isCancelled()) // 头结点不为null并且头结点被取消
casHead(h, h.next); // 重新设置头结点（弹出之前的头结点） // pop cancelled node
else // 头结点为null或者头结点没有被取消
// 返回null
return null;
} else if (casHead(h, s = snode(s, e, h, mode))) { // 生成一个SNode结点；将原来的head头结点设置为该结点的next结点；将head头结点设置为该结点
// Spins/blocks until node s is matched by a fulfill operation.
// 空旋或者阻塞直到s结点被FulFill操作所匹配
SNode m = awaitFulfill(s, timed, nanos);
if (m == s) { // 匹配的结点为s结点（s结点被取消） // wait was cancelled
// 清理s结点
clean(s);
// 返回
return null;
}
if ((h = head) != null && h.next == s) // h重新赋值为head头结点，并且不为null；头结点的next域为s结点，表示有结点插入到s结点之前，完成了匹配
// 比较并替换head域（移除插入在s之前的结点和s结点）
casHead(h, s.next); // help s‘s fulfiller
// 根据此次转移的类型返回元素
return (E) ((mode == REQUEST) ? m.item : s.item);
}
} else if (!isFulfilling(h.mode)) { // 没有FULFILLING标记，尝试匹配 // try to fulfill
if (h.isCancelled()) // 被取消 // already cancelled
// 比较并替换head域（弹出头结点）
casHead(h, h.next); // pop and retry
else if (casHead(h, s=snode(s, e, h, FULFILLING|mode))) { // 生成一个SNode结点；将原来的head头结点设置为该结点的next结点；将head头结点设置为该结点
for (;;) { // 无限循环 // loop until matched or waiters disappear
// 保存s的next结点
SNode m = s.next; // m is s‘s match
if (m == null) { // next域为null // all waiters are gone
// 比较并替换head域
casHead(s, null); // pop fulfill node
// 赋值s为null
s = null; // use new node next time
break; // restart main loop
}
// m结点的next域
SNode mn = m.next;
if (m.tryMatch(s)) { // 尝试匹配，并且成功
// 比较并替换head域（弹出s结点和m结点）
casHead(s, mn); // pop both s and m
// 根据此次转移的类型返回元素
return (E) ((mode == REQUEST) ? m.item : s.item);
} else // 匹配不成功 // lost match
// 比较并替换next域（弹出m结点）
s.casNext(m, mn); // help unlink
}
}
} else { // 头结点正在匹配 // help a fulfiller
// 保存头结点的next域
SNode m = h.next; // m与h可以匹配 // m is h‘s match
if (m == null) // next域为null // waiter is gone
// 比较并替换head域（m被其他结点匹配了，需要弹出h）
casHead(h, null); // pop fulfilling node
else { // next域不为null
// 获取m结点的next域
SNode mn = m.next;
if (m.tryMatch(h)) // m与h匹配成功 // help match
// 比较并替换head域（弹出h和m结点）
casHead(h, mn); // pop both h and m
else // 匹配不成功 // lost match
// 比较并替换next域（移除m结点）
h.casNext(m, mn); // help unlink
}
}
}
}
说明：此函数用于生产或者消费一个元素，并且transfer函数调用了awaitFulfill函数，之后会通过一个例子给出流程。
4.3 awaitFulfill函数
SNode awaitFulfill(SNode s, boolean timed, long nanos) {
/*
* When a node/thread is about to block, it sets its waiter
* field and then rechecks state at least one more time
* before actually parking, thus covering race vs
* fulfiller noticing that waiter is non-null so should be
* woken.
*
* When invoked by nodes that appear at the point of call
* to be at the head of the stack, calls to park are
* preceded by spins to avoid blocking when producers and
* consumers are arriving very close in time. This can
* happen enough to bother only on multiprocessors.
*
* The order of checks for returning out of main loop
* reflects fact that interrupts have precedence over
* normal returns, which have precedence over
* timeouts. (So, on timeout, one last check for match is
* done before giving up.) Except that calls from untimed
* SynchronousQueue.{poll/offer} don‘t check interrupts
* and don‘t wait at all, so are trapped in transfer
* method rather than calling awaitFulfill.
*/
// 根据timed标识计算截止时间
final long deadline = timed ? System.nanoTime() + nanos : 0L;
// 获取当前线程
Thread w = Thread.currentThread();
// 根据s确定空旋等待的时间
int spins = (shouldSpin(s) ?
(timed ? maxTimedSpins : maxUntimedSpins) : 0);
for (;;) { // 无限循环，确保操作成功
if (w.isInterrupted()) // 当前线程被中断
// 取消s结点
s.tryCancel();
// 获取s结点的match域
SNode m = s.match;
if (m != null) // m不为null，存在匹配结点
// 返回m结点
return m;
if (timed) { // 设置了timed
// 确定继续等待的时间
nanos = deadline - System.nanoTime();
if (nanos <= 0L) { // 继续等待的时间小于等于0，等待超时
// 取消s结点
s.tryCancel();
// 跳过后面的部分，继续
continue;
}
}
if (spins > 0) // 空旋等待的时间大于0
// 确实是否还需要继续空旋等待
spins = shouldSpin(s) ? (spins-1) : 0;
else if (s.waiter == null) // 等待线程为null
// 设置waiter线程为当前线程
s.waiter = w; // establish waiter so can park next iter
else if (!timed) // 没有设置timed标识
// 禁用当前线程并设置了阻塞者
LockSupport.park(this);
else if (nanos > spinForTimeoutThreshold) // 继续等待的时间大于阈值
// 禁用当前线程，最多等待指定的等待时间，除非许可可用
LockSupport.parkNanos(this, nanos);
}
}
说明：此函数表示当前线程自旋或阻塞，直到结点被匹配。awaitFulfill函数调用了shouldSpin函数
4.4 shouldSpin函数
boolean shouldSpin(SNode s) {
// 获取头结点
SNode h = head;
// s为头结点或者头结点为null或者h包含FULFILLING标记,返回true
return (h == s || h == null || isFulfilling(h.mode));
}
说明：此函数表示是当前结点所包含的线程（当前线程）进行空旋等待，有如下情况需要进行空旋等待
① 当前结点为头结点
② 头结点为null
③ 头结点正在匹配中
4.5 clean函数
void clean(SNode s) {
// s结点的item设置为null
s.item = null; // forget item
// waiter域设置为null
s.waiter = null; // forget thread
/*
* At worst we may need to traverse entire stack to unlink
* s. If there are multiple concurrent calls to clean, we
* might not see s if another thread has already removed
* it. But we can stop when we see any node known to
* follow s. We use s.next unless it too is cancelled, in
* which case we try the node one past. We don‘t check any
* further because we don‘t want to doubly traverse just to
* find sentinel.
*/
// 获取s结点的next域
SNode past = s.next;
if (past != null && past.isCancelled()) // next域不为null并且next域被取消
// 重新设置past
past = past.next;
// Absorb cancelled nodes at head
SNode p;
while ((p = head) != null && p != past && p.isCancelled()) // 从栈顶头结点开始到past结点（不包括），将连续的取消结点移除
// 比较并替换head域（弹出取消的结点）
casHead(p, p.next);
// Unsplice embedded nodes
while (p != null && p != past) { // 移除上一步骤没有移除的非连续的取消结点
// 获取p的next域
SNode n = p.next;
if (n != null && n.isCancelled()) // n不为null并且n被取消
// 比较并替换next域
p.casNext(n, n.next);
else
// 设置p为n
p = n;
}
}
说明：此函数用于移除从栈顶头结点开始到该结点（不包括）之间的所有已取消结点。
③ TransferQueue
1. 类的继承关系
static final class TransferQueue<E> extends Transferer<E> {}
说明：TransferQueue继承Transferer抽象类，其实现了transfer方法。
2. 类的属性
static final class TransferQueue<E> extends Transferer<E> {
/*
* This extends Scherer-Scott dual queue algorithm, differing,
* among other ways, by using modes within nodes rather than
* marked pointers. The algorithm is a little simpler than
* that for stacks because fulfillers do not need explicit
* nodes, and matching is done by CAS‘ing QNode.item field
* from non-null to null (for put) or vice versa (for take).
*/
/** Head of queue */
// 队列的头结点
transient volatile QNode head;
/** Tail of queue */
// 队列的尾结点
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;
}
说明：队列存在一个头结点和一个尾节点，分别指示队头和队尾，还包含了一个指示取消结点的域。
3. 类的内部类
QNode类
QNode的源码如下
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
// 是否为数据
final boolean isData;
// 构造函数
QNode(Object item, boolean isData) {
// 初始化item域
this.item = item;
// 初始化isData域
this.isData = isData;
}
// 比较并替换next域
boolean casNext(QNode cmp, QNode val) {
return next == cmp &&
UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
}
// 比较并替换item域
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.
*/
// 取消本结点，将item域设置为自身
void tryCancel(Object cmp) {
UNSAFE.compareAndSwapObject(this, itemOffset, cmp, this);
}
// 是否被取消
boolean isCancelled() {
// item域是否等于自身
return item == this;
}
/**
* Returns true if this node is known to be off the queue
* because its next pointer has been forgotten due to
* an advanceHead operation.
*/
// 是否不在队列中
boolean isOffList() {
// next与是否等于自身
return next == this;
}
// Unsafe mechanics
// 反射机制
private static final sun.misc.Unsafe UNSAFE;
// item域的偏移地址
private static final long itemOffset;
// next域的偏移地址
private static final long nextOffset;
static {
try {
UNSAFE = sun.misc.Unsafe.getUnsafe();
Class<?> k = QNode.class;
itemOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("item"));
nextOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("next"));
} catch (Exception e) {
throw new Error(e);
}
}
}
说明：QNode表示队列中的结点，并且通过反射和CAS原子性的修改对应的域值。
4. 类的构造函数
TransferQueue() {
// 初始化一个哨兵结点
QNode h = new QNode(null, false); // initialize to dummy node.
// 设置头结点
head = h;
// 设置尾结点
tail = h;
}
说明：该构造函数用于初始化一个队列，并且初始化了一个哨兵结点，头结点与尾节点均指向该哨兵结点。
5. 核心函数分析
5.1 transfer函数
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 and try to help
* advance head and tail on behalf of other stalled/slow
* threads.
*
* The loop starts off with a null check guarding against
* seeing uninitialized head or tail values. 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 implicitly interspersed.
*/
QNode s = null; // constructed/reused as needed
// 确定此次转移的类型（put or take）
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
// 获取尾结点的next域
QNode tn = t.next;
if (t != tail) // t不为尾结点，不一致，重试 // inconsistent read
continue;
if (tn != null) { // tn不为null，有其他线程添加了tn结点 // lagging tail
// 设置新的尾结点为tn
advanceTail(t, tn);
// 跳过后面的部分，继续
continue;
}
if (timed && nanos <= 0) // 设置了timed并且等待时间小于等于0，表示不能等待，需要立即操作 // can‘t wait
// 返回null
return null;
if (s == null) // s为null
// 新生一个结点并赋值给s
s = new QNode(e, isData);
if (!t.casNext(null, s)) // 设置t结点的next域不成功 // failed to link in
// 跳过后面的部分，继续
continue;
// 设置新的尾结点
advanceTail(t, s); // swing tail and wait
// Spins/blocks until node s is fulfilled
// 空旋或者阻塞直到s结点被匹配
Object x = awaitFulfill(s, e, timed, nanos);
if (x == s) { // x与s相等，表示已经取消 // wait was cancelled
// 清除
clean(t, s);
// 返回null
return null;
}
if (!s.isOffList()) { // s结点还没离开队列 // not already unlinked
// 设置新的头结点
advanceHead(t, s); // unlink if head
if (x != null) // x不为null // and forget fields
// 设置s结点的item
s.item = s;
// 设置s结点的waiter域为null
s.waiter = null;
}
return (x != null) ? (E)x : e;
} else { // 模式互补 // complementary-mode
// 获取头结点的next域（匹配的结点）
QNode m = h.next; // node to fulfill
if (t != tail || m == null || h != head) // t不为尾结点或者m为null或者h不为头结点（不一致）
// 跳过后面的部分，继续
continue; // inconsistent read
// 获取m结点的元素域
Object x = m.item;
if (isData == (x != null) || // m结点被匹配 // m already fulfilled
x == m || // m结点被取消 // m cancelled
!m.casItem(x, e)) { // CAS操作失败 // lost CAS
advanceHead(h, m); // 队列头结点出队列，并重试 // dequeue and retry
continue;
}
// 匹配成功，设置新的头结点
advanceHead(h, m); // successfully fulfilled
// unpark m结点对应的等待线程
LockSupport.unpark(m.waiter);
return (x != null) ? (E)x : e;
}
}
}
说明：此函数用于生产或者消费一个元素，并且transfer函数调用了awaitFulfill函数，之后会通过一个例子给出流程。
5.2 awaitFulfill函数
Object awaitFulfill(QNode s, E e, boolean timed, long nanos) {
/* Same idea as TransferStack.awaitFulfill */
// 根据timed标识计算截止时间
final long deadline = timed ? System.nanoTime() + nanos : 0L;
// 获取当前线程
Thread w = Thread.currentThread();
// 计算空旋时间
int spins = ((head.next == s) ?
(timed ? maxTimedSpins : maxUntimedSpins) : 0);
for (;;) { // 无限循环，确保操作成功
if (w.isInterrupted()) // 当前线程被中断
// 取消
s.tryCancel(e);
// 获取s的元素域
Object x = s.item;
if (x != e) // 元素不为e
// 返回
return x;
if (timed) { // 设置了timed
// 计算继续等待的时间
nanos = deadline - System.nanoTime();
if (nanos <= 0L) { // 继续等待的时间小于等于0
// 取消
s.tryCancel(e);
// 跳过后面的部分，继续
continue;
}
}
if (spins > 0) // 空旋时间大于0
// 减少空旋时间
--spins;
else if (s.waiter == null) // 等待线程为null
// 设置等待线程
s.waiter = w;
else if (!timed) // 没有设置timed标识
// 禁用当前线程并设置了阻塞者
LockSupport.park(this);
else if (nanos > spinForTimeoutThreshold) // 继续等待的时间大于阈值
// 禁用当前线程，最多等待指定的等待时间，除非许可可用
LockSupport.parkNanos(this, nanos);
}
}
说明：此函数表示当前线程自旋或阻塞，直到结点被匹配。
5.3 clean函数
void clean(QNode pred, QNode s) {
// 设置等待线程为null
s.waiter = null; // forget thread
/*
* At any given time, exactly one node on list cannot be
* deleted -- the last inserted node. To accommodate this,
* if we cannot delete s, we save its predecessor as
* "cleanMe", deleting the previously saved version
* first. At least one of node s or the node previously
* saved can always be deleted, so this always terminates.
*/
/*
* 在任何时候，最后插入的结点不能删除，为了满足这个条件
* 如果不能删除s结点，我们将s结点的前驱设置为cleanMe结点
* 删除之前保存的版本，至少s结点或者之前保存的结点能够被删除
* 所以最后总是会结束
*/
while (pred.next == s) { // pred的next域为s // Return early if already unlinked
// 获取头结点
QNode h = head;
// 获取头结点的next域
QNode hn = h.next; // Absorb cancelled first node as head
if (hn != null && hn.isCancelled()) { // hn不为null并且hn被取消
// 设置新的头结点
advanceHead(h, hn);
// 跳过后面的部分，继续
continue;
}
// 获取尾结点，保证对尾结点的读一致性
QNode t = tail; // Ensure consistent read for tail
if (t == h) // 尾结点为头结点，表示队列为空
// 返回
return;
// 获取尾结点的next域
QNode tn = t.next;
if (t != tail) // t不为尾结点，不一致，重试
// 跳过后面的部分，继续
continue;
if (tn != null) { // tn不为null
// 设置新的尾结点
advanceTail(t, tn);
// 跳过后面的部分，继续
continue;
}
if (s != t) { // s不为尾结点，移除s // If not tail, try to unsplice
QNode sn = s.next;
if (sn == s || pred.casNext(s, sn)) //
return;
}
// 获取cleanMe结点
QNode dp = cleanMe;
if (dp != null) { // dp不为null，断开前面被取消的结点 // Try unlinking previous cancelled node
// 获取dp的next域
QNode d = dp.next;
QNode dn;
if (d == null || // d is gone or
d == dp || // d is off list or
!d.isCancelled() || // d not cancelled or
(d != t && // d not tail and
(dn = d.next) != null && // has successor
dn != d && // that is on list
dp.casNext(d, dn))) // d unspliced
casCleanMe(dp, null);
if (dp == pred)
return; // s is already saved node
} else if (casCleanMe(null, pred))
return; // Postpone cleaning s
}
}
说明：此函数用于移除已经被取消的结点。
3.3 类的属性
public class SynchronousQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
// 版本序列号
private static final long serialVersionUID = -3223113410248163686L;
// 可用的处理器
static final int NCPUS = Runtime.getRuntime().availableProcessors();
// 最大空旋时间
static final int maxTimedSpins = (NCPUS < 2) ? 0 : 32;
// 无限时的等待的最大空旋时间
static final int maxUntimedSpins = maxTimedSpins * 16;
// 超时空旋等待阈值
static final long spinForTimeoutThreshold = 1000L;
// 用于序列化
private ReentrantLock qlock;
private WaitQueue waitingProducers;
private WaitQueue waitingConsumers;
}
说明：SynchronousQueue类的属性包含了空旋等待时间相关的属性。
3.4 类的构造函数
1. SynchronousQueue()型构造函数
public SynchronousQueue() {
// 非公平策略（先进后出）
this(false);
}
说明：该构造函数用于创建一个具有非公平访问策略的 SynchronousQueue。
2. SynchronousQueue(boolean)型构造函数
public SynchronousQueue(boolean fair) {
// 根据指定的策略生成不同的结构
transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
}
说明：创建一个具有指定公平策略的 SynchronousQueue。
3.5 核心函数分析
在分析了TransferStack和TransferQueue的相关函数后，SynchronousQueue的函数的分析就非常简单。
// 将指定元素添加到此队列，如有必要则等待另一个线程接收它
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();
}
}
// 将指定元素插入到此队列，如有必要则等待指定的时间，以便另一个线程接收它
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();
}
// 如果另一个线程正在等待以便接收指定元素，则将指定元素插入到此队列
public boolean offer(E e) {
// e为null则抛出异常
if (e == null) throw new NullPointerException();
return transferer.transfer(e, true, 0) != null; // 进行转移操作
}
// 获取并移除此队列的头，如有必要则等待另一个线程插入它
public E take() throws InterruptedException {
// 进行转移操作
E e = transferer.transfer(null, false, 0);
if (e != null)
return e;
Thread.interrupted();
throw new InterruptedException();
}
// 获取并移除此队列的头，如有必要则等待指定的时间，以便另一个线程插入它
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
E e = transferer.transfer(null, true, unit.toNanos(timeout));
if (e != null || !Thread.interrupted()) // 元素不为null或者当前线程没有被中断
return e;
throw new InterruptedException();
}
// 如果另一个线程当前正要使用某个元素，则获取并移除此队列的头
public E poll() {
return transferer.transfer(null, true, 0);
}
// 始终返回 true
public boolean isEmpty() {
return true;
}
// 始终返回 0
public int size() {
return 0;
}
// 始终返回 0
public int remainingCapacity() {
return 0;
}
// 不执行任何操作
public void clear() {
}
// 始终返回false
public boolean contains(Object o) {
return false;
}
// 始终返回false
public boolean remove(Object o) {
return false;
}
// 除非给定 collection 为空，否则返回 false
public boolean containsAll(Collection<?> c) {
return c.isEmpty();
}
// 始终返回 false
public boolean removeAll(Collection<?> c) {
return false;
}
// 始终返回 false
public boolean retainAll(Collection<?> c) {
return false;
}
// 始终返回 null
public E peek() {
return null;
}
// 返回一个空迭代器，其中 hasNext 始终返回 false
public Iterator<E> iterator() {
return Collections.emptyIterator();
}
//
public Spliterator<E> spliterator() {
return Spliterators.emptySpliterator();
}
// 返回一个 0 长度的数组
public Object[] toArray() {
return new Object[0];
}
// 将指定数组的第 0 个元素设置为 null（如果该数组有非 0 的长度）并返回它
public <T> T[] toArray(T[] a) {
if (a.length > 0)
a[0] = null;
return a;
}
// 移除此队列中所有可用的元素，并将它们添加到给定 collection 中
public int drainTo(Collection<? super E> c) {
if (c == null)
throw new NullPointerException();
if (c == this)
throw new IllegalArgumentException();
int n = 0;
for (E e; (e = poll()) != null;) {
c.add(e);
++n;
}
return n;
}
// 最多从此队列中移除给定数量的可用元素，并将这些元素添加到给定 collection 中
public int drainTo(Collection<? super E> c, int maxElements) {
if (c == null)
throw new NullPointerException();
if (c == this)
throw new IllegalArgumentException();
int n = 0;
for (E e; n < maxElements && (e = poll()) != null;) {
c.add(e);
++n;
}
return n;
}
说明：SynchronousQueue的函数很大程度都是依托于TransferStack或TransferQueue的transfer函数，所以，了解transfer函数就可以了解SynchronousQueue的原理。
四、示例
下面通过一个示例来详细了解SynchronousQueue的使用。
package com.hust.grid.leesf.collections;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
public class SynchronousQueueDemo {
public static void main(String[] args) {
SynchronousQueue<Integer> queue = new SynchronousQueue<Integer>();
Producer p1 = new Producer("p1", queue, 10);
Producer p2 = new Producer("p2", queue, 50);
Consumer c1 = new Consumer("c1", queue);
Consumer c2 = new Consumer("c2", queue);
c1.start();
try {
TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
c2.start();
try {
TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
p1.start();
try {
TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
p2.start();
}
static class Producer extends Thread {
private SynchronousQueue<Integer> queue;
private int n;
public Producer(String name, SynchronousQueue<Integer> queue, int n) {
super(name);
this.queue = queue;
this.n = n;
}
public void run() {
System.out.println(getName() + " offer result " + queue.offer(n));
}
}
static class Consumer extends Thread {
private SynchronousQueue<Integer> queue;
public Consumer(String name, SynchronousQueue<Integer> queue) {
super(name);
this.queue = queue;
}
public void run() {
try {
System.out.println(getName() + " take result " + queue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
运行结果（某一次）
p1 offer result true c2 take result 10 p2 offer result true c1 take result 50
说明：该示例中，有两个生产者p1、p2和两个消费者c1、c2，按照c1、c2、p1、p2的顺序启动，并且每个线程启动后休眠100ms，则可能有如下的时序图
说明：时序图中，c1线程的take操作早于c2线程的take操作早于p1线程的offer操作早于p2线程的offer操作。
根据示例源码可知，此SynchronousQueue采用非公平策略，即底层采用栈结构。
① c1执行take操作，主要的函数调用如下
说明：其中，c1线程进入awaitFulfill后，会空旋等待，直到空旋时间消逝，会调用LockSupport.park函数，会禁用当前线程（c1），直至许可可用。
② c1执行take操作，主要的函数调用如下
说明：其中，c2线程进入awaitFulfill后，会空旋等待，直到空旋时间消逝，会调用LockSupport.park函数，会禁用当前线程（c2），直至许可可用。并且此时栈中有两个节点，c2线程所在的结点和c1线程所在的结点。
③ p1线程执行offer(10)操作，主要的函数调用如下
说明：在执行offer(10)操作后，c2线程所在的结点与头结点进行了匹配（头结点生产数据，c2线程所在的结点消费数据），c2线程被unpark，可以继续运行，而c1线程还是被park中（非公平策略）。
③ c2线程被unpark后，继续运行，主要函数调用如下（由于c2线程是在awaitFulfill函数中被park的，所以，恢复也是在awaitFulfill函数中）
说明：c2线程从unpark恢复时，结构如上图所示，先从awaitFulfill函数中返回，然后再从transfer函数中返回10，再从take函数中返回10。
④ p2线程执行offer(50)操作，主要的函数调用如下
说明：在执行offer(50)操作后，c1线程所在的结点与头结点进行了匹配（头结点生产数据，c1线程所在的结点消费数据），c1线程被unpark，可以继续运行。
⑤ c1线程被unpark后，继续运行，主要函数调用如下（由于c1线程是在awaitFulfill函数中被park的，所以，恢复也是在awaitFulfill函数中）
说明：c1线程从unpark恢复时，结构如上图所示，先从awaitFulfill函数中返回，然后再从transfer函数中返回50，再从take函数中返回50。
上述是使用非公平策略的结果（首先匹配c2线程所在的结点，之后再匹配c1线程所在结点）。
修改示例，改用公平策略。
package com.hust.grid.leesf.collections;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
public class SynchronousQueueDemo {
public static void main(String[] args) {
SynchronousQueue<Integer> queue = new SynchronousQueue<Integer>(true);
Producer p1 = new Producer("p1", queue, 10);
Producer p2 = new Producer("p2", queue, 50);
Consumer c1 = new Consumer("c1", queue);
Consumer c2 = new Consumer("c2", queue);
c1.start();
try {
TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
c2.start();
try {
TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
p1.start();
try {
TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
p2.start();
}
static class Producer extends Thread {
private SynchronousQueue<Integer> queue;
private int n;
public Producer(String name, SynchronousQueue<Integer> queue, int n) {
super(name);
this.queue = queue;
this.n = n;
}
public void run() {
System.out.println(getName() + " offer result " + queue.offer(n));
}
}
static class Consumer extends Thread {
private SynchronousQueue<Integer> queue;
public Consumer(String name, SynchronousQueue<Integer> queue) {
super(name);
this.queue = queue;
}
public void run() {
try {
System.out.println(getName() + " take result " + queue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
运行结果（某一次）
p1 offer result true c1 take result 10 p2 offer result true c2 take result 50
说明：从运行结果可知，c1线程会比c2线程先匹配（因为采用公平策略，先入队列先匹配，所以c1先得到匹配，然后再匹配c2）。具体的流程图与非公平策略类似，在此不再累赘。
当再次修改源码，还是使用非公平策略，只是改变c1、c2、p1、p2之间的启动顺序。更改为p1->c1->p2->c2。
package com.hust.grid.leesf.collections;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
public class SynchronousQueueDemo {
public static void main(String[] args) {
SynchronousQueue<Integer> queue = new SynchronousQueue<Integer>();
Producer p1 = new Producer("p1", queue, 10);
Producer p2 = new Producer("p2", queue, 50);
Consumer c1 = new Consumer("c1", queue);
Consumer c2 = new Consumer("c2", queue);
p1.start();
try {
TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
c1.start();
try {
TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
p2.start();
try {
TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
c2.start();
}
static class Producer extends Thread {
private SynchronousQueue<Integer> queue;
private int n;
public Producer(String name, SynchronousQueue<Integer> queue, int n) {
super(name);
this.queue = queue;
this.n = n;
}
public void run() {
System.out.println(getName() + " offer result " + queue.offer(n));
}
}
static class Consumer extends Thread {
private SynchronousQueue<Integer> queue;
public Consumer(String name, SynchronousQueue<Integer> queue) {
super(name);
this.queue = queue;
}
public void run() {
try {
System.out.println(getName() + " take result " + queue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
运行结果（某一次）
p1 offer result false p2 offer result true c1 take result 50
说明：此时，只有c1线程得到了匹配，p1线程存放元素，直接返回的false，因为此时没有消费者线程等待，而p2线程与c1线程进行了匹配，p2线程存放元素成功，c1线程获取元素成功，并且此时，c2线程还是处于park状态，此时应用程序无法正常结束。所以，可知，必须要先有取操作，然后存操作，两者才能正确的匹配，若先是存操作，然后再是取操作，此时无法匹配成功，会阻塞，取操作期待下一个存操作进行匹配。
五、总结
SynchronousQueue的源码就分析到这里，SynchronousQueue适合一对一的匹配场景，没有容量，无法缓存。有了这个基础，之后会方便分析线程池框架的源码，谢谢各位园友的观看~
时间： 2024-10-05 04:26:25