【Java并发编程】17、SynchronousQueue源码分析

时间:2022-12-09 06:13:12

SynchronousQueue是一种特殊的阻塞队列,不同于LinkedBlockingQueue、ArrayBlockingQueue和PriorityBlockingQueue,其内部没有任何容量,任何的入队操作都需要等待其他线程的出队操作,反之亦然。如果将SynchronousQueue用于生产者/消费者模式,那么相当于生产者和消费者手递手交易,即生产者生产出一个货物,则必须等到消费者过来取货,方可完成交易。 
SynchronousQueue有一个fair选项,如果fair为true,称为fair模式,否则就是unfair模式。fair模式使用一个先进先出的队列保存生产者或者消费者线程,unfair模式则使用一个后进先出的栈保存。

基本原理

SynchronousQueue通过将入队出队的线程绑定到队列的节点上,并借助LockSupport的park()和unpark()实现等待,先到达的线程A需调用LockSupport的park()方法将当前线程进入阻塞状态,知道另一个与之匹配的线程B调用LockSupport.unpark(Thread)来唤醒在该节点上等待的线程A。 
基本逻辑:

  1. 初始状态队列为null
  2. 当一个线程到达,如果队列为null,无与之匹配的线程,则进入队列等待;队列不为null,参考3
  3. 当另一个线程到达,如果队列不为null,则判断队列中的第一个元素(针对fair和unfair不同)是否与其匹配,如果匹配则完成交易,不匹配则也入队;队列为null,参考2

常用方法解析

在深入分析其实现机制之前,我们先了解对于SynchronousQueue可执行哪些操作,由于SynchronousQueue的容量为0,所以一些针对集合的操作,如:isEmpty()/size()/clear()/remove(Object)/contains(Object)等操作都是无意义的,同样peek()也总是返回null。所以针对SynchronousQueue只有两类操作:

  • 入队(put(E)/offer(E, long, TimeUnit)/offer(E))
  • 出队(take()/poll(long, TimeUnit)/poll())

这两类操作内部都是调用Transferer的transfer(Object, boolean, long)方法,通过第一个参数是否为null,来区分是生产者还是消费者(生产者不为null)。 
针对以上情况,我们将着重分析Transferer的transfer(Object, boolean, long)方法,这里由于两种不同的公平模式,会存在两个Transferer的派生类:

public SynchronousQueue(boolean fair) {
transferer = (fair)? new TransferQueue() : new TransferStack();
}

可见fair模式使用TransferQueue,unfair模式使用TransferStack,下面我们将分别对这两种模式进行着重分析。

fair模式

fair模式使用一个FIFO的队列保存线程,TransferQueue的结构如下:

/** Dual Queue */
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;
} ...
} /** 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 cancelled.
*/
transient volatile QNode cleanMe; TransferQueue() {
QNode h = new QNode(null, false); // initialize to dummy node.
head = h;
tail = h;
} ...
}

以上是TransferQueue的大致结构,可以看到TransferQueue同一个普通的队列,同时存在一个指向队列头部的指针——head,和一个指向队列尾部的指针——tail;cleanMe的存在主要是解决不可清楚队列的尾节点的问题,后面会介绍到;队列的节点通过内部类QNode封装,QNode包含四个变量:

  • next:指向队列中的下一个节点
  • item:节点包含的数据
  • waiter:等待在该节点上的线程
  • isData:表示该节点由生产者创建还是由消费者创建,由于生产者是放入数据,所以isData==true,而消费者==false

其他的内容就是一些CAS变量以及操作,下面主要分析TransferQueue的三个重要方法:transfer(Object, boolean, long)、awaitFulfill(QNode, Object, boolean, long)、clean(QNode, QNode)。这三个方法是TransferQueue的核心,入口是transfer(),下面具体看代码。

transfer

