ConcurrentLinkedQueue全解析
使用java的concurrent包写程序已经快两三年了,至今对其实现原理不怎么了解,突然感到非常惭愧。下定决心从这个包里面找几个核心类,从源码级别做一个全面的分析,搞懂它们的实现原理,以及它们为啥这么快,这一篇挑选ConcurrentLinkedQueue。
ConcurrentLinkedQueue
ConcurrentLinkedQueue是一个无锁化、非阻塞、线程安全的单向队列,JDK1.5提供,由大名鼎鼎的Doug Lea编写。不过要是你认为这是大神灵感爆发的杰作,那就错了,仔细读读java源文件的注释就知道,这个类的实现原理来自Michael & Scott设计的算法(请参考论文),而Michael & Scott也是在很多前人研究的基础上加以综合完善得到该算法的,成为很多具体平台并发FIFO队列实现的蓝本。如果不读读这篇论文,直接阅读java源码,光凭那点注释,会有点摸不着头脑。
我们要研究的ConcurrentLinkedQueue可以说是这个算法的java版本,当然针对java语言做了很多修改。到这里,你要是认为Doug Lea就干了点将原始算法翻译成java的活,那又错了,从一个理论算法到具体某个语言的实现还有相当长的距离;而且java语言本来就比较慢,所以为了挖掘性能,作者做了大量精细而巧妙的设计。
ConcurrentLinkedQueue实现的依赖cas操作和java的valotile语义,所以不理解这两点是无法看懂代码的。关于cas操作和valotile关键字,不是本文要讲的内容,有很多的资料可以查询,比如:
JAVA中的CAS
volatile关键字解析
说明:
- java里面并没有指针的概念,但是算法原理结构理论讲队列,有“头指针”、“尾指针”的说法,为了方便,下面在讲解ConcurrentLinkedQueue内部结构时也会使用这两个术语;
- 采用的源码是Java 1.8版本;
- 强烈建议先阅读cas和java内存模型相关资料。
一、队列的内部链表结构
下面是摘录自源码,经过简化的ConcurrentLinkedQueue定义片段。
public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
implements Queue<E>, java.io.Serializable {
private transient volatile Node<E> head;
private transient volatile Node<E> tail;
public ConcurrentLinkedQueue() {
head = tail = new Node<E>(null);
}
}
一个空的ConcurrentLinkedQueue包含一个空节点(数据字段=null),队列的HEAD和TAIL指针都指向这个节点。当经过一些出入队列的操作以后,队列的结构可能类似下面:
【N1】->【N2】->【N3】->【N4】->【N5】->【N6】
HEAD TAIL
但是,也有可能类似这样:
【N1】->【N2】->【N3】->【N4】->【N5】->【N6】
HEAD TAIL
上边这个结构,表示N1节点已经出了队列,此时的HEAD是N2节点,但是N1节点的next指针仍然保持着。而TAIL节点指向的N5却不是真正的队尾,队尾是N6。这是因为在入队列的过程中,由于并发,导致TAIL更新不及时。
同样还是上面的结构,如果我把里面的数据字段标识出来,有可能是这样的:
【N1(null)】->【N2(null)】->【N3(data)】->【N4(data)】->【N5(data)】->【N6(data)】
HEAD TAIL
HEAD节点的数据字段是NULL,说明这是一个空节点,这个队列语义上第一个节点应该是N3才对。
再来看一个状态:
【N1】<-| 【N2】->【N3】->【N4】->【N5】->【N6】
HEAD TAIL
N1节点的next字段指向自身,这表示N1节点彻底断开和队列的关系,为什么不直接将N1的next置为null呢,因为这样就不能直接断定N1到底是不是尾节点了。
上面这些情形是ConcurrentLinkedQueue在执行一些操作之后可能处于的状态,之所于允许这些看起来不够严谨的状态,是为了在并发过程中提高效率。但是不管如何,在整个生命周期内,算法保持以下属性:
- 链表里面至少会有一个节点,数据字段为null的节点是空节点;
- 顺着HEAD指针肯定能找到的真正的头节点,并能访问到所有节点;
- TAIL指针不一定指向有效节点,更不能保证指向真正的尾节点,但是它大部分情况下指向尾节点或接近尾节点,因此可以提高效率;
- 和队列断开的节点,next字段指向自身;
- 对于入了队列的节点(哪怕又出了队列),只有尾节点的next字段才等于null。
在ConcurrentLinkedQueue内部链表上,可能有一个或多个数据字段为null的空节点,空节点虽然没有数据,但是对高效地保持链表的连接状态至关重要。另一方面,节点的数据字段和next字段值的变化有很强的规律性:数据字段在节点入队列是不为null,出队列时变为null;next字段入队列时是null,出队列时先保持不变,再指向自身。这种规律性可以有效规避掉cas操作的ABA问题。
上面这些也可以认为是ConcurrentLinkedQueue实现算法的不变式,有些在源码注释里面就有说明,有些是我总结的。带着这些原则才能理解每个方法的实现为什么是这样的。
二、内部节点类
直接上源码吧,由于Node提供一个比较简单的节点数据结构,逻辑不多,但是我在读的过程中还是有一些疑惑,直接将解释写在下面的源码块里面了。
private static class Node<E> {
volatile E item;
volatile Node<E> next;
/**
* Constructs a new node. Uses relaxed write because item can
* only be seen after publication via casNext.
*/
Node(E item) {
UNSAFE.putObject(this, itemOffset, item);
}
为什么要使用UNSAFE.putObject而不直接赋值呢?刚看到“relaxed write”这个注释时一脸懵逼,
后来才搞明白,因为item是volatile修饰的,如果直接赋值,会触发volatile的内存同步语义,
在初始化阶段不需要如此,所以使用UNSAFE.putObject能提高一些性能。
boolean casItem(E cmp, E val) {
return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
}
cas操作设置数据字段,这个操作在出队列操作时,保证并发安全。
void lazySetNext(Node<E> val) {
UNSAFE.putOrderedObject(this, nextOffset, val);
}
这个操作相比UNSAFE.putObject,会一定程度禁止指令重排,
相比volatile赋值,不保证全局可见性,性能也稍好一些,用于节点出队列后断开链接,
boolean casNext(Node<E> cmp, Node<E> val) {
return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
}
给节点设置next字段,入队列的关键操作,cas操作保证并发安全
// Unsafe mechanics
private static final sun.misc.Unsafe UNSAFE;
private static final long itemOffset;
private static final long nextOffset;
}
三、出队列:poll方法
poll的目标是安全地取出第一个有效节点的数据,如果没有返回null。
源码如下,为了方便分析,对关键行,我加上了L1~Ln这样的行标记 。
public E poll() {
restartFromHead:
for (;;) {
L1 for (Node<E> h = head, p = h, q;;) {
E item = p.item;
L2 if (item != null && p.casItem(item, null)) {
// Successful CAS is the linearization point
// for item to be removed from this queue.
L3 if (p != h) // hop two nodes at a time
L4 updateHead(h, ((q = p.next) != null) ? q : p);
return item;
}
L5 else if ((q = p.next) == null) {
L6 updateHead(h, p);
return null;
}
L7 else if (p == q)
continue restartFromHead;
else
L8 p = q;
}
}
}
- L1行:开始一个类似遍历的for循环,初始化h,p为head节点,p是当前检查的节点,q是下一个节点。
- L2行:如果p节点的数据不为空,那么p肯定就是真正的第一个节点,只要能将节点的数据字段(item)置为null,出队列就算成功(参考第一节总结的节点属性);p.casItem(item, null)操作保证了并发poll的安全,如果有其他线程抢先一步执行成功,那么casItem操作会失败。
L2行如果执行成功,那么出队列可以算是完成了,此时的head指向一个空节点,这种情况是允许的,但是队列里面也不能总是留很多空节点,所以L3L4行,就是把head指针后移。
- L3行:p==h,说明刚出队列的p就是遍历开始的第一个节点,有2种情况,1)没有并发poll,head指向了空节点,队列里面只有且仅有一个空节点,此时没有必要做head移动;2)有并发poll,p==h说明本线程在竞争中领先,那么交由那个落后的线程来做head移动。
- L4行:如果p.next==null,说明p是尾节点,链表空了,head也指向p;否则head指向p.next。两种情况都是正确的。
- L5行:无论p的item字段为null,还是在并发中竞争失败,我们都要往后遍历;于是q=p.next,如果q==null,说明链表空了(参考第一节总结的节点属性),此时将head置为尾节点;当然,如果h==p是不需要执行的,updateHead里面有这个判断。
- L7行:p==q实际上就是p==p.next,也即p的next指向自身,这说明p节点已经和链表断开了(参考第一节总结的节点属性),这种情况下,只能重新开始poll。
- L8行:正常迭代下一个节点。
上面L2~L4行,每个线程在成功poll一个节点之后,就会尝试把头结点移到p的下一个节点,这样保证了head后面最多只有一个空节点。L3行的那个判断如果没有,也不会影响程序的正确性,但是性能会差一些。
再介绍一下更新头指针的updateHead方法:
final void updateHead(Node<E> h, Node<E> p) {
if (h != p && casHead(h, p))
h.lazySetNext(h);
}
将头指针从h转移到p,就是将节点h出队列,一旦出队列成功,就将h的next字段指向自身,防止那些出队列的节点仍然互相链接,妨碍垃圾收集,这个操作也维护了第一节总结的节点属性。值得注意的是h.lazySetNext(h)操作没有volatile的语义,有可能对其他线程暂时不可见。 假设poll方法执行到L5行时,p节点已经被另外一个线程出队列了,但是本线程不知道,那么L7行的判断也可能失败,最终执行到L8行,这种情况下程序仍然是正确的,不过会多遍历一些节点。(只能认为,作者经过测量,认为这种情况造成的性能损失低于volatile造成的性能损失)。
还有一个有意思的现象是,在poll的过程中,并不会去修正TAIL指针,所以在链表的TAIL指针是有可能落在HEAD之后的,甚至暂时指向一个已经出队列的节点。
四、入队列:offer方法
offer的目标是将新节点插入到尾节点的后面,即使在并发情况下也不丢失数据。
public boolean offer(E e) {
checkNotNull(e);
final Node<E> newNode = new Node<E>(e);
L1 for (Node<E> t = tail, p = t;;) {
L2 Node<E> q = p.next;
L3 if (q == null) {
// p is last node
L4 if (p.casNext(null, newNode)) {
// Successful CAS is the linearization point
// for e to become an element of this queue,
// and for newNode to become "live".
L5 if (p != t) // hop two nodes at a time
L6 casTail(t, newNode); // Failure is OK.
return true;
}
// Lost CAS race to another thread; re-read next
}
L7 else if (p == q)
// We have fallen off list. If tail is unchanged, it
// will also be off-list, in which case we need to
// jump to head, from which all live nodes are always
// reachable. Else the new tail is a better bet.
L8 p = (t != (t = tail)) ? t : head;
else
// Check for tail updates after two hops.
L9 p = (p != t && t != (t = tail)) ? t : q;
}
}
- L1行:开启一个从tail指针开始遍历的循环,p指向当前疑似尾节点;
- L2行:初始化q=p的下一个节点;
- L3行:如果q==null,说明此刻p确实是尾节点,新节点应该插到后面;
- L4行:casNext执行安全的插入操作;
如果成功插入,那么应该考虑更新tail指针,L5L6行执行这个操作; - L5行:和poll方法一样,如果插入的地方不是在tail处,才往后移,这个行为可以保证,tail后面最多还有一个节点;
- L6行:cas操作安全移动tail指针。
后面的两个if分支说明p不是尾节点。 - L7行:p==q,也即p的next字段指向自身,说明p是一个脱离了链表的节点(并发poll操作造成的,参考第一节总结的节点属性)),需要找一个节点重新开始遍历。
- L8行:从head重新开始肯定正确,但是如果tail指针有更新过,那么从tail开始大概率可能效率更高。
- L9行:按道理直接p=q跳到下一个节点就Ok了,但是代码里面做了这个判断
(p != t && t != (t = tail))
,如果p已经正常往后移动了一次,且tail发生了变化,那么从新的tail重新开始。为什么要加个(p!=t)的前置判断呢?我认为是为了提高效率,因为tail是valotile变量,读取有一定代价,当p==t的时候,作者认为再往后跳一下成功的概率挺高。(作者应该经过测量,可见对性能的压榨已经丧心病狂了)。
移除操作:remove
再来看移除操作,通过数据字段的比较,来移除数据相等的节点。
public boolean remove(Object o) {
if (o != null) {
Node<E> next, pred = null;
for (Node<E> p = first(); p != null; pred = p, p = next) {
boolean removed = false;
E item = p.item;
L1 if (item != null) {
L2 if (!o.equals(item)) {
L3 next = succ(p);
L4 continue;
}
L5 removed = p.casItem(item, null);
}
L6 next = succ(p);
L7 if (pred != null && next != null) // unlink
pred.casNext(p, next);
if (removed)
return true;
}
}
return false;
}
- L1行:判断当前是不是空节点。
- L2行:如果节点数据与输入参数相等,说明这是一个需要移除的节点,否则应该跳到下一个节点;
- L3/4行:节点数据不相等,取出当前节点p的下一个节点,然后重新开始循环;
- L5行:节点数据相等,尝试清空数据;如果别的线程并发执行remove或poll,这一步操作可能失败;
- L6行:执行到这一行说明,p是空节点,(本来就是空,或在L5被清空);取出next节点;
- L7行:由于p是空节点,尝试将p的前继和后继相连;
上面再取p的下一个节点的时候,用了调用了succ这个方法,这里也看一下:
final Node<E> succ(Node<E> p) {
Node<E> next = p.next;
return (p == next) ? head : next;
}
很简单,如果p的next指向自己,说明p已经脱离链表,此时返回head指向的节点。
源码可以看出,如果队列里面有多个相同数据的节点,一次remove调用最多删除一个。
迭代器
private class Itr implements Iterator<E> {
L1 private Node<E> nextNode;
L2 private E nextItem;
L3 private Node<E> lastRet;
Itr() {
L4 advance();
}
private E advance() {
L5 lastRet = nextNode;
L6 E x = nextItem;
Node<E> pred, p;
L7 if (nextNode == null) {
L8 p = first();
pred = null;
} else {
pred = nextNode;
L9 p = succ(nextNode);
}
for (;;) {
L10 if (p == null) {
nextNode = null;
nextItem = null;
return x;
}
E item = p.item;
L11 if (item != null) {
nextNode = p;
nextItem = item;
return x;
} else {
// skip over nulls
L12 Node<E> next = succ(p);
L13 if (pred != null && next != null)
L14 pred.casNext(p, next);
L15 p = next;
}
}
}
public boolean hasNext() {
L16 return nextNode != null;
}
public E next() {
if (nextNode == null) throw new NoSuchElementException();
L17 return advance();
}
public void remove() {
L18 Node<E> l = lastRet;
if (l == null) throw new IllegalStateException();
// rely on a future traversal to relink.
L19 l.item = null;
L20 lastRet = null;
}
}
源代码稍微长点,我删掉了注释。
- L1行:成员变量nextNode指向下一个节点,用来支持hasNext,next等操作;
- L2行:成员变量nextItem是下一个节点的值,即nextNode的item值;为什么单独需要这个变量呢,因为由于并发,nextNode的item字段有可能被置空,而迭代器在hasNext的时候是不允许返回空值的,所以在迭代到nextNode的时候,立即取出item字段保存起来。
- L3行:成员变量lastRet指向上一个节点,用来支持删除操作;
- L4行:构造函数仅调用advance方法,从名字上推测,应该是将迭代器往后推一步;
- L5行:核心方法advance的第一行,先保存lastRet,因为这个方法执行完了以后,nextNode就指向下一个节点了;
- L6行:临时变量x保存nextItem,因为advance执行完以后,nextItem就变成下一个节点的值了;
- L7行:如果nextNode==null,说明这是初始化操作;
- L8行:直接调用p=first(),将p赋值成第一个有效节点或null(队列空的);
- L9行:不是初始化调用,那么p赋值成nextNode的后继节点;
至此p保存了nextNode的备选值 - L10行:如果p==null,说明已经到了链表尾部了,迭代结束;
- L11行:如果p.item!=null,说明p是有效节点,ok,就是它了;
- L12~L15行:如果p.item==null,说明p是无效节点,我们应该跳过它,继续寻找;
- L13~L14行:如果p是空节点,我们尝试将它的前继与后继节点相连;
- L16行:hasNext()方法的实现,就看nextNode是否空;
- L17行:next()方法,很简单就是调用advance;
- L18~L20行:remove()方法,将lastRet指向节点的数据清空;因为当我们调用next方法时,返回的值对应的是lastRet节点。
从上面的实现我们可以知道,在迭代过程中,如果有数据并发入队列,这些数据是可以被迭代到的;如果有值被并发删除或出队列,那么这些数据有可能也被迭代到。尤其是迭代器的remove方法,删除的有可能是一个不存在的数据。
队列长度
ConcurrentLinkedQueue的size方法是很没效率的的,实际是把队列遍历了一遍来计算长度。所以推荐大家使用isEmpty来判断队列是否为空,而不要使用size()==0。
public boolean isEmpty() {
return first() == null;
}
public int size() {
int count = 0;
for (Node<E> p = first(); p != null; p = succ(p))
if (p.item != null)
// Collection.size() spec says to max out
if (++count == Integer.MAX_VALUE)
break;
return count;
}
为什么ConcurrentLinkedQueue内部不维护一个size变量来跟踪队列的长度呢?不是不想,是做不到,由于ConcurrentLinkedQueue使用无锁化设计,通过cas操作来保证并发安全。而cas操作只能保证单个变量的并发安全性,无法在出入队列操作的同时,维护size变量。
小结
上面基本讲解了ConcurrentLinkedQueue源代码的大部分核心内容。在开篇的时候也讲过,高效的队列算法不是一朝一夕的出来的,看似短短一两百行代码,是经过前人大量的研究才得到的;因此我们读起来有些吃力是理所当然的,如果不去读一读相关的论文及其他资料,那就更困难了。ConcurrentLinkedQueue的代码在深入挖掘jvm特性的基础上,做了不少性能优化,比如offer方法的L8L9行;当然,上文的解释是我的推测,不一定正确,如果大家发现谬误之处,欢迎指正。