Java并发包探秘 (一) ConcurrentLinkedQueue-白红宇

Java并发包探秘 (一) ConcurrentLinkedQueue

阅读量：5343 次

发布时间：2019-06-15

本文共 7840 字，大约阅读时间需要 26 分钟。

本文是Java并发包探秘的第一篇，旨在介绍一下Java并发容器中用的一些思路和技巧，帮助大家更好的理解Java并发容器，让我们更好的使用并发容器打造更高效的程序。本人能力有限，错误难免。希望及时指出。

Java并发包中有很多精心设计的高并发容器。有、、等。ConcurrentLinkedQueue就是其中设计最为优雅的高并发容器。它被设计成了无锁的、无界的、非阻塞式的单向链表结构。现在就让我们来一步一步揭开他们神秘的面纱。

正文开始：

一说到链表结构，我们首先就会想到的就是组成链表结构的原件，那就是节点。或者有的人称之为元素。ConcurrentLinkedQueue（为了叙述方便后面用CLQ指代）中称之为Node.

我们先来看看CLQ中Node的结构：

Java代码

private static class Node<E> {

private volatile E item;

private volatile Node<E> next;

private static final

AtomicReferenceFieldUpdater<Node, Node>

nextUpdater =

AtomicReferenceFieldUpdater.newUpdater

(Node.class, Node.class, "next");

private static final

AtomicReferenceFieldUpdater<Node, Object>

itemUpdater =

AtomicReferenceFieldUpdater.newUpdater

(Node.class, Object.class, "item");

Node(E x) { item = x; }

Node(E x, Node<E> n) { item = x; next = n; }

E getItem() {

return item;

}

boolean casItem(E cmp, E val) {

return itemUpdater.compareAndSet(this, cmp, val);

}

void setItem(E val) {

itemUpdater.set(this, val);

}

Node<E> getNext() {

return next;

}

boolean casNext(Node<E> cmp, Node<E> val) {

return nextUpdater.compareAndSet(this, cmp, val);

}

void setNext(Node<E> val) {

nextUpdater.set(this, val);

}

}

[java]

private static class Node<E> {

private volatile E item;

private volatile Node<E> next;

private static final

AtomicReferenceFieldUpdater<Node, Node>

nextUpdater =

AtomicReferenceFieldUpdater.newUpdater

(Node.class, Node.class, "next");

private static final

AtomicReferenceFieldUpdater<Node, Object>

itemUpdater =

AtomicReferenceFieldUpdater.newUpdater

(Node.class, Object.class, "item");

Node(E x) { item = x; }

Node(E x, Node<E> n) { item = x; next = n; }

E getItem() {

return item;

}

boolean casItem(E cmp, E val) {

return itemUpdater.compareAndSet(this, cmp, val);

}

void setItem(E val) {

itemUpdater.set(this, val);

}

Node<E> getNext() {

return next;

}

boolean casNext(Node<E> cmp, Node<E> val) {

return nextUpdater.compareAndSet(this, cmp, val);

}

void setNext(Node<E> val) {

nextUpdater.set(this, val);

}

}

1.CLQ中的Node定义成了私有的静态类说明该节点描述只适用在CLQ中。

2.其中用到了AtomicReferenceFieldUpdater原子属性引用原子更新器。该类是抽象的，但该类的内部已经给出了包访问控制级别的一个实现AtomicReferenceFieldUpdaterImpl，原理是利用反射将一个被声明成 volatile 的属性通过Java native interface （JNI）调用，使用cpu指令级的命令将一个变量进行更新，该操作是原子的。atomic 包中还有很多类似的更新器分别针对不同的类型进行原子级别的比较更新原子操作。这里用到了sun.misc.Unsafe 的 compareAndSwap操作（简称CAS）它有三种不同的本地命令分别针对Int、Long、Object进行原子更新操作。

3.我们可以看出CLQ中的Node结构是一个单向的链表结构，因为每个Node只有一个向后的next和一个item用来装内容。CLQ将通过casNext和casItem方法来原子更新Node链的结构。setNext 和setItem则是直接放入

我们再来看CLQ的链结构

Java代码

private static final

AtomicReferenceFieldUpdater<ConcurrentLinkedQueue, Node>

tailUpdater =

AtomicReferenceFieldUpdater.newUpdater

(ConcurrentLinkedQueue.class, Node.class, "tail");

private static final

AtomicReferenceFieldUpdater<ConcurrentLinkedQueue, Node>

headUpdater =

AtomicReferenceFieldUpdater.newUpdater

(ConcurrentLinkedQueue.class, Node.class, "head");

