ConcurrentLinkedQueue 源码解析

简介

ConcurrentLinkedQueue 位于 java.util.concurrent 包下，是一个基于链表、无界、非阻塞的线程安全队列，采用 CAS（无锁） 算法（Michael-Scott 算法）实现高并发。

• 非阻塞、无锁：所有入队 / 出队都用 CAS 原子操作，不用 synchronized / Lock，线程不会阻塞，适合高并发生产者-消费者场景。
• 不允许 null（CAS 用 null 表示「节点已出队」，故不能混用）。
• 迭代器是弱一致（weakly consistent）的；size() 需要遍历链表，是 O(n) 且不是一个精确快照，生产环境应避免频繁调用。
• 与 LinkedBlockingQueue（基于锁、可指定容量）不同，它是无界且非阻塞的。

若需要「有界」或「阻塞等待」的并发队列，应使用 ArrayBlockingQueue / LinkedBlockingQueue / SynchronousQueue 等。

底层数据结构分析

底层是一个单向链表，头尾各用一个 volatile 节点指针。为了性能，尾指针 tail 采用懒更新策略：并不是每次入队都更新 tail，而是「每隔一个节点」才更新一次（hops），从而减少 CAS 竞争。

// 链表节点：item 和 next 都用 volatile，并用 CAS 修改
private static class Node<E> {
    volatile E item;
    volatile Node<E> next;
    // CAS 修改 item / next（基于 Unsafe）
}

private transient volatile Node<E> head; // 头指针（可能有已出队的哨兵节点）
private transient volatile Node<E> tail; // 尾指针（懒更新）

哨兵机制：poll 出队后，被取出的节点 item 置为 null（不立即从链表摘除），头指针后移；这样即便并发，head 指向的节点仍可能已被取空，需向后跳过。

核心成员变量

private transient volatile Node<E> head; // 头节点（可能指向已出队的哨兵）
private transient volatile Node<E> tail; // 尾节点（懒更新，可能落后）

源码分析

入队：offer（CAS 追加，尾指针懒更新）

public boolean offer(E e) {
    if (e == null) throw new NullPointerException(); // 不允许 null
    Node<E> n = new Node<E>(e);
    // 从 tail 出发（tail 可能落后，需要往后找真正的末节点 p）
    for (Node<E> t = tail, p = t;;) {
        Node<E> q = p.next;
        if (q == null) {
            // p 是末节点，尝试用 CAS 把 n 挂到 p.next
            if (p.casNext(null, n)) {
                // 每两个节点更新一次 tail（懒更新，减少竞争）
                if (p != t) casTail(t, n);
                return true;
            }
        }
        else if (p == q) // p 已被出队（自引用），从 head 重新开始
            p = (t != (t = tail)) ? t : head;
        else             // 往后推进
            p = (p != t && t != (t = tail)) ? t : q;
    }
}

出队：poll（CAS 取头，更新头指针）

public E poll() {
    for (Node<E> h = head, p = h, q;;) {
        E item = p.item;
        // 头节点有值且能 CAS 把 item 置空，则出队成功
        if (item != null && p.casItem(item, null)) {
            if (p != h) 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)                 // 遇到哨兵，从新 head 重来
            continue;
        else p = q;                      // 当前头已被取空，向后跳
    }
}

要点：

• 全程无锁，靠 volatile 可见性 + CAS 自旋保证并发正确，不会让线程挂起。
• head / tail 都允许「暂时落后」，算法保证最终一致，因此 size() 只能遍历得到近似值。

与 LinkedBlockingQueue / ArrayBlockingQueue 的对比

维度	ConcurrentLinkedQueue	LinkedBlockingQueue	ArrayBlockingQueue
实现	链表 + CAS（无锁）	链表 + 锁（ReentrantLock）	数组 + 锁（单锁）
是否阻塞	否	是（可指定容量）	是（有界）
有界	无界	可选有界	有界
并发度	高（无锁）	中（读写两把锁）	低（单锁）
size()	O(n) 近似	O(1) 精确	O(1) 精确

经典 Demo

import java.util.concurrent.ConcurrentLinkedQueue;

public class CLQDemo {
    public static void main(String[] args) {
        ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<>();
        queue.offer("a"); queue.offer("b");
        System.out.println(queue.poll()); // a
        System.out.println(queue.peek()); // b（不移除）
        System.out.println(queue.size()); // 1（生产环境慎用，O(n)）
    }
}