ConcurrentLinkedQueue 源码解读

你的名字 2020-11-07 02:00 779阅读 0赞

## 一、介绍 ##

ConcurrentLinkedQueue 是一个基于链接节点的无界线程安全队列，非阻塞，它采用先进先出的规则对节点进行排序，当我们添加一个元素的时候，它会添加到队列的尾部；当我们获取一个元素时，它会返回队列头部的元素。

ConcurrentLinkedQueue 采用非阻塞的方式实现线程安全队列，它采用了["wait-free"][wait-free]算法（即CAS算法）来实现。

![1153954-20190830083419343-954687132.png][]

ConcurrentLinkedQueue 由 head 节点和 tail 节点组成，每个节点（Node）由节点元素（item）和指向下一个节点（next）的引用组成，节点与节点之间就是通过这个 next 关联起来，从而组成一张链表结构的队列。默认情况下head节点存储的元素为 null，tail 节点等于 head 节点。

## 二、源码解读 ##

现在我们有了 head 和 tail 节点，如果按照我们平常的思维，head 节点即头节点，tail 节点即尾节点。那么入队列的时候，将 tail 的 next 节点设置为 newNode，将 newNode 设置为 tail；出队列的时候，将 head 节点元素返回，head 的 next 节点设置为 head。实现代码如下：

public boolean offer(E e) {
            if (e == null)
                throw new NullPointerException();
            Node<E> n = new Node<E>(e);
            for (; ; ) {
                Node<E> t = tail;
                if (t.casNext(null, n) && casTail(t, n)) {
                    return true;
                }
            }
        }

不要怀疑你的思维，这样的做法完全是可行的。

这样的做法 tail 节点永远作为队列的尾节点，head 节点永远为队列的头节点。实现代码量非常少，而且逻辑清晰和易懂。但是，这么做有个缺点，每次都需要使用循环 CAS 更新 tail 节点。所以 doug lea 为了减少 CAS 更新 tail 节点的次数，提高入队的效率，使用增加循环来控制 tail 节点的更新频率，并不是每次节点入队后都将 tail 节点更新成尾节点，而是当 tail 节点和尾节点不一致时（也就是循环两次）才更新 tail 节点。如下图：  
![1153954-20190830221051850-1275722096.png][]

想要读懂 ConcurrentLinkedQueue 的源码，最好先搞懂以下特质：

*  队列中任意时刻只有最后一个元素的 next 为 null
 *  head 和 tail 不会是 null（哨兵节点的设计）
 *  head 未必是队列中第一个元素（head指向的可能是一个已经被移除的元素）
 *  tail 未必是队列中最后一个元素（tail.next 可以不为 null）
 *  一旦某个元素的 item 变为 null，就意味着它不再是队列中的有效元素了，并且会将已删除节点的 next 指针指向自身。

### \- 入队列 ###

public boolean offer(E e) { // 1. checkNotNull(e); final Node<E> newNode = new Node<E>(e); for (Node<E> t = tail, p = t;;) { Node<E> q = p.next; // 2. if (q == null) { // 3. if (p.casNext(null, newNode)) { // 4. if (p != t) casTail(t, newNode); return true; } } // 5. else if (p == q) p = (t != (t = tail)) ? t : head; else // 6. p = (p != t && t != (t = tail)) ? t : q; } }

1.  队列中的元素不允许为 null，否则抛出一个 NullPointerException
2.  q == null 说明 p 是队列中最后一个元素
3.  CAS 设置 newNode 入队
4.  t 是当前线程读到的 tail 快照，p 是上面 CAS 队列中最后一个元素。当 p != t 时说明需要更新 tail 节点。如果 CAS 失败则说明 tail 已经被其它线程更新过了，这没关系。
5.  什么情况下 p == q 呢？只有当 p 元素已经不在队列中了，即 p == p.next。这时候怎么办呢？重新读取一次 tail 到快照 t。如果 t 发生变化，则从新的 tail 节点继续下去（注意这里的设值和 for 循环中的初始值一样，表明重新开始，继续尝试）。如果 t 没有发生变化，说明 tail 元素也已经在队列外了（因为队列是先进先出），这种情况下我们需要从 head 继续遍历下去。
6.  如果 p 与 t 相等，说明 p 不是尾节点，则让 p 继续向后移动一个节点； 如果 p 和 t 不相等，则说明已经经历至少两轮循环(仍然没有入队)， 则重新读取一次 tail 到 t，p = t（和上面 for 的初始值一样），表示重新开始尝试入队。

### \- 出队列 ###

public E poll() { restartFromHead: for (;;) { // 1. for (Node<E> h = head, p = h, q;;) { E item = p.item; // 2. if (item != null && p.casItem(item, null)) { // 3. if (p != h) updateHead(h, ((q = p.next) != null) ? q : p); return item; } // 4. else if ((q = p.next) == null) { updateHead(h, p); return null; } // 5. else if (p == q) continue restartFromHead; // 6. else p = q; } } }

1.  h 作为 head 的快照节点，p 初始设置为 head，q = null。
2.  CAS 成功将 item 设置为 null，表明已经移除了元素（注意这个时候该元素还没有真正移出队列，p = p.next 的动作在 updateHead 中完成）。这里的 item != null 判断也是为了尽可能避免无意义的CAS。
3.  当 p 不等于 h，说明 head 节点存在滞后性，需要更新 head 节点。如果 p.next 节点不等于 null，则把 head 设置为 p.next；如果 p.next == null，表明 p 已经是最后一个节点了，没办法，也只能将 head 放在 p 节点上了。updateHead 还会做一个动作 — p = p.next，把滞后的 p 节点正式移出队列。
4.  同第3点解释类似，如果 p.next == null，表明已经是最后一个节点了，则只能更新 head 为 p 节点，返回 null。
5.  什么情况下会出现 p == q 呢？即 p == p.next 。出现这种情况，表明其他线程提前完成， p 元素已经被移出队列。这时候再继续循环意义不大，所以干脆重新开始，重新读一次 head 到快照 h，再尝试移除元素。
6.  p 向后移一位（这时 q = p.next），继续尝试移除元素。

## 三、API 使用 ##

| 返回值   | 方法                              | 说明                                                |
| -------- | --------------------------------- | --------------------------------------------------- |
| boolean  | add(E e) / offer(E e)             | 在该队列的尾部插入指定的元素                        |
| boolean  | addAll(Collection<? extends E> c) | 按照集合迭代器返回的顺序追加到队列的末尾            |
| boolean  | contains(Object o)                | 队列中是否包含指定元素                              |
| boolean  | isEmpty()                         | 如果队列中部包含元素，则返回 true                   |
| Iterator | iterator()                        | 返回此队列中元素的迭代器，从头元素开始迭代          |
| E        | peek()                            | 检索但不删除队列的头部，如果此队列为空，则返回 null |
| E        | poll()                            | 检索并删除队列的头部，如果此队列为空，则返回 null   |
| boolean  | remove(Object o)                  | 从该队列中删除指定元素的单个实例（如果存在）        |
| int      | size()                            | 返回队列中元素的个数                                |
| T[]      | toArray(T[] a)                    | 队列转成指定类型的数组                              |

[wait-free]: http://www.cs.rochester.edu/~scott/papers/1996_PODC_queues.pdf
[1153954-20190830083419343-954687132.png]: /images/1604714402060.png
[1153954-20190830221051850-1275722096.png]: /images/1604714380078.png