/**
* @By Vicky:交换数据,生产者和消费者通过e==null来区分
*/
Object transfer(Object e, boolean timed, long nanos) {
QNode s = null; // constructed/reused as needed
boolean isData = (e != null);// e==null,则isData==false,else idData==true for (;;) {// 循环
QNode t = tail;
QNode h = head;
if (t == null || h == null) // 无视即可,具体信息在方法开始的注释中有提到
continue; // spin // h==t队列为null,tail的isData==isData表示该队列中的等待的线程与当前线程是相同模式
//(同为生产者,或者同为消费者)(队列中只存在一种模式的线程)
// 此时需要将该线程插入到队列中进行等待
if (h == t || t.isData == isData) {
QNode tn = t.next;
if (t != tail) // inconsistent read
continue;
// 这里的目的是为了帮助其他线程完成入队操作
if (tn != null) { // lagging tail
// 原子性将tail从t更新为tn,即将tail往后移动,直到队列的最后一个元素
advanceTail(t, tn);
continue;
}
// 如果nanos<=0则说明不等待,那么到这里已经说到队列没有可匹配的线程,所以直接返回null即可
if (timed && nanos <= 0) // can't wait
return null;
// 仅初始化一次s,节点s会保存isData信息作为生产者和消费者的区分
if (s == null)
s = new QNode(e, isData);
// 原子性的更新t的next指针指向s,上面将tail从t更新为tn就是为了处理此处剩下的操作
// 由于此处插入一个节点分成了两个步骤,所以过程中会插入其他线程,导致看到不一致状态
// 所以其他线程会执行剩下的步骤帮助其完成入队操作
if (!t.casNext(null, s)) // failed to link in
continue; // 如果自己执行失败没有关系,会有其他线程帮忙执行完成的,所以才无需锁,类似ConcurrentLinkedQueue
advanceTail(t, s); // swing tail and wait
// 等待匹配
Object x = awaitFulfill(s, e, timed, nanos);
// 这里有两种情况:
// A:匹配完成,返回数据
// B:等待超时/取消,返回原节点s
if (x == s) { // wait was cancelled
// 情况B则需要清除掉节点s
clean(t, s);
return null;
} // 情况A,则匹配成功了,但是还需要将该节点从队列中移除
// 由于FIFO原则,所以匹配上的元素必然是队列的第一个元素,所以只需要移动head即可
if (!s.isOffList()) { // not already unlinked
// 移动head指向s,则下次匹配从s.next开始
advanceHead(t, s); // unlink if head
// 清除对节点中保存的数据的引用,GC友好
if (x != null) // and forget fields
s.item = s;
s.waiter = null;
}
return (x != null)? x : e; } else { // complementary-mode
// 进行匹配,从队列的头部开始,即head.next,非head
QNode m = h.next; // node to fulfill
if (t != tail || m == null || h != head)
continue; // inconsistent read // 判断该节点的isData是否与当前线程的isData匹配
// 相等则说明m已经匹配过了,因为正常情况是不相等才对
// x==m说明m被取消了,见QNode的tryCancel()方法
// CAS设置m.item为e,这里的e,如果是生产者则是数据,消费者则是null,
// 所以m如果是生产者,则item变为null,消费者则变为生产者的数据
// CAS操作失败,则直接将m出队,CAS失败说明m已经被其他线程匹配了,所以将其出队,然后retry
Object x = m.item;
if (isData == (x != null) || // m already fulfilled
x == m || // m cancelled
!m.casItem(x, e)) { // lost CAS
advanceHead(h, m); // dequeue and retry
continue;
}
// 与m匹配成功,将m出队,并唤醒等待在m上的线程m.waiter
advanceHead(h, m); // successfully fulfilled
LockSupport.unpark(m.waiter);
return (x != null)? x : e;
}
}
}

从上面的代码可以看出TransferQueue.transfer()的整体流程:

  1. 判断当前队列是否为null或者队尾线程是否与当前线程匹配,为null或者不匹配都将进行入队操作
  2. 入队主要很简单,分成两步:修改tail的next为新的节点,修改tail为新的节点,这两步操作有可能分在两个不同的线程执行,不过不影响执行结果
  3. 入队之后需要将当前线程阻塞,调用LockSupport.park()方法,直到打断/超时/被匹配的线程唤醒
  4. 如果被取消,则需要调用clean()方法进行清除
  5. 由于FIFO,所以匹配总是发生在队列的头部,匹配将修改等待节点的item属性传递数据,同时唤醒等待在节点上的线程

awaitFulfill

