Jdk1.6 JUC源码解析(15)-SynchronousQueue
作者：大飞
功能简介：
- SynchronousQueue是一种特殊的阻塞队列，它本身没有容量，只有当一个线程从队列取数据的同时，另一个线程才能放一个数据到队列中，反之亦然。存取过程相当于一个线程把数据(安全的)交给另一个线程的过程。
- SynchronousQueue也支持公平和非公平模式。
源码分析：
- SynchronousQueue内部采用伪栈和伪队列来实现，分别对应非公平模式和公平模式。先看下这部分实现。
伪栈和伪队列的公共基类：
static abstract class Transferer {
/**
* 转移数据的方法，用来实现put或者take。
*
* @param e 如果不为null，相当于将一个数据交给消费者；
* 如果为null，相当于从一个生产者接收一个消费者交出的数据。
* @param timed 操作是否支持超时。
* @param nanos 超时时间，单位纳秒。
* @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.
*/
abstract Object transfer(Object e, boolean timed, long nanos);
}
先看下伪栈实现，内部结构如下：
static final class TransferStack extends Transferer {
/** 表示一个没有得到数据的消费者 */
static final int REQUEST = 0;
/** 表示一个没有交出数据的生产者 */
static final int DATA = 1;
/**
* 表示正在匹配另一个生产者或者消费者。
*/
static final int FULFILLING = 2;
/** 判断是否包含正在匹配(FULFILLING)的标记 */
static boolean isFulfilling(int m) { return (m & FULFILLING) != 0; }
/** Node class for TransferStacks. */
static final class SNode {
volatile SNode next; // 栈中的下一个节点
volatile SNode match; // 和当前节点完成匹配的节点
volatile Thread waiter; // to control park/unpark
Object item; // data; or null for REQUESTs
int mode;
// Note: item和mode不需要volatile修饰，
// 是因为它们在其他的volatile/atomic操作之前写，之后读。
SNode(Object item) {
this.item = item;
}
static final AtomicReferenceFieldUpdater<SNode, SNode>
nextUpdater = AtomicReferenceFieldUpdater.newUpdater
(SNode.class, SNode.class, "next");
boolean casNext(SNode cmp, SNode val) {
return (cmp == next &&
nextUpdater.compareAndSet(this, cmp, val));
}
static final AtomicReferenceFieldUpdater<SNode, SNode>
matchUpdater = AtomicReferenceFieldUpdater.newUpdater
(SNode.class, SNode.class, "match");
/**
* 尝试匹配节点s和当前节点，如果匹配成功，唤醒等待线程。
* (向消费者传递数据或向生产者获取数据)调用tryMatch方法
* 来确定它们的等待线程，然后唤醒这个等待线程。
*
* @param s the node to match
* @return true if successfully matched to s
*/
boolean tryMatch(SNode s) {
if (match == null &&
matchUpdater.compareAndSet(this, null, s)) {
//如果当前节点的match为空,那么CAS设置s为match，然后唤醒waiter。
Thread w = waiter;
if (w != null) { // waiters need at most one unpark
waiter = null;
LockSupport.unpark(w);
}
return true;
}
//如果match不为null，或者CAS设置match失败，那么比较match和s是否为相同对象。
//如果相同，说明已经完成匹配，匹配成功。
return match == s;
}
/**
* 尝试取消当前节点(有线程等待)，通过将match设置为自身。
*/
void tryCancel() {
matchUpdater.compareAndSet(this, null, this);
}
boolean isCancelled() {
return match == this;
}
}
/** The head (top) of the stack */
volatile SNode head;
static final AtomicReferenceFieldUpdater<TransferStack, SNode>
headUpdater = AtomicReferenceFieldUpdater.newUpdater
(TransferStack.class, SNode.class, "head");
下面看下伪栈中transfer方法实现细节吧：
/**
* Puts or takes an item.
*/
Object transfer(Object e, boolean timed, long nanos) {
/*
* 基本算法是在一个无限循环中尝试下面三种情况里面的一种：
*
* 1. 如果当前栈为空或者包含与给定节点模式相同的节点，尝试
* 将节点压入栈内，并等待一个匹配节点，最后返回匹配节点
* 或者null(如果被取消)。
*
* 2. 如果当前栈包含于给定节点模式互补的节点，尝试将这个节
* 点打上FULFILLING标记，然后压入栈中，和相应的节点进行
* 匹配，然后将两个节点(当前节点和互补节点)弹出栈，并返
* 回匹配节点的数据。匹配和删除动作不是必须要做的，因为
* 其他线程会执行动作3:
*
* 3. 如果栈顶已经存在一个FULFILLING(正在满足其他节点)的节
* 点，帮助这个节点完成匹配和移除(出栈)的操作。然后继续
* 执行(主循环)。这部分代码基本和动作2的代码一样，只是
* 不会返回节点的数据。
*/
SNode s = null; // constructed/reused as needed
int mode = (e == null)? REQUEST : DATA;
for (;;) {
SNode h = head;
if (h == null || h.mode == mode) { // head为null或者head和e的mode相同。
if (timed && nanos <= 0) { // 如果超时
if (h != null && h.isCancelled())
casHead(h, h.next); // 如果h不为null且被取消，弹出h。
else
return null; // 否则返回null。
} else if (casHead(h, s = snode(s, e, h, mode))) {//创建一个SNode，赋给s，将原本的head节点做为其next节点，并尝试将其设置为新的head。
//等待其他线程来满足当前线程。
SNode m = awaitFulfill(s, timed, nanos);
if (m == s) { // awaitFulfill方法返回后，判断下是否被取消。
clean(s); // 如果取消，清理一下s节点。
return null;
}
if ((h = head) != null && h.next == s) //因为上面已经将s设置为head，如果满足这个条件说明有其他节点t插入到s前面，变成了head，而且这个t就是和s匹配的节点，他们已经完成匹配。
casHead(h, s.next); // 将s的next节点设置为head。相当于把s和t一起移除了。
return mode == REQUEST? m.item : s.item;
}
} else if (!isFulfilling(h.mode)) {
/*
* 如果栈中存在头节点，且和当前节点不是相同模式，
* 那么说明它们是一对儿对等的节点，尝试用当前节
* 点s来满足h节点。
*/
if (h.isCancelled()) // 如果h节点已经被取消
casHead(h, h.next); // 将h节点弹出，并将h节点的next节点设置为栈的head。
else if (casHead(h, s=snode(s, e, h, FULFILLING|mode))) {//尝试将当前节点打上"正在匹配"的标记，并设置为head。
for (;;) {
SNode m = s.next; // s是当前节点，m是s的next节点，它们是正在匹配的两个节点。
if (m == null) { // 如果m为空，可能其他节点把m匹配走了。
casHead(s, null); // 将s弹出
s = null; // 将s置空，下轮循环的时候还会新建。
break; // 回退到主循环再来一次。
}
SNode mn = m.next; // 获取m的next节点，如果s和m匹配成功，mn就得补上head的位置了。
if (m.tryMatch(s)) { // 尝试匹配一下，匹配成功的话会把m上等待的线程唤醒。
casHead(s, mn); // 如果匹配成功，把s和m弹出。
return (mode == REQUEST)? m.item : s.item;
} else // 没匹配成功的话，说明m可能被其他节点满足了。
s.casNext(m, mn); // 说明m已经被其他节点匹配了，那就把m移除掉。
}
}
} else { // 到这儿的话，说明栈顶的h正在匹配过程中。
SNode m = h.next; // m是h的配对儿，h正在和m匹配。
if (m == null) // 如果m为空，其他节点把m匹配走了。
casHead(h, null); // 弹出h。
else {
SNode mn = m.next; // 获取m的next节点，如果m和h匹配成功，mn就得补上head的位置了。
if (m.tryMatch(h)) // 帮忙匹配一下m和h。
casHead(h, mn); // 匹配成功的话，把h和m弹出。
else // 没匹配成功的话，说明m可能被其他节点满足了。
h.casNext(m, mn); // 没成功的话，说明m已经被其他节点匹配了，那就把m移除掉。
}
}
}
}
看下上面方法中调用的创建节点的snode方法：
static SNode snode(SNode s, Object e, SNode next, int mode) {
if (s == null) s = new SNode(e);
s.mode = mode;
s.next = next;
return s;
}
再看下等待被匹配的方法：
/**
* 自旋/阻塞直到节点被匹配。
*
* @param s the waiting node
* @param timed true if timed wait
* @param nanos timeout value
* @return matched node, or s if cancelled
*/
SNode awaitFulfill(SNode s, boolean timed, long nanos) {
/*
* 在s节点真正阻塞之前，将当前线程设置到s上面，然后检
* 查中断状态(不少于一次)，以确保后续和s匹配的节点来
* 唤醒当前线程。
*
* 当执行此方法时，如果执行节点恰好在栈顶，阻塞之前会
* 做一些自旋，为的是如果有生产者或消费者马上到来，就
* 不需要执行节点阻塞了。这种优化在多核下是有意义的。
*/
long lastTime = (timed)? System.nanoTime() : 0;
Thread w = Thread.currentThread();
SNode h = head;
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) {
long now = System.nanoTime();
nanos -= now - lastTime;
lastTime = now;
if (nanos <= 0) {
s.tryCancel(); //如果超时了，也取消当前节点。
continue;
}
}
if (spins > 0)
spins = shouldSpin(s)? (spins-1) : 0; //自旋控制，每次循环都检测是否满足自旋条件，满足的话，自旋值就减去1，然后进入下次循环(一直减到0)
else if (s.waiter == null)
s.waiter = w; //第一次循环时，会将当前线程设置到s上。
else if (!timed)
LockSupport.park(this);
else if (nanos > spinForTimeoutThreshold)
LockSupport.parkNanos(this, nanos); //有超时条件下，会检测超时时间是否大于超时阀值(这应该是一个经验值)，大于就阻塞，小于就自旋。
}
}
/**
* 如果s节点就是当前栈中头节点，或者头节点正在匹配过程中，那么可以自旋一下。
*/
boolean shouldSpin(SNode s) {
SNode h = head;
return (h == s || h == null || isFulfilling(h.mode));
}
//=========下面是自旋相关参数，定义在SynchronousQueue类中============================================
/** The number of CPUs, for spin control */
static final int NCPUS = Runtime.getRuntime().availableProcessors();
/**
* The number of times to spin before blocking in timed waits.
* The value is empirically derived -- it works well across a
* variety of processors and OSes. Empirically, the best value
* seems not to vary with number of CPUs (beyond 2) so is just
* a constant.
*/
static final int maxTimedSpins = (NCPUS < 2)? 0 : 32;
/**
* The number of times to spin before blocking in untimed waits.
* This is greater than timed value because untimed waits spin
* faster since they don't need to check times on each spin.
*/
static final int maxUntimedSpins = maxTimedSpins * 16;
/**
* The number of nanoseconds for which it is faster to spin
* rather than to use timed park. A rough estimate suffices.
*/
static final long spinForTimeoutThreshold = 1000L;
最后看清理节点方法：
/**
* 当s节点被取消时，才会调用这个方法。
*/
void clean(SNode s) {
s.item = null; // forget item
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.
*/
SNode past = s.next;
if (past != null && past.isCancelled())
past = past.next;
// 将从栈顶节点开始到past的连续的取消节点移除。
SNode p;
while ((p = head) != null && p != past && p.isCancelled())
casHead(p, p.next);
// 如果p本身未取消(上面的while碰到一个未取消的节点就会退出，但这个节点和past节点之间可能还有取消节点)，再把p到past之间的取消节点都移除。
while (p != null && p != past) {
SNode n = p.next;
if (n != null && n.isCancelled())
p.casNext(n, n.next);
else
p = n;
}
}
}
再看下伪队列实现，内部结构如下：
static final class TransferQueue extends Transferer {
/** Node class for TransferQueue. */
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) {
this.item = item;
this.isData = isData;
}
...
/**
* 尝试取消节点。
* 取消就是将节点的item域指向自身。
*/
void tryCancel(Object cmp) {
itemUpdater.compareAndSet(this, cmp, this);
}
boolean isCancelled() {
return item == this;
}
/**
* 判断节点是否离开了队列。
*/
boolean isOffList() {
return next == this;
}
}
/** 队列头节点 */
transient volatile QNode head;
/** 队列尾节点 */
transient volatile QNode tail;
/**
* 指向一个被取消的节点，如果取消这个节点时，它是最后一个进入队列的节点，
* 那么这个节点可能还没有离开队列。
*/
transient volatile QNode cleanMe;
TransferQueue() {
QNode h = new QNode(null, false); // 初始化一个哨兵节点。
head = h;
tail = h;
}
看下伪队列中transfer方法实现：
/**
* Puts or takes an item.
*/
Object transfer(Object e, boolean timed, long nanos) {
/*
* 基本算法是在一个无限循环中尝试下面两种动作里面的一种：
*
* 1. 如果队列为空，或者包含相同模式(存或者取)的节点。
* 尝试将节点加入等待的队列，直到被匹配(或被取消)，
* 同时返回匹配节点的数据。
*
* 2. 如果队列中包含等待的节点，并且当前节点和这个等待
* 节点能相互匹配，那么尝试匹配等待节点并将这个节点
* 出队，然后返回匹配节点的数据。
*
* 在每个动作里面，都会检测并帮助其他线程来完成节点推进。
*
* 在循环开始的时候会做一个非空检测，以避免当前线程看到
* 未初始化的头尾节点。这种情况在当前SynchronousQueue中
* 永远不会发生，但如果调用者持有一个非volatile/final域
* 的话，就有可能会发生。在循环开始的时间做这个非空检测
* 要比在内部(分支里)做性能好一些。
*/
QNode s = null; // constructed/reused as needed
boolean isData = (e != null);
for (;;) {
QNode t = tail;
QNode h = head;
if (t == null || h == null) // 如果看到未初始化的头尾节点
continue;
if (h == t || t.isData == isData) { // 队列为空或者当前节点和队列中节点模式相同。
QNode tn = t.next;
if (t != tail) // 读取到不一致的结果，说明同时有其他线程修改了tail。
continue;
if (tn != null) { // 说明其他线程已经添加了新节点tn，但还没将其设置为tail。
advanceTail(t, tn); // 当前线程帮忙推进尾节点，就是尝试将tn设置为尾节点。
continue;
}
if (timed && nanos <= 0) // 超时。
return null;
if (s == null)
s = new QNode(e, isData); // 初始化s。
if (!t.casNext(null, s)) // 尝试将当前节点s拼接到t后面。
continue; // 不成功就继续下次循环
advanceTail(t, s); // 尝试将s设置为队列尾节点。
Object x = awaitFulfill(s, e, timed, nanos); // 然后等着被匹配。
if (x == s) { // 如果被取消。
clean(t, s); // 清理s节点。
return null;
}
if (!s.isOffList()) { // 如果s节点还没有离开队列。
advanceHead(t, s); // 尝试将s设置为头节点，移除t。
if (x != null)
s.item = s;
s.waiter = null;
}
return (x != null)? x : e;
} else { // 模式正好互补。
QNode m = h.next; // 找到能匹配的节点。
if (t != tail || m == null || h != head)
continue; // 读取到不一致的结果，进入下一轮循环。
Object x = m.item;
if (isData == (x != null) || // 如果m已经被匹配了。
x == m || // 或者m被取消了。
!m.casItem(x, e)) { // 如果尝试将数据e设置到m上失败。
advanceHead(h, m); // 将h出队，m设置为头结点，然后重试。
continue;
}
advanceHead(h, m); // 成功匹配，推进头节点。
LockSupport.unpark(m.waiter); // 唤醒m上的等待线程。
return (x != null)? x : e;
}
}
}
看下transfer方法中调用的advanceHead和advanceTail方法：
/**
* Tries to cas nh as new head; if successful, unlink
* old head's next node to avoid garbage retention.
*/
void advanceHead(QNode h, QNode nh) {
if (h == head && headUpdater.compareAndSet(this, h, nh))
h.next = h; // forget old next
}
/**
* Tries to cas nt as new tail.
*/
void advanceTail(QNode t, QNode nt) {
if (tail == t)
tailUpdater.compareAndSet(this, t, nt);
}
再看下transfer方法中调用的awaitFulfill方法：
/**
* Spins/blocks until node s is fulfilled.
*
* @param s the waiting node
* @param e the comparison value for checking match
* @param timed true if timed wait
* @param nanos timeout value
* @return matched item, or s if cancelled
*/
Object awaitFulfill(QNode s, Object e, boolean timed, long nanos) {
/* Same idea as TransferStack.awaitFulfill */
long lastTime = (timed)? System.nanoTime() : 0;
Thread w = Thread.currentThread();
int spins = ((head.next == s) ?
(timed? maxTimedSpins : maxUntimedSpins) : 0);
for (;;) {
if (w.isInterrupted())
s.tryCancel(e);
Object x = s.item;
if (x != e)
return x;
if (timed) {
long now = System.nanoTime();
nanos -= now - lastTime;
lastTime = now;
if (nanos <= 0) {
s.tryCancel(e);
continue;
}
}
if (spins > 0)
--spins;
else if (s.waiter == null)
s.waiter = w;
else if (!timed)
LockSupport.park(this);
else if (nanos > spinForTimeoutThreshold)
LockSupport.parkNanos(this, nanos);
}
}
TransferStack的awaitFulfill方法思路差不多，代码就不做解析了。再看下transfer中调用的clean方法：
void clean(QNode pred, QNode s) {
s.waiter = null; // forget thread
/*
* 在任意给定的时间点，能删除的节点一定是最后入队的节点。
* 为了满足这个条件，如果当前无法删除s，就将其前驱节点保
* 存为"cleanMe"，先删除之前保存的版本。至少节点s和之前
* 保存的节点里面有一个能被删除，所以方法一定会结束。
*/
while (pred.next == s) { // Return early if already unlinked
QNode h = head;
QNode hn = h.next;
if (hn != null && hn.isCancelled()) {
advanceHead(h, hn); //如果head节点的next节点被取消，那么推进一下head节点。
continue;
}
QNode t = tail; // Ensure consistent read for tail
if (t == h) // 如果队列为空，
return;
QNode tn = t.next;
if (t != tail) // 出现不一致读，重试。
continue;
if (tn != null) {
advanceTail(t, tn); // 帮助推进尾节点。
continue;
}
if (s != t) { // 如果s不是尾节点，移除s。
QNode sn = s.next;
if (sn == s || pred.casNext(s, sn)) //如果s已经被移除退出循环，否则尝试断开s
return;
}
/*
* 下面要做的事情大体就是：如果s是位节点，那么不会马上删除s，
* 而是将s的前驱节点设置为cleanMe，下次清理其他取消节点的时候
* 会顺便把s移除。
*/
QNode dp = cleanMe;
if (dp != null) { // 如果dp不为null，说明是前一个被取消节点，将其移除。
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))) // 把之前标记为cleanMe节点的next节点d移除。
casCleanMe(dp, null);
if (dp == pred)
return; // 说明s的前驱已经是cleanMe了(后续会被删掉)。
} else if (casCleanMe(null, pred))
return; // 如果当前cleanMe为null，那么将s前驱节点设置为cleanMe，并退出。
}
}
小总结一下：
从上面的分析可以看出，伪栈的结构下，新来的线程会作为栈顶节点或者优先和栈顶的等待节点进行匹配，并不是公平的；但伪队列的结构下，新来的线程会在队尾，或者和队头的等待节点(最前到的)进行匹配，能够保证一定的公平性。
- 有了内部伪栈和伪队列的实现，SynchronousQueue实现起来就很容易了，看下代码：
private transient volatile Transferer transferer;
public SynchronousQueue() {
this(false);
}
public SynchronousQueue(boolean fair) {
transferer = (fair)? new TransferQueue() : new TransferStack();
}
/**
* 添加一个数据到队列，等到其他线程接收这个数据。
*
* @throws InterruptedException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
*/
public void put(E o) throws InterruptedException {
if (o == null) throw new NullPointerException();
if (transferer.transfer(o, false, 0) == null) {
Thread.interrupted();
throw new InterruptedException();
}
}
/**
* 添加一个数据到队列，等到其他线程接收这个数据或者超时。
*
* @return <tt>true</tt> if successful, or <tt>false</tt> if the
* specified waiting time elapses before a consumer appears.
* @throws InterruptedException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
*/
public boolean offer(E o, long timeout, TimeUnit unit)
throws InterruptedException {
if (o == null) throw new NullPointerException();
if (transferer.transfer(o, true, unit.toNanos(timeout)) != null)
return true;
if (!Thread.interrupted())
return false;
throw new InterruptedException();
}
/**
* 添加一个数据到队列，如果有其他线程正等待接收这个数据且接收成功，返回true；否则返回false。
*
* 这个方法不阻塞。
* @param e the element to add
* @return <tt>true</tt> if the element was added to this queue, else
* <tt>false</tt>
* @throws NullPointerException if the specified element is null
*/
public boolean offer(E e) {
if (e == null) throw new NullPointerException();
return transferer.transfer(e, true, 0) != null;
}
/**
* 获取并移除队列前端的数据，如果队列中没有数据，就等待其他线程添加一个数据。
*
* @return the head of this queue
* @throws InterruptedException {@inheritDoc}
*/
public E take() throws InterruptedException {
Object e = transferer.transfer(null, false, 0);
if (e != null)
return (E)e;
Thread.interrupted();
throw new InterruptedException();
}
/**
* 获取并移除队列前端的数据，如果队列中没有数据，就等待其他线程添加一个数据或者超时。
*
* @return the head of this queue, or <tt>null</tt> if the
* specified waiting time elapses before an element is present.
* @throws InterruptedException {@inheritDoc}
*/
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
Object e = transferer.transfer(null, true, unit.toNanos(timeout));
if (e != null || !Thread.interrupted())
return (E)e;
throw new InterruptedException();
}
/**
* 如果其他线程正在添加数据到队列，那么尝试获取并移除这个数据。
*
* 这个方法不阻塞。
* @return the head of this queue, or <tt>null</tt> if no
* element is available.
*/
public E poll() {
return (E)transferer.transfer(null, true, 0);
}
由于SynchronousQueue没有实际的容量，所以其他方法实现起来很简单了：
public boolean isEmpty() {
return true;
}
public int size() {
return 0;
}
public int remainingCapacity() {
return 0;
}
public void clear() {
}
public boolean contains(Object o) {
return false;
}
public boolean remove(Object o) {
return false;
}
public boolean containsAll(Collection<?> c) {
return c.isEmpty();
}
public boolean removeAll(Collection<?> c) {
return false;
}
public boolean retainAll(Collection<?> c) {
return false;
}
...
- 最后看一下SynchronousQueue的序列化，序列化比较特别，因为transferer域本身不需要序列化，但需要记住transferer是内部伪栈和伪队列：
static class WaitQueue implements java.io.Serializable { }
static class LifoWaitQueue extends WaitQueue {
private static final long serialVersionUID = -3633113410248163686L;
}
static class FifoWaitQueue extends WaitQueue {
private static final long serialVersionUID = -3623113410248163686L;
}
private ReentrantLock qlock;
private WaitQueue waitingProducers;
private WaitQueue waitingConsumers;
/**
* Save the state to a stream (that is, serialize it).
*
* @param s the stream
*/
private void writeObject(java.io.ObjectOutputStream s)
throws java.io.IOException {
//序列化时根据TransferQueue类型来创建WaitQueue实例。
boolean fair = transferer instanceof TransferQueue;
if (fair) {
qlock = new ReentrantLock(true);
waitingProducers = new FifoWaitQueue();
waitingConsumers = new FifoWaitQueue();
}
else {
qlock = new ReentrantLock();
waitingProducers = new LifoWaitQueue();
waitingConsumers = new LifoWaitQueue();
}
s.defaultWriteObject();
}
private void readObject(final java.io.ObjectInputStream s)
throws java.io.IOException, ClassNotFoundException {
s.defaultReadObject();
if (waitingProducers instanceof FifoWaitQueue)
transferer = new TransferQueue();
else
transferer = new TransferStack();
}
SynchronousQueue的代码解析完毕！ 参见：
Jdk1.6 JUC源码解析(5)-locks-LockSupport