private boolean casTail(Node<E> cmp, Node<E> val) {

return tailUpdater.compareAndSet(this, cmp, val);

}

private boolean casHead(Node<E> cmp, Node<E> val) {

return headUpdater.compareAndSet(this, cmp, val);

}

* Pointer to header node, initialized to a dummy node. The first

* actual node is at head.getNext().

private transient volatile Node<E> head = new Node<E>(null, null);

/** Pointer to last node on list **/

private transient volatile Node<E> tail = head;

[java]

private static final

AtomicReferenceFieldUpdater<ConcurrentLinkedQueue, Node>

tailUpdater =

AtomicReferenceFieldUpdater.newUpdater

(ConcurrentLinkedQueue.class, Node.class, "tail");

private static final

AtomicReferenceFieldUpdater<ConcurrentLinkedQueue, Node>

headUpdater =

AtomicReferenceFieldUpdater.newUpdater

(ConcurrentLinkedQueue.class, Node.class, "head");

private boolean casTail(Node<E> cmp, Node<E> val) {

return tailUpdater.compareAndSet(this, cmp, val);

}

private boolean casHead(Node<E> cmp, Node<E> val) {

return headUpdater.compareAndSet(this, cmp, val);

}

* Pointer to header node, initialized to a dummy node. The first

* actual node is at head.getNext().

private transient volatile Node<E> head = new Node<E>(null, null);

/** Pointer to last node on list **/

private transient volatile Node<E> tail = head;

1.实际上经过对Node的分析。CLQ中的头尾指针的更新原理其实也是一样的。都是通过cpu原子操作命令进行的更新。

2.这样我们就有了在高并发下原子更新的基础支持，但是除了原子更新的支持是不够的。原因很简单，这是因为当多个线程同时使用原子更新操作来更新一个链表结构的时候只有一个成功其它的都会失败。失败的操作如何再让它成功才是问题的关键。CLQ优雅的解决了这一问题。

我们再来看看CLQ的放入元素操作：

Java代码

public boolean offer(E e) {

if (e == null) throw new NullPointerException();

Node<E> n = new Node<E>(e, null);

for (;;) { //1

Node<E> t = tail; //2

Node<E> s = t.getNext(); //3

if (t == tail) { //4

if (s == null) { //5

if (t.casNext(s, n)) { //6

casTail(t, n); //7

return true; //8

}

} else {

casTail(t, s); //9

}

}

}

}

[java]

public boolean offer(E e) {

if (e == null) throw new NullPointerException();

Node<E> n = new Node<E>(e, null);

for (;;) { //1

Node<E> t = tail; //2

Node<E> s = t.getNext(); //3

if (t == tail) { //4

if (s == null) { //5

if (t.casNext(s, n)) { //6

casTail(t, n); //7

return true; //8

}

} else {

casTail(t, s); //9

}

}

}

}

在有锁得情况下我们只要让获得锁得线程更新，其它线程等待即可解决并发更新的问题，但是在上述的单向链表结构中有更好的无锁解决方法。

1.代码1 啊！死循环，对，就是利用反复轮询的重复一段逻辑操作。

2.代码2 代码3 先用两个临时变量指向CLQ的尾和尾的下一个节点。这样有什么好处？直接用tail和tail.getNext不行吗？我们说了。这是一个无锁得方法。可能有多个线程同时执行到代码3处，因为临时变量是每线程的而tail是公共的。这样成功执行到代码3的线程都有自己当时的临时CLQ队列结构引用。为后面的判断做好准备。

3.开始判断代码4 证明在代码2 和代码4之间没有被其它线程修改过，因为有可能已经被修改了。那么这时进入新的轮询。

4.代码5 在看代码5之前我们先要明确一个概念就是把一个Node放入一个CLQ队列有两步操作。第一步是tail的next指向新的节点。第二步是tail指向新的节点。代码5 先判断是不是有线程已经在完成加入一个节点的第一步，如果是就帮助它完成第二步，再次进入循环。如果没有线程已经完成第一步。那就自己来完成插入节点的第一步，当然就是调用casNext比较更新的原子操作。上文已经讲过。再来完成插入元素的第二步，以上逻辑由代码6、代码7完成。注意看代码8 恒为真？为什么？自己调用casTail如果成功返回真毫无疑问。如果失败为什么也返回真？答案很简单，这是因为如果失败说明一定有其它线程进入了代码9 帮自己完成了插入一个节点的第二步操作。所以自己操作肯定是失败的。所以也返回真。

从上面的代码分析可以看出打造一个无锁得并发容器处处都要十分小心。这也是CLQ的高明之处。

我们再来看看删除一个元素的代码：

Java代码