下面看看具体如何让一个线程进入阻塞。

/**
*@ By Vicky:等待匹配,该方法会进入阻塞,直到三种情况下才返回:
* a.等待被取消了,返回值为s
* b.匹配上了,返回另一个线程传过来的值
* c.线程被打断,会取消,返回值为s
*/
Object awaitFulfill(QNode s, Object e, boolean timed, long nanos) {
// timed==false,则不等待,lastTime==0即可
long lastTime = (timed)? System.nanoTime() : 0;
// 当前线程
Thread w = Thread.currentThread();
// 循环次数,原理同自旋锁,如果不是队列的第一个元素则不自旋,因为压根轮不上他,自旋只是浪费CPU
// 如果等待的话则自旋的次数少些,不等待就多些
int spins = ((head.next == s) ?
(timed? maxTimedSpins : maxUntimedSpins) : 0);
for (;;) {
if (w.isInterrupted())// 支持打断
s.tryCancel(e);
// 如果s的item不等于e,有三种情况:
// a.等待被取消了,此时x==s
// b.匹配上了,此时x==另一个线程传过来的值
// c.线程被打断,会取消,此时x==s
// 不管是哪种情况都不要再等待了,返回即可
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;
}
}
// 自旋,直到spins==0,进入等待
if (spins > 0)
--spins;
// 设置等待线程
else if (s.waiter == null)
s.waiter = w;
// 调用LockSupport.park进入等待
else if (!timed)
LockSupport.park(this);
else if (nanos > spinForTimeoutThreshold)
LockSupport.parkNanos(this, nanos);
}
}

awaitFulfill()主要涉及自旋以及LockSupport.park()两个关键点,自旋可去了解自旋锁的原理。

自旋锁原理:通过空循环则霸占着CPU,避免当前线程进入睡眠,因为睡眠/唤醒是需要进行线程上下文切换的,所以如果线程睡眠的时间很段,那么使用空循环能够避免线程进入睡眠的耗时,从而快速响应。但是由于空循环会浪费CPU,所以也不能一直循环。自旋锁一般适合同步快很小,竞争不是很激烈的场景。

LockSupport.park()可到API文档进行了解。

clean

下面再看看如何清除被取消的节点。

/**
*@By Vicky:清除节点被取消的节点
*/
void clean(QNode pred, QNode s) {
s.waiter = null; // forget thread
// 如果pred.next!=s则说明s已经出队了
while (pred.next == s) { // Return early if already unlinked
QNode h = head;
QNode hn = h.next; // Absorb cancelled first node as head
// 从队列头部开始遍历,遇到被取消的节点则将其出队
if (hn != null && hn.isCancelled()) {
advanceHead(h, hn);
continue;
}
QNode t = tail; // Ensure consistent read for tail
// t==h则队列为null
if (t == h)
return;
QNode tn = t.next;
if (t != tail)
continue;
// 帮助其他线程入队
if (tn != null) {
advanceTail(t, tn);
continue;
}
// 只能出队非尾节点
if (s != t) { // If not tail, try to unsplice
// 出队方式很简单,将pred.next指向s.next即可
QNode sn = s.next;
if (sn == s || pred.casNext(s, sn))
return;
}
// 如果s是队尾元素,那么就需要cleanMe出场了,如果cleanMe==null,则只需将pred赋值给cleanMe即可,
// 赋值cleanMe的意思是等到s不是队尾时再进行清除,毕竟队尾只有一个
// 同时将上次的cleanMe清除掉,正常情况下此时的cleanMe已经不是队尾了,因为当前需要清除的节点是队尾
// (上面说的cleanMe其实是需要清除的节点的前继节点)
QNode dp = cleanMe;
if (dp != null) { // Try unlinking previous cancelled node
QNode d = dp.next;
QNode dn;
// d==null说明需要清除的节点已经没了
// d==dp说明dp已经被清除了,那么dp.next也一并被清除了
// 如果d未被取消,说明哪里出错了,将cleanMe清除,不清除这个节点了
// 后面括号将清除cleanMe的next出局,前提是cleanMe.next没有已经被出局
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);
// dp==pred说明cleanMe.next已经其他线程被更新了
if (dp == pred)
return; // s is already saved node
} else if (casCleanMe(null, pred))
return; // Postpone cleaning s
}
}

