Java并发容器之ConcurrentLinkeQueue
ConcurrentLinkeQueue是基于LinkedBlockingDeque产生的一种适用于并发操作的双端、无阻塞的队列，可以FIFO，也可以LIFO。
ConcurrentLinkeQueue的线程安全且无阻塞主要取决于其节点是volatile定义的Node，且入队和出队采用CAS操作。
在入队操作中，ConcurrentLinkeQueue支持尾部入队，也支持头部入队。现在看一下尾部入队：
/**
* Links e as last element.
*/
private void linkLast(E e) {
checkNotNull(e);
final Node<E> newNode = new Node<E>(e);
restartFromTail:
for (;;)
for (Node<E> t = tail, p = t, q;;) {
if ((q = p.next) != null && //判断p的下一个节点是否null
(q = (p = q).next) != null) //判断p的下一个节点的下一个节点是否是null
// Check for tail updates every other hop.
// If p == q, we are sure to follow tail instead.
//如果都不是null，走下一步
p = (t != (t = tail)) ? t : q; //如果有其他线程更新了，那么就将p指向新的tail；如果没有，就将p移到下一个节点的下一个节点
//如果p已经出队了，重来
else if (p.prev == p) // NEXT_TERMINATOR
continue restartFromTail;
else {
// p is 最后一个节点
newNode.lazySetPrev(p); // CAS piggyback
//更新p的下一个节点为新入队的节点
if (p.casNext(null, newNode)) {
// Successful CAS is the linearization point
// for e to become an element of this deque,
// and for newNode to become "live".
//如果p和t不相等，那么就更新尾节点tail，一挪两hops
if (p != t) // hop two nodes at a time
casTail(t, newNode); // Failure is OK.
return;
}
// Lost CAS race to another thread; re-read next
}
}
}
具体可以看一个入队演示图，我就不画了： ConcurrentLinkedQueue源码分析 。不过它里面的代码不是最新的JDK源码了。
出队也支持两种方式：
public E pollFirst() {
for (Node<E> p = first(); p != null; p = succ(p)) {
E item = p.item;
//item不为null，且没被其他线程更新
if (item != null && p.casItem(item, null)) {
unlink(p);
return item;
}
}
return null;
}
public E pollLast() {
for (Node<E> p = last(); p != null; p = pred(p)) {
E item = p.item;
if (item != null && p.casItem(item, null)) {
unlink(p);
return item;
}
}
return null;
}重要的是unlink：
void unlink(Node<E> x) {
// assert x != null;
// assert x.item == null;
// assert x != PREV_TERMINATOR;
// assert x != NEXT_TERMINATOR;
final Node<E> prev = x.prev;
final Node<E> next = x.next;
if (prev == null) {
unlinkFirst(x, next);
} else if (next == null) {
unlinkLast(x, prev);
} else {
// Unlink interior node.
//
// This is the common case, since a series of polls at the
// same end will be "interior" removes, except perhaps for
// the first one, since end nodes cannot be unlinked.
//
// At any time, all active nodes are mutually reachable by
// following a sequence of either next or prev pointers.
//
// Our strategy is to find the unique active predecessor
// and successor of x. Try to fix up their links so that
// they point to each other, leaving x unreachable from
// active nodes. If successful, and if x has no live
// predecessor/successor, we additionally try to gc-unlink,
// leaving active nodes unreachable from x, by rechecking
// that the status of predecessor and successor are
// unchanged and ensuring that x is not reachable from
// tail/head, before setting x's prev/next links to their
// logical approximate replacements, self/TERMINATOR.
Node<E> activePred, activeSucc;
boolean isFirst, isLast;
int hops = 1;
// Find active predecessor
for (Node<E> p = prev; ; ++hops) {
if (p.item != null) {
activePred = p;
isFirst = false;
break;
}
Node<E> q = p.prev;
if (q == null) {
if (p.next == p)
return;
activePred = p;
isFirst = true;
break;
}
else if (p == q)
return;
else
p = q;
}
// Find active successor
for (Node<E> p = next; ; ++hops) {
if (p.item != null) {
activeSucc = p;
isLast = false;
break;
}
Node<E> q = p.next;
if (q == null) {
if (p.prev == p)
return;
activeSucc = p;
isLast = true;
break;
}
else if (p == q)
return;
else
p = q;
}
// TODO: better HOP heuristics
if (hops < HOPS
// always squeeze out interior deleted nodes
&& (isFirst | isLast))
return;
// Squeeze out deleted nodes between activePred and
// activeSucc, including x.
skipDeletedSuccessors(activePred);
skipDeletedPredecessors(activeSucc);
// Try to gc-unlink, if possible
if ((isFirst | isLast) &&
// Recheck expected state of predecessor and successor
(activePred.next == activeSucc) &&
(activeSucc.prev == activePred) &&
(isFirst ? activePred.prev == null : activePred.item != null) &&
(isLast ? activeSucc.next == null : activeSucc.item != null)) {
updateHead(); // Ensure x is not reachable from head
updateTail(); // Ensure x is not reachable from tail
// Finally, actually gc-unlink
x.lazySetPrev(isFirst ? prevTerminator() : x);
x.lazySetNext(isLast ? nextTerminator() : x);
}
}
}
/**
* Unlinks non-null first node.
*/
private void unlinkFirst(Node<E> first, Node<E> next) {
// assert first != null;
// assert next != null;
// assert first.item == null;
for (Node<E> o = null, p = next, q;;) {
if (p.item != null || (q = p.next) == null) {
if (o != null && p.prev != p && first.casNext(next, p)) {
skipDeletedPredecessors(p);
if (first.prev == null &&
(p.next == null || p.item != null) &&
p.prev == first) {
updateHead(); // Ensure o is not reachable from head
updateTail(); // Ensure o is not reachable from tail
// Finally, actually gc-unlink
o.lazySetNext(o);
o.lazySetPrev(prevTerminator());
}
}
return;
}
else if (p == q)
return;
else {
o = p;
p = q;
}
}
}
/**
* Unlinks non-null last node.
*/
private void unlinkLast(Node<E> last, Node<E> prev) {
// assert last != null;
// assert prev != null;
// assert last.item == null;
for (Node<E> o = null, p = prev, q;;) {
if (p.item != null || (q = p.prev) == null) {
if (o != null && p.next != p && last.casPrev(prev, p)) {
skipDeletedSuccessors(p);
if (last.next == null &&
(p.prev == null || p.item != null) &&
p.next == last) {
updateHead(); // Ensure o is not reachable from head
updateTail(); // Ensure o is not reachable from tail
// Finally, actually gc-unlink
o.lazySetPrev(o);
o.lazySetNext(nextTerminator());
}
}
return;
}
else if (p == q)
return;
else {
o = p;
p = q;
}
}
}
--------EOF---------
微信分享/微信扫码阅读
微信分享/微信扫码阅读
