深刻Java集合学习系列：SynchronousQueue实现原理
http://m.blog.csdn.net/blog/luoyuyou/30256497java
背景：node
一个BlockingQueue的是一个这样的队列，每一个插入操做都必须等待另外一个删除操做，反过来也同样。一个同步队列没有内部容量这个概念。你不能使用peek操做，由于一个元素仅在你试着删除它的时候才可以被取得。你不能插入一个元素（任何方法），直到另外一个线程试着删除它。你不能迭代它们，由于没有东西能够被迭代。queue的头元素head是第一个进入队列中的元素，而且插入线程正在为它等待。若是队列中没有像以上的元素存在，那么调用poll的方法会返回null。对于Collection的其余方法（好比contains），SynchronousQueue表现得像一个空的集合。它不容许null入队。算法
这个队列相似于CSP和Ada中使用的会合信道。它们适合于切换的设计，好比一个线程中的对象必须同步等待另外一个线程中运行的对象从而传递一些信息/事件/任务。数据结构
这个类支持可选的公平策略从而制订生产者和等待者的等待顺序。默认状况下，这个顺序是没有保证的，使用true能够确保队列是FIFO的。app
这个类以及它的迭代器实现了某些Collection和Iterator中的方法。less
算法：ide
这个算法实现了双重栈和双重队列算法。oop
(LIfo)栈用做非公平模式，(Fifo)队列用于公平模式。这二者的性能类似。Fifo常常能在竞争状况下提供更高的吞吐量，可是Lifo可以在通常应用中维持更高的线程局部性。性能
双重队列（以及类似的栈）在任什么时候刻都是持有“数据”--item经过put操做提供或者“请求”--slo经过take操做提供，或者为空。一个调用试着“知足”（一个请求的调用获得数据或者一个数据的调用匹配了请求）结果是出队了一个模式互补的节点。最有趣的的地方在于任何操做都可以明确当前队列处于哪一种模式，以及表现得好像不须要锁。
this
队列和栈都扩展了抽象的Transferer接口，它定义了惟一一个transfer方法用于put或者take。这些方法统一在一个方法下是由于在双重数据结构中，put和take方法是对称的两种方法，因此几乎全部代码能够被组合。结果transfer方法是比较长的，可是这样相对于把他们分红几乎重复的几部分代码仍是更好的。
这个队列和栈共享了不少类似的概念可是不多的具体细节。为了简单性，他们保持不一样从而在后续能够分开演化。
这个同步队列的算法不一样于之前的算法，包括消除的机制。主要的不一样在于：
- 其余的算法使用位-掩码的方式，可是如今的节点直接使用了模式位，从而致使了其余的不一样。
- 同步队列必须等待直到被其余线程来“知足”。
- 提供了超时和中断的支持，包括从链表中清理节点/线程从而避免垃圾数据的持有和内存消耗。
阻塞操做主要经过LockSupport的park/unpark方法来实现，而那些极可能被下一个调用知足的节点会首先自旋必定次数（仅在多核处理器上）。在很是繁忙的同步队列中，自旋可以显著地提高吞吐量。可是在通常状况下，这个自旋次数也是足够小从而不那么明显。
清理操做在队列和栈中的方式是不一样的。队列中，咱们几乎使用O(1)的时间来清除被取消的节点。可是若是它正被做为tail元素，那么就必须等待后续的清除操做来清理它。栈中，咱们须要潜在的O(n)时间来遍历，从而确认咱们可以清除这个节点，可是这个操做能够和其余线程共同做用于这个栈。
当垃圾回收关注于大部分节点的回收问题时，复杂的非阻塞算法会试着“忘记”数据、其余节点的引用，以及线程会持有这些数据在一段长的阻塞时间内。在这种状况下给一个节点中的值设置null将会违背主要的算法，咱们会代替使用将引用指向节点自身的方式。这个状况不怎么会在栈中发生（由于阻塞线程不会挂在旧的头节点上），可是队列中的节点引用必须积极地忘记，从而避免已经结束的节点的可达性。
实现：
实现的重点在于transfer这个方法：
- 不能放入null元素，不然抛出NullPointerException异常。
- 不限时版本（timed==false），返回非null为成功插入，返回null抛出InterruptedException异常。
- 限时版本（timed==true），返回非null为成功插入，返回null的确状况下，若处于中断状态则抛出InterruptedException异常，不然代表超时。
Transferer接口分为queue和stack两个版本，
咱们先分析queue版本（源码）：
queue
/**
* Puts or takes an item.
*/
@SuppressWarnings("unchecked")
E transfer(E e, boolean timed, long nanos) {
QNode s = null; // constructed/reused as needed
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
QNode tn = t.next;
if (t != tail) // inconsistent read
continue;
if (tn != null) { // lagging tail
advanceTail(t, tn);
continue;
}
if (timed && nanos <= 0) // can't wait
return null;
if (s == null)
s = new QNode(e, isData);
if (!t.casNext(null, s)) // failed to link in
continue;
advanceTail(t, s); // swing tail and wait
Object x = awaitFulfill(s, e, timed, nanos);
if (x == s) { // wait was cancelled
clean(t, s);
return null;
}
if (!s.isOffList()) { // not already unlinked
advanceHead(t, s); // unlink if head
if (x != null) // and forget fields
s.item = s;
s.waiter = null;
}
return (x != null) ? (E)x : e;
} else { // complementary-mode
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.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;
}
}
}
这里的基本算法是循环试着执行下列行动之一：
- 若是队列是空的或者拥有和当前调用一样的模式，试着在队列中增长等待节点，等待这个节点被知足（或者取消），而后返回相匹配的数据。
- 若是队列不为空以及队列拥有和当前调用相补的模式，试着去经过CAS操做来改变等待节点的item域，弹出队列，以及返回相匹配的数据。
在每一种状况中，都会检查而且帮助递增head和tail域，在其余的线程相对停滞和缓慢的状况下。
一开始对于null的检查是为了不看到没有初始化的head和tail值。在当前的SynchronousQueue中这种状况不会发生，除非调用者使用的是非volatile/final的Transferer。
这个检查存在的缘由是，放在循环的顶部快于将它在循环中穿插地放置。
详情以下：
- 取得当前调用的模式isData，进入循环，首先排除head和tail都为null的状况。
- 当队列为空（head=tail)或者队尾节点的模式和当前调用模式相同的状况下：假如tail元素已经改变则更新重来，假如当前是限时版本而且时间参数不大于0则返回null，建立新节点而且以CAS方式入队，更新tail节点，而后等待相补的调用的来临（awaitFulfill），假如返回的数据是节点自己（说明是中断或者超市）则作清理，并返回null，不然假如节点还在队列中则更新head元素而且修改当前节点的item和waiter域，而后返回得到的x和e中非空的元素（此处一定是1空1非空）。
- 不然，队列中有相补模式的节点，则试着取得队首元素即head.next，尝试知足他（fulfill)，假如失败则重试，成功以后会递增head的值，而后唤醒（unpark）等待线程以及返回相匹配的数据。
接着咱们来看阻塞操做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, E e, boolean timed, long nanos) {
/* Same idea as TransferStack.awaitFulfill */
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);
Object x = s.item;
if (x != e)
return x;
if (timed) {
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
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);
}
}
这里的s是根据调用构造的节点，e为传递的数据（null是请求，不然数据），timed为限时标示，nanos时间量（纳秒）：
- 首先咱们根据timed取得结束时间，假如非限时则为0，而后取得线程对象以及自旋次数：队首元素&非限时（maxUntimedSpins最大）；队首元素&限时（maxTimedSpins次之）；不然为0。
- 进入循环：假如中断则尝试取消节点，假如item变化则返回数据，假如限时版本而且超时则尝试取消并重试或者更新nanos，递减自旋次数直到0----->设置节点中的线程对象----->调用非限时挂起版本或者根据阈值判断是否调用限时挂起版本。
注意：以上循环在有相补的调用发生时老是会返回对应的数据，在被中断或者超时处理成功状况下会返回当前节点自己。
咱们最后看transfer方法中的清理操做：
void clean(QNode pred, QNode s) {
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.
*/
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
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
QNode sn = s.next;
if (sn == s || pred.casNext(s, sn))
return;
}
QNode dp = cleanMe;
if (dp != null) { // Try unlinking previous cancelled node
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
}
}
方法中的pred为s入队时的前驱，这个方法中注释里有以下说明：
在任一时刻，只有一个链表中的节点不能被删除---最后一个插入的节点。为了迁就这个原则，若是咱们没法删除s，那么咱们就保存它的前驱节点为“cleanMe”，首先删除以前保存的版本。因此至少s或者以前保存的节点可以被删除，因此最后老是可以被删除！
详情以下：
- 首先既然s已经被取消，则设置它的等待者waiter为null，进入循环（条件：pred.next为s推断出s不为head，s为head的话不须要删除了）。
- 取得head和tail，探测他们是否停滞，过于停滞时更新他们的值。
- 当s节点不为队尾节点时候（尝试更新前驱pred，或者当s处于OffList时返回）。
- s节点为尾元素：假如cleanMe不为空，则说明有以前并未删除的尾节点，那么则尝试删除以前的cleanMe以后的节点，不然尝试设置当前的pred为cleanMe，等待下一次的删除，而后返回。
这里的难点在于： 当咱们取得cleanMe时，若是不为空，并不了解是否有其余线程进过一样的操做，因此咱们首先要判断d=dp.nex不为空（若为空则修改cleanMe而后返回）、d != dp（不然说明dp已经offList）、d.isCancelled（不然说明以前的节点已经被删除、以及d不为尾节点&&dn=d.next不为空&&dn != d（不然d已经offList），以上状况下才能够修改dp的next为dn从而删除了以前存储的待删除尾节点，而且修改cleanMe的值从dp变为null。假如cleanMe为空，则能够尝试设置本身的pred节点，在竞争失败的状况下能够重试是能够取得进展的，由于任什么时候候只有一个节点可以做为tail。
因此咱们会尝试清理以前被删除的尾节点，以及尝试设置本身的前驱为cleanMe节点。
if (dp == pred)这一行在我看来是不必的，由于永远不可能出现。
咱们接着分析stack:
stack
/**
* Puts or takes an item.
*/
@SuppressWarnings("unchecked")
E transfer(E e, boolean timed, long nanos) {
SNode s = null; // constructed/reused as needed
int mode = (e == null) ? REQUEST : DATA;
for (;;) {
SNode h = head;
if (h == null || h.mode == mode) { // empty or same-mode
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))) {
SNode m = awaitFulfill(s, timed, nanos);
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 (E) ((mode == REQUEST) ? m.item : s.item);
}
} else if (!isFulfilling(h.mode)) { // try to fulfill
if (h.isCancelled()) // already cancelled
casHead(h, h.next); // pop and retry
else if (casHead(h, s=snode(s, e, h, FULFILLING|mode))) {
for (;;) { // loop until matched or waiters disappear
SNode m = s.next; // m is s's match
if (m == null) { // all waiters are gone
casHead(s, null); // pop fulfill node
s = null; // use new node next time
break; // restart main loop
}
SNode mn = m.next;
if (m.tryMatch(s)) {
casHead(s, mn); // pop both s and m
return (E) ((mode == REQUEST) ? m.item : s.item);
} else // lost match
s.casNext(m, mn); // help unlink
}
}
} else { // help a fulfiller
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
}
}
}
}
这里的Node实际上拥有三种状态：REQUEST/DATA/FULFILLING，基本算法是循环试着执行下列3种行动之一：
- 若是栈为空或者包含相同模式的节点，试着入栈而后等待匹配，返回匹配数据，在取消时返回null。
- 若是包含相补的节点，试着入栈一个fulfill模式的节点，匹配相对应的等待节点，弹出这两个节点，返回匹配数据。这里的匹配或者取消连接操做可能没有被实际执行，由于第三种行动：
- 若是栈顶元素为fulfill模式的节点，尝试帮助它执行match以及pop操做，而后再重试。这里的代码和fulfill的行为几乎相同的，只不过不返回数据。
详情以下：
- 取得当前调用的模式mode，REQUEST或则DATA，而后进入循环。
- 取得栈顶元素h，当队列为空或者栈顶元素模式与当前模式相同时：首先排除限时调用和nanos不大于0的状况，而后返回null。不然连接到h，并尝试入栈，成功以后经过awaitFulfill等待相补调用的来临，而后根据返回SNode节点是否为s自己来判断是否须要清理操做，假如获取了数据那么便协助更新head，以及根据当前模式返回数据。
- 当前栈顶元素与当前调用模式不一样，那么假如当前栈顶元素模式不为fulfill时（经过 (mode & FULFILLING) != 0来判断），进入循环，s为当前栈顶元素，m为下一个待匹配元素（一定不为null），尝试取得mn（m.next），试着使用m.tryMatch(s)，来完成m和s的匹配，而且传递s到了m.match（注意，这一步可能也会由另外一种状况完成），成功以后改变head值，以及返回当前数据或者匹配数据。假如失败（意味着最后探测到的match不为s，惟一的场景为等待线程中断或者超时），则从新设置s的next值为mn，而后重试。
- 当前栈顶元素的模式为fulfill时，尝试取得栈顶元素h和next节点m，而后试着匹配m和h的值，这里的做用几乎和上一种状况中相似，只不过不返回数据，以及只协助一次。
接着咱们来看阻塞操做awaitFulfill：
/**
* Spins/blocks until node s is matched by a fulfill operation.
*
* @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) {
final long deadline = timed ? System.nanoTime() + nanos : 0L;
Thread w = Thread.currentThread();
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) {
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
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);
}
}
这个操做事实上和队列版本中的相似，首先来解释下注释中的内容：当一个节点/线程试着去阻塞，它会在设置waiter域以后至少检查一次状态，而后才会调用parking（阻塞），这样子能够经过waiter从而和它的知足者协做从而确保不会丢失信号。若是当前调用的几点位于栈顶，那么在park以前会首先尝试自旋，这样能够在生产者和消费者很是接近时避免阻塞。可是这个只在多核处理器下才会有用。从代码中的检查状况能够看出，在优先级上，中断状态--->正式返回---->超时。（因此最后一个检查是用来探测超时的）除了非限时的同步队列。{poll/offer}方法不会检查中断以及等待过久，因此对于中断和超时的判断被放置于transfer方法中，这样要好于调用awaitFulfill。详情以下（与队列版本中相似）：取得当前的结束时间，当前线程，以及自旋次数。而后进入循环。首先判断是否中断，判断限时版本下的时间流逝，判断自旋，以及根据当前节点s所处的位置来设置自旋次数。设置线程对象（用于唤醒）。最后根据是否限时来阻塞当前线程，限时版本下会根据阈值来判断是否须要阻塞。最后咱们来看处理中断和超时状况下的清理操做clean:
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;
// Absorb cancelled nodes at head
SNode p;
while ((p = head) != null && p != past && p.isCancelled())
casHead(p, p.next);
// Unsplice embedded nodes
while (p != null && p != past) {
SNode n = p.next;
if (n != null && n.isCancelled())
p.casNext(n, n.next);
else
p = n;
}
}
这里的s为被取消了的节点，这里的注释有以下说明：
最坏状况下咱们须要遍历整个栈才能取消s的连接。若是有其余的取消操做同时在进行，咱们可能看不到s，由于它已经被其余的线程删除了。可是咱们能够观察跟随s以后的节点，若是这个节点也是取消状态，那么咱们会使用下一个节点，咱们不会再检查，由于不想要遍历两遍仅仅是为了找到哨兵节点。
详情以下：
- 设置s节点的item和waiter都为null，由于已经不须要了，而且它的状态能够由match为this来判断。
- 取得s的下一个节点past，假如past也是取消的，那么再取下一节点。
- 从头p=head开始，逐步断开past以前的那些被取消的节点。
- 再从p开始删除嵌在栈中的节点，知道栈为空或者找到哨兵节点（past）。
相关文章
- 1. 深刻Java集合学习系列：HashMap的实现原理
- 2. 深刻Java集合学习系列：ArrayBlockingQueue及其实现原理
- 3. 深刻Java集合学习系列：HashSet的实现原理
- 4. 深刻Java集合学习系列：Hashtable的实现原理
- 5. 深刻Java集合学习系列：WeakHashMap的实现原理
- 6. 深刻Java集合学习系列：ConcurrentSkipListMap实现原理
- 7. 深刻Java集合学习系列：ArrayList的实现原理
- 8. 深刻Java集合学习系列：ConcurrentLinkedQueue及其实现原理
- 9. 深刻Java集合学习系列：ConcurrentHashSet简单实现
- 10. 深刻Java集合学习系列：深刻CopyOnWriteArraySet
- 更多相关文章...
- • Scala Set(集合) - Scala教程
- • C# 集合（Collection） - C#教程
- • ☆技术问答集锦（13）Java Instrument原理
- • Java Agent入门实战（三）-JVM Attach原理与使用
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
欢迎关注本站公众号,获取更多信息
相关文章
- 1. 深刻Java集合学习系列：HashMap的实现原理
- 2. 深刻Java集合学习系列：ArrayBlockingQueue及其实现原理
- 3. 深刻Java集合学习系列：HashSet的实现原理
- 4. 深刻Java集合学习系列：Hashtable的实现原理
- 5. 深刻Java集合学习系列：WeakHashMap的实现原理
- 6. 深刻Java集合学习系列：ConcurrentSkipListMap实现原理
- 7. 深刻Java集合学习系列：ArrayList的实现原理
- 8. 深刻Java集合学习系列：ConcurrentLinkedQueue及其实现原理
- 9. 深刻Java集合学习系列：ConcurrentHashSet简单实现
- 10. 深刻Java集合学习系列：深刻CopyOnWriteArraySet