清除节点时有个原则:不能清除队尾节点。所以如果对尾节点需要被清除,则将其保存到cleanMe变量,等待下次进行清除。在清除cleanMe时可能说的有点模糊,因为涉及到太多的并发会出现很多情况,所以if条件太多,导致难以分析全部情况。

以上就是TransferQueue的操作逻辑,下面看看后进先出的TransferStack。

unfair模式

unfair模式使用一个LIFO的队列保存线程,TransferStack的结构如下:

/** Dual stack */
static final class TransferStack extends Transferer {
/* 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;// 正在匹配中... /** 只需要判断mode的第二位是否==1即可,==1则正在匹配中...*/
static boolean isFulfilling(int m) { return (m & FULFILLING) != 0; } /** Node class for TransferStacks. */
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. SNode(Object item) {
this.item = item;
}
} /** The head (top) of the stack */
volatile SNode head; 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;
}
}

TransferStacks比TransferQueue的结构复杂些。使用一个head指向栈顶元素,使用内部类SNode封装栈中的节点信息,SNode包含5个变量:

  • next:指向栈中下一个节点
  • match:与之匹配的节点
  • waiter:等待的线程
  • item:数据
  • mode:模式,对应REQUEST/DATA/FULFILLING(第三个并不是FULFILLING,而是FULFILLING | REQUEST或者FULFILLING | DATA)

SNode的5个变量,三个是volatile的,另外两个item和mode没有volatile修饰,代码注释给出的解释是:对这两个变量的写总是发生在volatile/原子操作的之前,读总是发生在volatile/原子操作的之后。

上面提到SNode.mode的三个常量表示栈中节点的状态,f分别为:

  • REQUEST:0,消费者的请求生成的节点
  • DATA:1,生产者的请求生成的节点
  • FULFILLING:2,正在匹配中的节点,具体对应的mode值是FULFILLING | REQUEST和FULFILLING | DATA

其他内部基本同TransferQueue,不同之处是当匹配到一个节点时并非是将被匹配的节点出栈,而是将匹配的节点入栈,然后同时将匹配上的两个节点一起出栈。下面我们参照TransferQueue来看看TransferStacks的三个方法:transfer(Object, boolean, long)、awaitFulfill(QNode, Object, boolean, long)、clean(QNode, QNode)。

transfer

/**
* @By Vicky:交换数据,生产者和消费者通过e==null来区分
*/
Object transfer(Object e, boolean timed, long nanos) {
SNode s = null; // constructed/reused as needed
int mode = (e == null)? REQUEST : DATA;// 根据e==null判断生产者还是消费者,对应不同的mode值 for (;;) {
SNode h = head;
// 栈为null或者栈顶元素的模式同当前模式,则进行入栈操作
if (h == null || h.mode == mode) { // empty or same-mode
// 不等待,则直接返回null,返回之前顺带清理下被取消的元素
if (timed && nanos <= 0) { // can't wait
if (h != null && h.isCancelled())
casHead(h, h.next); // pop cancelled node
else
return null;
} else if (casHead(h, s = snode(s, e, h, mode))) {// 入栈,更新栈顶为新节点
// 等待,返回值m==s,则被取消,需清除
SNode m = awaitFulfill(s, timed, nanos);
// m==s说明s被取消了,清除
if (m == s) { // wait was cancelled
clean(s);
return null;
}
// 帮忙出栈
if ((h = head) != null && h.next == s)
casHead(h, s.next); // help s's fulfiller
// 消费者则返回生产者的数据,生产者则返回自己的数据
return mode == REQUEST? m.item : s.item;
}
} else if (!isFulfilling(h.mode)) { // try to fulfill // 栈顶未开始匹配,则开始匹配
// h被取消,则出栈
if (h.isCancelled()) // already cancelled
casHead(h, h.next); // pop and retry
// 更新栈顶为新插入的节点,并更新节点的mode为FULFILLING,对应判断是否正在出栈的方法
// 匹配需要先将待匹配的节点入栈,所以不管是匹配还是不匹配都需要创建一个节点入栈
else if (casHead(h, s=snode(s, e, h, FULFILLING|mode))) {
// 循环直到找到一个可以匹配的节点
for (;;) { // loop until matched or waiters disappear
// m即与s匹配的节点
SNode m = s.next; // m is s's match
// m==null说明栈s之后无元素了,直接将栈顶设置为null,并重新进行最外层的循环
if (m == null) { // all waiters are gone
casHead(s, null); // pop fulfill node
s = null; // use new node next time
break; // restart main loop
}
// 将s设置为m的匹配节点,并更新栈顶为m.next,即将s和m同时出栈
SNode mn = m.next;
if (m.tryMatch(s)) {
casHead(s, mn); // pop both s and m
return (mode == REQUEST)? m.item : s.item;
} else // lost match
// 设置匹配失败,则说明m正准备出栈,帮助出栈
s.casNext(m, mn); // help unlink
}
}
} else { // help a fulfiller // 栈顶已开始匹配,帮助匹配
// 此处的操作逻辑同上面的操作逻辑一致,目的就是帮助上面进行操作,因为此处完成匹配需要分成两步:
// a.m.tryMatch(s)和b.casHead(s, mn)
// 所以必然会插入其他线程,只要插入的线程也按照这个步骤执行那么就避免了不一致问题
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
}
}
}
}

