ConcurrentLinkedQueue1.8源码浅析
[TOC]
让客户满意是我们工作的目标，不断超越客户的期望值来自于我们对这个行业的热爱。我们立志把好的技术通过有效、简单的方式提供给客户，将通过不懈努力成为客户在信息化领域值得信任、有价值的长期合作伙伴，公司提供的服务项目有：域名与空间、网络空间、营销软件、网站建设、鄂城网站维护、网站推广。
ConcurrentLinkedQueue 1.8 源码浅析
一，简介
ConcurrentlinkedQueue 还是一个基于链表的,×××的,线程安全的单端队列，它采用先进先出（FIFO）的规则对节点进行排序，当我们加入一个元素时，它会插入队列的尾部，当我们获取元素时，会从队列的首部获取元素。它没有使用锁来保证线程安全，使用的是“wait-free”算法来保证整个队列的线程安全。
二，基本成员简介
Node 节点对象
// 存储的数据
volatile E item;
// 下一个节点引用
volatile Node next;
/**
* Constructs a new node. Uses relaxed write because item can
* only be seen after publication via casNext.
*/
// 构造一个node节点
Node(E item) {
UNSAFE.putObject(this, itemOffset, item);
}
// 修改节点的item
boolean casItem(E cmp, E val) {
return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
}
// 懒修改节点的next
void lazySetNext(Node val) {
UNSAFE.putOrderedObject(this, nextOffset, val);
}
// cas修改节点的next节点
boolean casNext(Node cmp, Node val) {
return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
} head 头节点
private transient volatile Node head; tail 尾节点
private transient volatile Node tail; 三，常用方法解析
无参构造方法
public ConcurrentLinkedQueue() {
head = tail = new Node(null);
} 构造节点是其实就是构造了一个node的item为null的节点，然后head和tail指向这个节点，如下图所示：
add 方法
public boolean add(E e) {
return offer(e);
}我们可以看出其实调用的是offer方法，具体参考offer方法的讲解。
offer 方法
源码解析：
public boolean offer(E e) {
// 入队元素不能为null
checkNotNull(e);
// 创建新的节点
final Node newNode = new Node(e);
// 死循环，设置节点
// p获取尾节点
for (Node t = tail, p = t;;) {
// q是p的next节点
Node q = p.next;
// 获取尾节点的next节点
// 尾节点没有下一个节点
if (q == null) {
// p is last node
// 这一步说明p是尾节点，新的节点设置为尾节点的next节点
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".
// 设置尾节点，当之前的尾节点和现在插入的节点之间有一个节点时
// 并不是每一次都cas设置尾节点（优化手段，是怎么想到这种优化的？？）
if (p != t) // hop two nodes at a time
// cas设置尾节点，可能会失败
casTail(t, newNode); // Failure is OK.
return true;
}
// Lost CAS race to another thread; re-read next
}
// 多线程操作时候，由于poll时候会把旧的head变为自引用，然后将head的next设置为新的head
// 所以这里需要重新找新的head
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.
p = (t != (t = tail)) ? t : head;
else
// 寻找尾节点
// Check for tail updates after two hops.
// p!=t
p = (p != t && t != (t = tail)) ? t : q;
}
} 分析插入过程，我们插入使用3个线程来调用offer 方法，ThreadA，ThreadB同时运行，ThreadC最后插入，分析下offer方法的流程。
第一步，队列属于初始化状态，ThreadA，ThreadB同时调用offer方法；创建节点，死循环设置节点，获取尾节点的next节点，此时q== null,两个线程都同时可能看见，然后cas设置尾节点的next节点（队列状态如图A所示），我们假设是ThreadA线程cas设置成功了，然后p==t此时的尾节点其实没有发生变化；此时我们来看ThreadB由于A成功了，所以ThreadB cas失败了，重新循环，此时q != null了，p == q显然不等于，再看下一个else判断p!=t,此时显然p == t,所以才是p = q,然后再次循环，此时的q==null,我们假设没有线程来和ThreadB竞争，所以cas设置成功，然后p!=t吗，显然满足所以设置尾节点，此时的设置尾节点的节点和之前的尾节点之间刚刚好有一个节点（如图B所示）。
第二步，ThreadC插入，此时的尾节点是ThreadB插入的节点假设是B,获取B的next节点，q == null,然后cas设置节点，完成，p==t，所以不用更新尾节点（如图C所示）。
peek 方法
注意：不会删除元素，要和poll方法区别
public E peek() {
restartFromHead:
for (;;) {
for (Node h = head, p = h, q;;) {
E item = p.item;
if (item != null || (q = p.next) == null) {
// 更新头结点
updateHead(h, p);
return item;
}
else if (p == q)
continue restartFromHead;
else
p = q;
}
}
} poll 方法
public E poll() {
restartFromHead:
// 循环
for (;;) {
// 获取头结点
for (Node h = head, p = h, q;;) {
// 获取节点的内容
E item = p.item;
// item不为null ，使用cas设置item为空
if (item != null && p.casItem(item, null)) {
// Successful CAS is the linearization point
// for item to be removed from this queue.
// 更新头结点，和尾节点一样，不是每次都更新
// 头结点item为null是，下个节点就必须更新头结点
// 头结点item不为null时，规则和更新尾节点一样
if (p != h) // hop two nodes at a time
updateHead(h, ((q = p.next) != null) ? q : p);
return item;
}
// 那么获取p节点的下一个节点，如果p节点的下一节点为null，则表明队列已经空了
else if ((q = p.next) == null) {
updateHead(h, p);
return null;
}
// p == q，说明别的线程调用了updateHead，
// 自己的next 指向了自己，重新循环，获取最新的头结点
else if (p == q)
continue restartFromHead;
// 如果下一个元素不为空，则将头节点的下一个节点设置成头节点
else
p = q;
}
}
}
final void updateHead(Node h, Node p) {
if (h != p && casHead(h, p))
h.lazySetNext(h);
} 分析我们按照offer时候元素来执行poll方法，ThreadD和ThreadE同时执行来分析下队列的变化（主要分析p==q的产生）。
初始状态（如图C所示）
第一步，ThreadD和ThreadE执行poll操作，item等于null,所以执行执行下面的操作(q = p.next) == null不等于，p == q不等于，所以p = q,其实就是上图的ThreadA插入的节点，此时的item已经不为null了，所以执行cas设置item为null的操作，假设ThreadD执行成功了,那么此时p!=h就满足了，所以此时要更新头结点调用updateHead，这个方法会更新头结点，并且把原来的头节点的next设置为自己）（如图D所示）；接下我们分析ThreadE，cas失败了需要重新执行，此时的item已经不为null,所以执行执行下面的操作(q = p.next) == null不等于，p == q这使其实已经是等于了，因为ThreadD改变了了以前头结点的next节点为自己，所以需要重新遍历，获取最新的头结点，此时的头结点其实就是ThreadA插入的节点，然后item为null,接着执行下面的判断，最终p就是p.next节点也就是ThreadB节点，然后cas设置item为null，由于p=p.next,所以p发生了变化，所以需要设置ThreadB为头结点（如图E所示）。
看到上面的执行流程可能就有人有疑问了，这不是每次都更新头结点吗，没有优化啊，只看poll方法确实是这样，那什么时候会产生不是每次都更新头节点了，那就是当头节点的item不为null的时候，但是如果按初始化的状况来看，头结点的item一直是null，但是当我看了peek方法之后才发现，peek可以改变这个情况，可以设置item不为null的头结点，其实我们可以在poll方法前调用下peek方法，其实就启动了优化策略，不是每次更新头结点，不知道作者是不是这么设计的，反正就是牛皮。
size 方法
public int size() {
int count = 0;
for (Node 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;
} 我们可以发现size没有加锁，就是遍历了整个队列，但是遍历的同时可能在发生poll或者offer,所以size不是特别的精确，用的时候要注意。
四，总结
ConcurrentLinkedQueue是×××的队列，所以使用时一定要注意内存溢出的问题，还有在执行size方法时一定要注意这个是不准确的值；在学poll和offer方法时，一定要理解更新head和tail节点的时机，这种优化手段值得我们去学习，我觉得这就是学习源码的作用，就是学习作者的源码思想。
参考《Java 并发编程的艺术》
分享标题：ConcurrentLinkedQueue1.8源码浅析
标题网址：http://6mz.cn/article/ihodjs.html