ConcurrentLinkedQueue
通过名字大家就可以知道, 这是一个通过链表实现的并发安全的队列, 它应该是java中并发环境下性能最好的队列, 为什么呢? 因为它的不变性(invariants) 与可变性(non-invariants)
1. 基本原则不变性(fundamental invariants)
1.整个队列中一定会存在一个 node(node.next = null), 并且仅存在一个, 但tail引用不一定指向它
2. 队列中所有 item != null 的节点, head一定能够到达; cas 设置 node.item = null, 意味着这个节点被删除
head引用的不变性和可变性
不变性(invariants)
1. 所有的有效节点通过 succ() 方法都可达
2. head != null
3. (tmp = head).next != tmp || tmp != head (其实就是 head.next != head)
可变性(Non-invariants)
1. head.item 可能是 null, 也可能不是 null
2. 允许 tail 滞后于 head, 也就是调用 succ() 方法, 从 head 不可达tail
tail 引用的不变性和可变性
不变性(invariants)
1. tail 节点通过succ()方法一定到达队列中的最后一个节点(node.next = null)
2. tail != null
可变性(Non-invariants)
1. tail.item 可能是 null, 也可能不是 null
2. 允许 tail 滞后于 head, 也就是调用 succ() 方法, 从 head 不可达tail
3. tail.next 可能指向 tail
这些不变性(invariants) 和 可变性(Non-invariants) 造成 ConcurrentLinkedQueue 有些异于一般queue的特点:
1. head 与 tail 都有可能指向一个 (item = null) 的节点
2. 如果 queue 是空的, 则所有 node.item = null
3. queue刚刚创建时 head = tail = dummyNode
4. head/tail 的 item/next 的操作都是通过 CAS
晕了, 是哇! 没事, 这些都是特性, 我们先看代码, 回头再回顾这些特性.
2. 内部节点 Node
import com.lami.tuomatuo.search.base.concurrent.unsafe.UnSafeClass;
import sun.misc.Unsafe;
/**
* http://hg.openjdk.java.net/jdk7/jdk7/jdk/file/9b8c96f96a0f/src/share/classes/sun/misc/Unsafe.java
* http://hg.openjdk.java.net/jdk7/jdk7/hotspot/file/9b0ca45cd756/src/share/vm/prims/unsafe.cpp
* http://mishadoff.com/blog/java-magic-part-4-sun-dot-misc-dot-unsafe/
*
* Created by xjk on 1/13/17.
*/
public class Node<E> {
volatile E item;
volatile Node<E> next;
Node(E item){
/**
* Stores a reference value into a given Java variable.
* <p>
* Unless the reference <code>x</code> being stored is either null
* or matches the field type, the results are undefined.
* If the reference <code>o</code> is non-null, car marks or
* other store barriers for that object (if the VM requires them)
* are updated.
* @see #putInt(Object, int, int)
*
* 将 Node 对象的指定 itemOffset 偏移量设置 一个引用值
*/
unsafe.putObject(this, itemOffset, item);
}
boolean casItem(E cmp, E val){
/**
* Atomically update Java variable to <tt>x</tt> if it is currently
* holding <tt>expected</tt>.
* @return <tt>true</tt> if successful
* 原子性的更新 item 值
*/
return unsafe.compareAndSwapObject(this, itemOffset, cmp, val);
}
void lazySetNext(Node<E> val){
/**
* Version of {@link #putObjectVolatile(Object, long, Object)}
* that does not guarantee immediate visibility of the store to
* other threads. This method is generally only useful if the
* underlying field is a Java volatile (or if an array cell, one
* that is otherwise only accessed using volatile accesses).
*
* 调用这个方法和putObject差不多, 只是这个方法设置后对应的值的可见性不一定得到保证,
* 这个方法能起这个作用, 通常是作用在 volatile field上, 也就是说, 下面中的参数 val 是被volatile修饰
*/
unsafe.putOrderedObject(this, nextOffset, val);
}
/**
* Atomically update Java variable to <tt>x</tt> if it is currently
* holding <tt>expected</tt>.
* @return <tt>true</tt> if successful
*
* 原子性的更新 nextOffset 上的值
*
*/
boolean casNext(Node<E> cmp, Node<E> val){
return unsafe.compareAndSwapObject(this, nextOffset, cmp, val);
}
private static Unsafe unsafe;
private static long itemOffset;
private static long nextOffset;
static {
try {
unsafe = UnSafeClass.getInstance();
Class<?> k = Node.class;
itemOffset = unsafe.objectFieldOffset(k.getDeclaredField("item"));
nextOffset = unsafe.objectFieldOffset(k.getDeclaredField("next"));
}catch (Exception e){
}
}
}
整个内部节点 Node 的代码比较简单, 若不了解 Unsafe 类使用的, 请点击链接 Unsafe 与 LockSupport
3. ConcurrentLinkedQueue 内部属性及构造方法
/** head 节点 */
private transient volatile Node<E> head;
/** tail 节点 */
private transient volatile Node<E> tail;
public ConcurrentLinkedList() {
/** 默认会构造一个 dummy 节点
* dummy 的存在是防止一些特殊复杂代码的出现
*/
head = tail = new Node<E>(null);
}
初始化 ConcurrentLinkedQueue时 head = tail = dummy node.
4. 查询后继节点方法 succ()
/**
* 获取 p 的后继节点, 若 p.next = p (updateHead 操作导致的), 则说明 p 已经 fall off queue, 需要 jump 到 head
*/
final Node<E> succ(Node<E> p){
Node<E> next = p.next;
return (p == next)? head : next;
}
获取一个节点的后继节点不是 node.next 吗, No, No, No, 还有特殊情况, 就是tail 指向一个哨兵节点 (node.next = node); 代码的注释中我提到了 哨兵节点是 updateHead 导致的, 那我们来看 updateHead方法.
5. 特别的更新头节点方法 updateHead
为什么说 updateHead 特别呢? 还是看代码
/**
* Tries to CAS head to p, If successfully, repoint old head to itself
* as sentinel for succ(), blew
*
* 将节点 p设置为新的节点(这是原子操作),
* 之后将原节点的next指向自己, 直接变成一个哨兵节点(为queue节点删除及garbage做准备)
*
* @param h
* @param p
*/
final void updateHead(Node<E> h, Node<E> p){
if(h != p && casHead(h, p)){
h.lazySetNext(h);
}
}
主要这个 h.lazySetNext(h), 将 h.next -> h 直接变成一个哨兵节点, 这种lazySetNext主要用于无阻塞数据结构的 nulling out, 要了解详情 点击 Unsafe 与 LockSupport
有了上面的这些辅助方法, 我们开始进入正题
6. 入队列操作 offer()
一般我们的思维: 入队操作就是 tail.next = newNode; 而这里不同, 为什么呢? 我们再来回顾一下 tail 的不变性和可变性
不变性(invariants)
1. tail 节点通过succ()方法一定到达队列中的最后一个节点(node.next = null)
2. tail != null
可变性(Non-invariants)
1. tail.item 可能是 null, 也可能不是 null
2. 允许 tail 滞后于 head, 也就是调用 succ() 方法, 从 head 不可达tail
3. tail.next 可能指向 tail
主要是这里 tail 会滞后于 head, 所以呢 要找到正真的 last node (node.next = null)
直接来代码
/**
* Inserts the specified element at the tail of this queue
* As the queue is unbounded, this method will never return {@code false}
*
* @param e {@code true} (as specified by {@link Queue#offer(Object)})
* @return NullPointerException if the specified element is null
*
* 在队列的末尾插入指定的元素
*/
public boolean offer(E e){
checkNotNull(e);
final Node<E> newNode = new Node<E>(e); // 1. 构建一个 node
for(Node<E> t = tail, p = t;;){ // 2. 初始化变量 p = t = tail
Node<E> q = p.next; // 3. 获取 p 的next
if(q == null){ // q == null, 说明 p 是 last Node
// p is last node
if(p.casNext(null, newNode)){ // 4. 对 p 进行 cas 操作, newNode -> p.next
// Successful CAS is the linearization point
// for e to become an element of the queue,
// and for newNode to become "live"
if(p != t){ // 5. 每每经过一次 p = q 操作(向后遍历节点), 则 p != t 成立, 这个也说明 tail 滞后于 head 的体现
casTail(t, newNode); // Failure is OK
}
return true;
}
}
else if(p == q){ // 6. (p == q) 成立, 则说明p是pool()时调用 "updateHead" 导致的(删除头节点); 此时说明 tail 指针已经 fallen off queue， 所以进行 jump 操作， 若在t没变化, 则 jump 到 head, 若 t 已经改变(jump操作在另外的线程中执行), 则jump到 head 节点, 直到找到 node.next = null 的节点
/** 1. 大前提 p 是已经被删除的节点
* 2. 判断 tail 是否已经改变
* 1) tail 已经变化, 则说明 tail 已经重新定位
* 2) tail 未变化, 而 tail 指向的节点是要删除的节点, 所以让 p 指向 head
* 判断尾节点是否有变化
* 1. 尾节点变化, 则用新的尾节点
* 2. 尾节点没变化, 将 tail 指向head
*
* public void test(){
* String tail = "";
* String t = (tail = "oldTail");
* tail = "newTail";
* boolean isEqual = t != (t = tail); // <- 神奇吧
* System.out.println("isEqual : "+isEqual); // isEqual : true
* }
*/
p = (t != (t = tail))? t : head;
}else{
// 7. (p != t) -> 说明执行过 p = q 操作(向后遍历操作), "(t != (t = tail)))" -> 说明尾节点在其他的线程发生变化
// 为什么 "(t != (t = tail)))" 一定要满足呢, 因为 tail变更, 节省了 (p = q) 后 loop 中的无畏操作, tail 更新说明 q节点肯定也是无效的
p = (p != t && (t != (t = tail))) ? t : q;
}
}
}
先瞄一下这段代码: 发现有3大疑惑:
- 明明 Node<E> q = p.next, 怎么会有 p = q ?
- "p = (t != (t = tail))? t : head" 这段代码是什么玩意, 是不是让你直接怀疑自己的java基础了, 不急我们慢慢来.
- 最后就是 "p = (p != t && (t != (t = tail))) ? t : q"
queue 初始化时是这样的:
state1.png
整个 queue 中 head = tail = dummyNode, 这时我们开始 offer 元素
1） 添加元素 a
1. 由于 head = tail = dummyNode, 所以 p.next = null
2. 直接操作步骤4 (p.casNext(null, newNode)), 若操作成功, 接着往下走, 不成功(并发时 其他的cas操作成功), 再loop 重试至成功
3. 判断 p != t, 这时没出现 tail指向的不是 last node，所以不成立, 直接return
添加元素a后:
state2.png
- 添加元素 b
- 此时还是 head = tail = dummyNode, p节点是 dummyNode, q.item = a, q.item != null 且 q != null, 直接执行步骤7 p = q (p != t && (t != (t = tail)) 下面说)
- 再次 判断 q == null, 所以 有执行步骤4 p.casNext(), 这时因为执行过 p = q, 所以 p != t 成立, 对tail进行cas操作
- 最后直接 return
添加 b 之后:
state3.png
- 添加元素c
- 这里操作步骤和添加 a 一样, 所以不说了
添加c后:
- 这里操作步骤和添加 a 一样, 所以不说了
state4.png
解决上面的疑惑(看这里时最好将下面的 poll也看一遍):
1. "p = q", 这是在poll方法中调用 updateHead 方法所致的
2. "p = (t != (t = tail))", 这段代码的意思是 若 tail 节点在另外的节点中有变化 tail != t, 则将 tail 赋值给 p.虽然只有这短短一行代码, 但是包含非常多的意思:
i!= 这个操作符号不是原子的, 它可以被中断;
ii) 执行时 先获取t的值, 再 t = tail, 赋值好了之后再与原来的t比较
iii) 在多线程环境中 tail 很可能在上面添加元素的过程中被改变, 所以会出现 t != tail, 若tail被修改, 则用新的tail, 不然直接跳到head节点
3. 多了一个 p != t , 因为 tail变更, 节省了 (p = q) 后 loop 中的无畏操作, tail 更新说明 q节点肯定也是无效的
OK 至此 整个offer是分析好了, 接下来 poll
7. 出队列操作 poll()
因为这个操作涉及 head 引用, 所以我们再来回顾一下head的不变性和可变性:
不变性(invariants)
1. 所有的有效节点通过 succ() 方法都可达
2. head != null
3. (tmp = head).next != tmp || tmp != head (其实就是 head.next != head)
可变性(Non-invariants)
1. head.item 可能是 null, 也可能不是 null
2. 允许 tail 滞后于 head, 也就是调用 succ() 方法, 从 head 不可达tail
head主要特点 tail 可能之后 head, 且head.item 可能是 null
不废话了, 直接上代码
public E poll(){
restartFromHead:
for(;;){ // 0. 为啥这里面是两个 for 循环? 不防, 你去掉个试试, 其实主要是为了在 "continue restartFromHead" 后进行第二个 for loop 中的初始化
for(Node<E> h = head, p = h, q;;){ // 1.进行变量的初始化 p = h = head,
E item = p.item;
if(item != null && p.casItem(item, null)){ // 2. 若 node.item != null, 则进行cas操作, cas成功则返回值
// Successful CAS is the linearization point
// for item to be removed from this queue
if(p != h){ // hop two nodes at a time // 3. 若此时的 p != h, 则更新 head(那啥时 p != h, 额, 这个绝对坑啊 -> 执行第8步后)
updateHead(h, ((q = p.next) != null)? q : p); // 4. 进行 cas 更新 head ; "(q = p.next) != null" 怕出现p此时是尾节点了; 在 ConcurrentLinkedQueue 中正真的尾节点只有1个(必须满足node.next = null)
}
return item;
}
else if((q = p.next) == null){ // 5. queue是空的, p是尾节点
updateHead(h, p); // 6. 这一步除了更新head 外, 还是helpDelete删除队列操作, 删除 p 之前的节点(和 ConcurrentSkipListMap.Node 中的 helpDelete 有异曲同工之妙)
return null;
}
else if(p == q){ // 7. p == q -> 说明 p节点已经是删除了的head节点, 为啥呢?(见updateHead方法)
continue restartFromHead;
}else
p = q; // 8. 将 q -> p, 进行下个节点的 poll 操作(初始化一个 dummy 节点, 在单线程情况下, 这个 if 判断是第一个执行的)
}
}
}
理解了offer之后我想 poll 应该比较简单了.
我们再来回顾一下刚刚添加了 a, b, c, 之后队列的状态:
state4.png
- poll 第一个元素 a
1. 此时 head指向 dummy, tail 指向 item = b 的节点, 所以在步骤2中 item == null, 而 (q = p.next) != null, 所以直接跳到步骤8,
2. 这时 p指向a, 且满足 item != null, 所以执行步骤2, 又因为执行了步骤8, 所以 p != h, 进行 head 节点的更新 (head 指向这时p.next节点)
poll item = a 后:
state6.png
- poll 第二个元素 b
1. 此时 head = tail = b 节点, 所以 item != null, 直接执行 步骤2, 而 p == h , 所以不更新head
poll 节点 b 后:
state7.png
- poll 第三个元素 c
poll 节点 c 和 poll 节点啊一样的, 所以不说了, 直接看结果图
state8.png
一目了然, tail 滞后于 head
- ok 这时我们再进行 offer() 节点 d, 则就会出现 offer 中的步骤 6 (p == q), 所以这时p直接跳到 head节点, 来进行更新, 步骤省略....
结果如图 :
state9.png
至此整个 poll 分析结束
8. 总结
ConcurrentLinkedQueue 的整个设计十分精妙, 它使用 CAS 处理对数据的操作, 同时允许队列处于不一致的状态; 这种特性分离了一般 poll/offer时需要两个原子的操作, 对了尤其是节点的删除 (updateHead) 和后继节点的访问 succ(), 而对 ConcurrentLinkedQueue的掌握有助于我们了解 SynchronousQueue, AQS, FutureTask 中的 Queue
参考资料:
Simple, Fast, and Practical Non-Blocking and Blocking Concurrent Queue
vickyqi ConcurrentLinkedQueue
大飞 ConcurrentLinkedQueue