从上面的代码可以看出TransferStack.transfer()的整体流程:

  1. 判断当前栈是否为null或者栈顶线程是否与当前线程匹配,为null或者不匹配都将进行入栈操作
  2. 入栈主要很简单,分成两步:插入一个节点入栈,该步无需同步,第二步需要head指针指向新节点,该步通过CAS保证安全
  3. 入栈之后需要将当前线程阻塞,调用LockSupport.park()方法,直到打断/超时/被匹配的线程唤醒
  4. 如果被取消,则需要调用clean()方法进行清除
  5. 由于LIFO,所以匹配的节点总是栈顶的两个节点,分成两步:原子性更新节点的match变量,更新head。由于两步无法保证原子性,所以通过将栈顶元素的mode更新为FULFILLING,阻止其他线程在栈顶发生匹配时进行其他操作,同时其他线程需帮助栈顶进行的匹配操作

awaitFulfill

下面看看TransferStack是如何让一个线程进入阻塞。

/**
*@ By Vicky:等待匹配,逻辑大致同TransferQueue可参考阅读
*/
SNode awaitFulfill(SNode s, boolean timed, long nanos) {
long lastTime = (timed)? System.nanoTime() : 0;
Thread w = Thread.currentThread();
SNode h = head;
// 计算自旋的次数,逻辑大致同TransferQueue
int spins = (shouldSpin(s)?
(timed? maxTimedSpins : maxUntimedSpins) : 0);
for (;;) {
if (w.isInterrupted())
s.tryCancel();
// 如果s的match不等于null,有三种情况:
// a.等待被取消了,此时x==s
// b.匹配上了,此时match==另一个节点
// c.线程被打断,会取消,此时x==s
// 不管是哪种情况都不要再等待了,返回即可
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;
// 设置等待线程
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);
}
}

逻辑基本同TransferQueue,不同之处是通过修改SNode的match变量标示匹配,以及取消。

clean

下面再看看如何清除被取消的节点。

/**
* @By Vicky:清除节点
*/
void clean(SNode s) {
s.item = null; // forget item
s.waiter = null; // forget thread
// 清除
SNode past = s.next;
if (past != null && past.isCancelled())
past = past.next; // Absorb cancelled nodes at head
// 从栈顶节点开始清除,一直到遇到未被取消的节点,或者直到s.next
SNode p;
while ((p = head) != null && p != past && p.isCancelled())
casHead(p, p.next); // Unsplice embedded nodes
// 如果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;
}
}

以上即全部的TransferStack的操作逻辑。

看完了TransferQueue和TransferStack的逻辑,SynchronousQueue的逻辑基本清楚了。

应用场景

SynchronousQueue的应用场景得看具体业务需求,J.U.C下有一个应用案例:Executors.newCachedThreadPool()就是使用SynchronousQueue作为任务队列。

出处:http://blog.csdn.net/vickyway/article/details/50113429