Java ConcurrentLinkedQueue 原理与源码解析

结果或目标：理解并发队列的设计

JDK中的并发队列主要按实现方式分为两类：使用锁实现的阻塞队列和使用CAS非阻塞算法实现的非阻塞队列。ConcurrentLinkedQueue属于后者，适用于高并发、低延迟的生产者-消费者场景。它解决了在多线程环境下，如何在没有显式锁的情况下保证队列元素操作的原子性和可见性。

准备条件：数据结构基础

ConcurrentLinkedQueue的内部实现依赖于以下核心组件：

• 单向链表：队列由Node节点链接而成，新元素插入尾部，从头部获取元素。
• head和tail指针：volatile类型，分别指向链表的头节点和尾节点。这两个变量的更新是线程安全的，确保其他线程能立即看到最新的队列状态。
• Node节点：item字段存储实际数据，next字段指向下一个节点，两者均使用volatile修饰，保证内存可见性。
• CAS机制：基于Unsafe类提供的原子比较交换功能，用于在无锁条件下进行线程竞争和资源分配。

默认构造函数会创建一个包含两个空哨兵节点的链表，初始状态下head和tail都指向同一个空的Node(null)。

public ConcurrentLinkedQueue() {
    head = tail = new Node<E>(null);
}

真实操作顺序与核心方法解析

第1步：入队操作（offer/add）

目的：将新元素追加到队列末尾。
入口：调用offer(E e)方法。
动作：

1. 参数检查：如果传入元素为null，直接抛出NullPointerException。这是ConcurrentLinkedQueue与某些允许null值的集合不同的地方。
2. 创建节点：实例化新的Node对象，存入待插入的元素。
3. 定位尾部：从tail节点开始遍历，寻找当前真正的尾节点。由于多线程竞争，tail可能不是最新的尾节点，因此需要循环直到找到next为null的节点。
4. CAS插入：尝试通过casNext将当前找到的尾节点的next指向新节点。这是一个原子操作。如果成功，说明该线程赢得了竞争，可以将元素加入队列；如果失败，说明有其他线程已经修改了next指针，当前线程需重新从头开始尝试或调整策略。
5. 更新尾部：插入成功后，尝试更新全局tail指针以指向新节点，以加速后续线程的入队速度（“跳跃”优化）。

public boolean offer(E e) {
    checkNotNull(e);
    final Node<E> newNode = new Node<E>(e);
    for (Node<E> t = tail, p = t;;) {
        Node<E> q = p.next;
        if (q == null) {
            if (p.casNext(null, newNode)) {
                // CAS成功，执行尾部更新优化
                if (p != t) 
                    casTail(t, newNode);  
                return true;
            }
        } else if (p == q)
            p = (t != (t = tail)) ? t : head;
        else
            p = (p != t && t != (t = tail)) ? t : q;
    }
}

第2步：出队操作（poll）

目的：获取并移除队列头部的第一个有效元素。
入口：调用poll()方法。
动作：

1. 定位头部：从head指针出发，沿链表向前查找。需要注意的是，head指向的不一定是首个有数据的节点，因为之前的出队操作只清空了item而未移动head指针。
2. CAS清空：找到第一个item不为null的节点后，尝试通过casItem将该节点的item设置为null。如果CAS成功，则该元素被“删除”，返回原值；如果失败，说明有其它线程抢先取走了该元素，需继续向后查找。
3. 推进头指针：一旦成功取出一个元素，为了保持head的高效性，会通过updateHead方法将head指针移动到刚被取走的元素的下一个节点（即新的有效头节点）。这避免了head指针无限滞后。

public E poll() {
    restartFromHead:
    for (;;) {
        for (Node<E> h = head, p = h, q;;) {
            E item = p.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)
                continue restartFromHead;
            else
                p = q;
        }
    }
}

第3步：查看队头元素（peek）

目的：获取但不移除队列头部的第一个有效元素。
入口：调用peek()方法。
区别：
与poll类似，但省略了casItem操作，因此不会产生副作用。同时，它也会尽力将head指针推进到离第一个非空item最近的位置，以优化未来操作的性能。

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;
        }
    }
}

完成判断：如何验证正确性与性能特征

线程安全性：所有公开方法均在并发环境下保证了线性一致性，多个线程同时进行offer或poll不会导致数据丢失或死锁。
无界特性：offer方法永远返回true（除非参数为null），因为它不受容量限制，仅在系统资源耗尽前持续工作。
内存效率：出队后的孤立节点会被GC回收，但由于指针推进有一定的延迟，可能会出现轻微的内存占用浪费，这在长期运行的大规模场景中需要注意。

简要总结

ConcurrentLinkedQueue通过精巧的CAS算法和无锁设计，在牺牲一定CPU运算成本的前提下，避免了传统锁机制带来的上下文切换开销，提供了高效的并发访问能力。它在需要高性能、非阻塞行为的并行计算系统中扮演着重要角色。以上就是Java并发编程之ConcurrentLinkedQueue队列详情的详细内容，更多关于Java并发编程 ConcurrentLinkedQueue的资料请关注本站其它相关文章！