Java并发编程之ConcurrentLinkedQueue队列详情
ConcurrentLinkedQueue
JDK中提供了一系列场景的并发安全队列。总的来说，按照实现方式的不同可分为阻塞队列和非阻塞队列，前者使用锁实现，而后则使用CAS非阻塞算法实现。
ConcurrentLinkedQueue 内部的队列使用单向链表方式实现，其中有两个volatile 类型的 Node 节点分别用来存放队列的首、尾节点。从下面的无参构造函数可知，默认头、尾节点都是指向 item 为null 的哨兵节点。新元素会被插入队列末尾，出队时从队列头部获取一个元素。
public ConcurrentLinkedQueue() {
head = tail = new Node<E>(null);
}
在 Node 节点内部则维护一个使用volatile 修饰的变量 item，用来存放节点的值;next用来存放链表的下一个节点，从而链接为一个单向无界链表。其内部则使用 UNSafe 工具类提供的CAS 算法来保证出入队时操作链表的原子性。
下面通过介绍ConcurrentLinkedQueue的几个方法来介绍其实现原理。
offer操作： offer操作是在队列末尾添加一个元素，如果传递的参数是null则抛出NPE异常，否则由于ConcurrentLinkedQueue是无界队列，该方法一直会返回true。另外，由于使用CAS无阻塞算法，因此该方法不会阻塞挂起调用线程。下面具体看下实现原理。
public boolean offer(E e) {
//（1）e为null这抛出空指针异常
checkNotNull(e);
//(2)构造Node节点，在构造函数内部调用unsafe.putObject
final Node<E> newNode = new Node<E>(e);
//(3) 从尾节点插入
for (Node<E> t = tail, p = t;;) {
Node<E> q = p.next;
//(4) 如果q==null说明p是尾节点，则执行插入
if (q == null) {
// p is last node
//（5）使用CAS设置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".
//（6）CAS成功，则说明新增节点已经放入链表，然后设置当前尾巴节点
if (p != t) // hop two nodes at a time
casTail(t, newNode); // Failure is OK.
return true;
}
// Lost CAS race to another thread; re-read next
}
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 = (p != t && t != (t = tail)) ? t : q;
}
}
- 首先看当一个线程调用
offer(item)时的情况。首先代码(1)对传参进行空检查， 由于使用 如果为null 则抛出NPE 异常，否则执行代码(2)并使用item作为构造函数参数创建一 个新的节点，然后代码(3)从队列尾部节点开始循环，打算从队列尾部添加元素。这时候节点p、t、head、tail同时指向了item为null的哨兵节点，由于哨兵节点的next 节点为null，所以这里q也指向null。代码(4)发现q->null则执行代码(5),通过CAS 原子操作判断p节点的next节点是否为null，如果为null 则使用节点newNode替换p的next节点，然后执行代码(6)，这里由于p=t所以没有设置尾部节点，然后退出 offer方法。 - 上面是一个线程调用offer方法的情况，如果多个线程同时调用，就会存在多个线程同时执行到代码(5)的情况。假设线程A调用offer(item1)，线程B调用 ofer(item2)，同时执行到代码(5)p.casNext(null, newNode)。由于CAS的比较设置操作是原子性的，所以这里假设线程A先执行了比较设置操作，发现当前p的 next 节点确实是null，则会原子性地更新next节点为iteml，这时候线程B也会判断p的next节点是否为null，结果发现不是null(因为线程A已经设置了p的next节点为iteml)，则会跳到代码(3)，然后执行到代码(4)。
可见，offer 操作中的关键步骤是代码(5)，通过原子CAS 操作来控制某时只有一个线程可以追加元素到队列末尾。进行CAS 竞争失败的线程会通过循环一次次尝试进行 CAS操作，直到CAS 成功才会返回，也就是通过使用无限循环不断进行 CAS 尝试方式来替代阻塞算法挂起调用线程。相比阻塞算法，这是使用CPU资源换取阻塞所带来的开销。
add操作：
add操作是在链表尾部添加一个元素，其实在内部调用的还是offer操作。
public boolean add(E e) {
return offer(e);
}
poll操作：
poll操作是在队列头部获取并移除一个元素，如果队列为空则返回null。
public E poll() {
restartFromHead:
for (;;) {
for (Node<E> h = head, p = h, q;;) {
E item = p.item;
if (item != null && p.casItem(item, null)) {
// Successful CAS is the linearization point
// for item to be removed from this queue.
if (p != h) // hop two nodes at a time
updateHead(h, ((q = p.next) != null) ? q : p);
return item;
}
else if ((q = p.next) == null) {
updateHead(h, p);
return null;
}
else if (p == q)
continue restartFromHead;
else
p = q;
}
}
}
poll方法在移除一个元素时，只是简单地使用 CAS操作把当前节点的item值设置为null，然后通过重新设置头节点将该元素从队列里面移除，被移除的节点就成了孤立节点，这个节点会在垃圾回收时被回收掉。另外，如果在执行分支中发现头节点被修改了，要跳到外层循环重新获取新的头节点。
peak操作：
peak操作是获取队列头部获一个元素，如果队列为空则返回null。
public E peek() {
restartFromHead:
for (;;) {
for (Node<E> 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;
}
}
}
Peek操作的代码结构与poll操作类似，不同之处在于我们在代码中标记注释的地方中少了castItem操作。其实这很正常，因为peek只是获取队列头元素值，并不清空其值。根据前面的介绍我们知道第一次执行offer后head指向的是哨兵节点(也就是item为null的节点)，那么第一次执行peek时在注释处会发现item==null，然后执行q=p.next，这时候q节点指向的才是队列里面第一个真正的元素，或者如果队列为 null 则 q 指向 null。
上一篇：Java十分钟精通异常处理机制
栏 目：JAVA代码
下一篇：java 方法与数组基础使用详解
本文标题：Java并发编程之ConcurrentLinkedQueue队列详情
本文地址：http://www.codeinn.net/misctech/222804.html
