Java并发编程笔记之ConcurrentLinkedQueue源码探究

简介: JDK 中基于链表的非阻塞无界队列 ConcurrentLinkedQueue 原理剖析,ConcurrentLinkedQueue 内部是如何使用 CAS 非阻塞算法来保证多线程下入队出队操作的线程安全? ConcurrentLinkedQueue是线程安全的无界非阻塞队列,其底层数据结构是使用单向链表实现,入队和出队操作是使用我们经常提到的CAS来保证线程安全的。

JDK 中基于链表的非阻塞无界队列 ConcurrentLinkedQueue 原理剖析,ConcurrentLinkedQueue 内部是如何使用 CAS 非阻塞算法来保证多线程下入队出队操作的线程安全?

ConcurrentLinkedQueue是线程安全的无界非阻塞队列,其底层数据结构是使用单向链表实现,入队和出队操作是使用我们经常提到的CAS来保证线程安全的。

我们首先看一下ConcurrentLinkedQueue的类图结构先,好有一个内部逻辑有一个大概的印象,如下图所示:

可以清楚的看到ConcurrentLinkedQueue内部的队列是使用单向链表方式实现,类中两个volatile 类型的Node 节点分别用来存放队列的首位节点。

首先我们先来看一下ConcurrentLinkedQueue的构造函数,如下:


public ConcurrentLinkedQueue() {
   head = tail = new Node<E>(null);
}
AI 代码解读

通过无参构造函数可知默认头尾节点都是指向 item 为 null 的哨兵节点。

Node节点内部则维护一个volatile 修饰的变量item 用来存放节点的值,next用来存放链表的下一个节点,从而链接为一个单向无界链表,这就是单向无界的根本原因。如下图:

 

接下来看ConcurrentLinkedQueue 主要关注入队,出队,获取队列元素的方法的源码,如下所示:

1.首先看入队方法offer,offer 操作是在队列末尾添加一个元素,如果传递的参数是 null 则抛出 NPE 异常,否者由于 ConcurrentLinkedQueue 是无界队列该方法一直会返回 true。另外由于使用 CAS 无阻塞算法,该方法不会阻塞调用线程,其源码如下:

 

public boolean offer(E e) {
    //(1)e为null则抛出空指针异常
    checkNotNull(e);

   //(2)构造Node节点
    final Node<E> newNode = new Node<E>(e);

    //(3)从尾节点进行插入
    for (Node<E> t = tail, p = t;;) {

        Node<E> q = p.next;

        //(4)如果q==null说明p是尾节点,则执行插入
        if (q == null) {

            //(5)使用CAS设置p节点的next节点
            if (p.casNext(null, newNode)) {
                //(6)cas成功,则说明新增节点已经被放入链表,然后设置当前尾节点
                if (p != t)
                    casTail(t, newNode);  // Failure is OK.
                return true;
            }
        }
        else if (p == q)//(7)
            //多线程操作时候,由于poll操作移除元素后有可能会把head变为自引用,然后head的next变为新head,所以这里需要
            //重新找新的head,因为新的head后面的节点才是正常的节点。
            p = (t != (t = tail)) ? t : head;
        else
            //(8) 寻找尾节点
            p = (p != t && t != (t = tail)) ? t : q;
    }
}
AI 代码解读

 类图结构时候谈到构造队列时候参构造函数创建了一个 item 为 null 的哨兵节点,并且 head 和 tail 都是指向这个节点,下面通过图形结合来讲解下 offer 操作的代码实现。

  1.首先看一下,当一个线程调用offer(item)时候情况:首先代码(1)对传参判断空检查,如果为null 则抛出空指针异常,然后代码(2)则使用item作为构造函数参数创建一个新的节点,

代码(3)从队列尾部节点开始循环,目的是从队列尾部添加元素。如下图:

 

上图是执行代码(4)时候队列的情况,这时候节点 p , t ,head ,tail 同时指向了item为null的哨兵节点,由于哨兵节点的next节点为null,所以这里q指向也是null。

代码(4)发现q==null  则执行代码(5),通过CAS原子操作判断p 节点的next节点是否为null,如果为null 则使用节点newNode替换p 的next节点,

然后执行代码(6),由于 p == t ,所以没有设置尾部节点,然后退出offer方法,这时候队列的状态图如下:

 

上面讲解的是一个线程调用offer方法的情况下,如果多个线程同时调用,就会存在多个线程同时执行到代码(5),假设线程A调用offer(item1),

线程B调用offer(item2),线程 A 和线程B同时到 p.casNext(null,newNode)。而CAS的比较并设置操作是原子性的,假设线程A先执行了比较设置操作,

则发现当前P的next节点确实是null ,则会原子性更新next节点为newNode,这时候线程B 也会判断p 的next节点是否为null,结果发现不是null,(因为线程 A 已经设置了 p 的 next 为 newNode)则会跳到代码(3),

然后执行到代码(4)的时候的队列分布图如下:

 根据这个状态图可知线程B会执行代码(8),然后q 赋值给了p,这个时候状态图为:

然后线程B再次跳转到代码(3)执行,当执行到代码(4)时候队列状态图为:

由于这时候q == null ,所以线程B 会执行步骤(5),通过CAS操作判断 当前p的next 节点是否为null ,不是则再次循环后尝试,是则使用newNode替换,假设CAS成功了,那么执行步骤(6),

由于 p != t 所以设置tail节点为newNode ,然后退出offer方法。这时候队列的状态图为:

到现在为止,offer代码在执行路径现在就差步骤(7)还没有执行过,其实这个要在执行poll操作才会出现的,这里先看一下执行poll操作后可能会存在的一种情况,如下图所示:

下面分析下当队列处于这种状态调用offer添加元素代码执行到代码(4)的时候的队列状态图,如下:

由于q节点不为空并且p==q 所以执行代码(7),因为 t == tail所以p 被赋值为head ,然后进入循环,循环后执行到代码(4)的时候的队列状态图,如下:

由于 q ==null,所以执行代码(5),进行CAS操作,如果当前没有其他线程执行offer操作,则CAS操作会成功,p的next节点被设置为新增节点,然后执行代码(6),

由于p != t 所以设置新节点为队列尾节点,现在队列状态图,如下:

在这里的自引用的节点会被垃圾回收掉,可见offer操作里面关键步骤是代码(5)通过原子CAS操作来进行控制同时只有一个线程可以追加元素到队列末尾,进行cas竞争失败的线程,

则会通过循环一次次尝试进行cas操作,知道cas成功才会返回,也就是通过使用无限循环里面不断进行CAS尝试方式来替代阻塞算法挂起调用线程,相比阻塞算法,这是使用CPU资源换取阻塞带来的开销。

 

  2.poll操作,poll 操作是在队列头部获取并且移除一个元素,如果队列为空则返回 null,我们首先看改方法的源码,如下:


public E poll() {
    //(1) goto标记
    restartFromHead:

    //(2)无限循环
    for (;;) {
        for (Node<E> h = head, p = h, q;;) {

            //(3)保存当前节点值
            E item = p.item;

            //(4)当前节点有值则cas变为null
            if (item != null && p.casItem(item, null)) {
                //(5)cas成功标志当前节点以及从链表中移除
                if (p != h) 
                    updateHead(h, ((q = p.next) != null) ? q : p);
                return item;
            }
            //(6)当前队列为空则返回null
            else if ((q = p.next) == null) {
                updateHead(h, p);
                return null;
            }
            //(7)自引用了,则重新找新的队列头节点
            else if (p == q)
                continue restartFromHead;
            else//(8)
                p = q;
        }
    }
 }
  final void updateHead(Node<E> h, Node<E> p) {
        if (h != p && casHead(h, p))
            h.lazySetNext(h);
    }
AI 代码解读


poll操作是从队头获取元素,所以代码(2)内层循环是从head节点开始迭代,代码(3)获取当前队头的节点,当队列一开始为空的时候队列状态为:

由于head 节点指向的item 为null 的哨兵节点,所以会执行到代码(6),假设这个过程没有线程调用offer,则此时q等于null  ,如下图:

所以执行updateHead方法,由于h 等于 p所以没有设置头节点,poll方法直接返回null。

假设执行到代码(6)的时候已经有其他线程调用了offer 方法成功添加了一个元素到队列,这时候q执行的是新增元素的节点,这时候队列状态图为:

所以代码(6)判断结果为false,然后会转向代码(7)执行,而此时p不等于q,所以转向代码(8)执行,执行结果是p指向了节点q,此时的队列状态如下:

然后程序转向代码(3)执行,p现在指向的元素值不为null,则执行p.casItem(item, null) 通过 CAS 操作尝试设置 p 的 item 值为 null,

如果此时没有其他线程进行poll操作,CAS成功则执行代码(5),由于此时 p != h ,所以设置头节点为p,poll然后返回被从队列移除的节点值item。此时队列状态为:

这个状态就是前面提到offer操作的时候,offer代码的执行路径(7)执行的前提状态。

假如现在一个线程调用了poll操作,则在执行代码(4)的时候的队列状态为:

可以看到这时候执行代码(6)返回null。

现在poll的代码还有个分支(7)还没有被执行过,那么什么时候会执行呢?假设线程A执行poll操作的时候,当前的队列状态,如下:

那么执行p.casItem(item, null) 通过 CAS 操作尝试设置 p 的 item 值为 null。

假设 CAS 设置成功则标示该节点从队列中移除了,此时队列状态为:

然后由于p != h,所以会执行updateHead 方法,假如线程A执行updateHead前,另外一个线程B开始poll操作,这时候线程B的p指向head节点,

但是还没有执行到代码(6),这时候队列状态为:

然后线程A执行 updateHead 操作,执行完毕后线程 A 退出,这时候队列状态为:

然后线程B继续执行代码(6)q=p.next由于该节点是自引用节点所以p==q,所以会执行代码(7)跳到外层循环restartFromHead,重新获取当前队列队头 head, 现在状态为:

 

总结:poll元素移除一个 元素的时候,只是简单的使用CAS操作把当前节点的item值设置为null,然后通过重新设置头节点让该元素从队列里面摘除,

被摘除的节点就成了孤立节点,这个节点会被在GC的时候会被回收掉。另外,执行分支中如果发现头节点被修改了要跳到外层循环重新获取新的头节点。

 

  3.peek操作,peek 操作是获取队列头部一个元素(只不获取不移除),如果队列为空则返回 null,其源码如下:


public E peek() {
   //(1)
    restartFromHead:
    for (;;) {
        for (Node<E> h = head, p = h, q;;) {
            //(2)
            E item = p.item;
            //(3)
            if (item != null || (q = p.next) == null) {
                updateHead(h, p);
                return item;
            }
            //(4)
            else if (p == q)
                continue restartFromHead;
            else
            //(5)
                p = q;
        }
    }
}
AI 代码解读


代码结构与poll操作类似,不同于代码(3)的使用只是少了castItem 操作,其实这很正常,因为peek只是获取队列头元素值,并不清空其值,

根据前面我们知道第一次执行 offer 后 head 指向的是哨兵节点(也就是 item 为 null 的节点),那么第一次peek的时候,代码(3)中会发现item==null,

然后会执行 q = p.next, 这时候 q 节点指向的才是队列里面第一个真正的元素或者如果队列为 null 则 q 指向 null。

 

当队列为空的时候,队列状态图,如下:

这时候执行updateHead 由于 h 节点等于 p 节点所以不进行任何操作,然后 peek 操作会返回 null。

当队列中至少有一个元素的时候(假如只有一个),这时候队列状态为:

这时候执行代码(5)这时候 p 指向了 q 节点,然后执行代码(3)这时候队列状态为:

执行代码(3)发现 item 不为 null,则执行 updateHead 方法,由于 h!=p, 所以设置头结点,设置后队列状态为:

可以看到其实就是剔除了哨兵节点。

 

总结:peek操作代码与poll操作类似,只是前者只获取队列头元素,但是并不从队列里面删除,而后者获取后需要从队列里面删除,另外,在第一次调用peek操作的时候,

会删除哨兵节点,并让队列的head节点指向队列里面第一个元素或者null。

 

  4.size方法,获取当前队列元素个数,在并发环境下不是很有用,因为 CAS 没有加锁所以从调用 size 函数到返回结果期间有可能增删元素,导致统计的元素个数不精确。源码如下:


public int size() {
    int count = 0;
    for (Node<E> p = first(); p != null; p = succ(p))
        if (p.item != null)
            // 最大返回Integer.MAX_VALUE
            if (++count == Integer.MAX_VALUE)
                break;
    return count;
}
//获取第一个队列元素(哨兵元素不算),没有则为null
Node<E> first() {
    restartFromHead:
    for (;;) {
        for (Node<E> h = head, p = h, q;;) {
            boolean hasItem = (p.item != null);
            if (hasItem || (q = p.next) == null) {
                updateHead(h, p);
                return hasItem ? p : null;
            }
            else if (p == q)
                continue restartFromHead;
            else
                p = q;
        }
    }
}
//获取当前节点的next元素,如果是自引入节点则返回真正头节点
final Node<E> succ(Node<E> p) {
    Node<E> next = p.next;
    return (p == next) ? head : next;
}
AI 代码解读


 

  5.remove方法,如果队列里面存在该元素则删除给元素,如果存在多个则删除第一个,并返回 true,否者返回 false。源码如下:


public boolean remove(Object o) {

    //查找元素为空,直接返回false
    if (o == null) return false;
    Node<E> pred = null;
    for (Node<E> p = first(); p != null; p = succ(p)) {
        E item = p.item;

        //相等则使用cas值null,同时一个线程成功,失败的线程循环查找队列中其它元素是否有匹配的。
        if (item != null &&
            o.equals(item) &&
            p.casItem(item, null)) {

            //获取next元素
            Node<E> next = succ(p);

            //如果有前驱节点,并且next不为空则链接前驱节点到next,
            if (pred != null && next != null)
                pred.casNext(p, next);
            return true;
        }
        pred = p;
    }
    return false;
}
AI 代码解读


 

ConcurrentLinkedQueue 底层使用单向链表数据结构来保存队列元素,每个元素被包装为了一个 Node 节点,队列是靠头尾节点来维护的,创建队列时候头尾节点指向一个 item 为 null 的哨兵节点,

第一次 peek 或者 first 时候会把 head 指向第一个真正的队列元素。由于使用非阻塞 CAS 算法,没有加锁,所以获取 size 的时候有可能进行了 offer,poll 或者 remove 操作,导致获取的元素个数不精确,所以在并发情况下 size 函数不是很有用。

 

目录
打赏
0
0
0
0
54
分享
相关文章
2025 更新必看:Java 编程基础入门级超级完整版指南
本教程为2025更新版Java编程基础入门指南,涵盖开发环境搭建(SDKMAN!管理JDK、VS Code配置)、Java 17+新特性(文本块、Switch表达式增强、Record类)、面向对象编程(接口默认方法、抽象类与模板方法)、集合框架深度应用(Stream API高级操作、并发集合)、模式匹配与密封类等。还包括学生成绩管理系统实战项目,涉及Maven构建、Lombok简化代码、JDBC数据库操作及JavaFX界面开发。同时提供JUnit测试、日志框架使用技巧及进阶学习资源推荐,助你掌握Java核心技术并迈向高级开发。
140 5
java 编程基础入门级超级完整版教程详解
这份文档是针对Java编程入门学习者的超级完整版教程,涵盖了从环境搭建到实际项目应用的全方位内容。首先介绍了Java的基本概念与开发环境配置方法,随后深入讲解了基础语法、控制流程、面向对象编程的核心思想,并配以具体代码示例。接着探讨了常用类库与API的应用,如字符串操作、集合框架及文件处理等。最后通过一个学生成绩管理系统的实例,帮助读者将理论知识应用于实践。此外,还提供了进阶学习建议,引导学员逐步掌握更复杂的Java技术。适合初学者系统性学习Java编程。资源地址:[点击访问](https://pan.quark.cn/s/14fcf913bae6)。
137 2
Java 集合面试题从数据结构到 HashMap 源码剖析详解及长尾考点梳理
本文深入解析Java集合框架,涵盖基础概念、常见集合类型及HashMap的底层数据结构与源码实现。从Collection、Map到Iterator接口,逐一剖析其特性与应用场景。重点解读HashMap在JDK1.7与1.8中的数据结构演变,包括数组+链表+红黑树优化,以及put方法和扩容机制的实现细节。结合订单管理与用户权限管理等实际案例,展示集合框架的应用价值,助你全面掌握相关知识,轻松应对面试与开发需求。
113 3
2025 年 Java 秋招面试必看 Java 并发编程面试题实操篇
Java并发编程是Java技术栈中非常重要的一部分,也是面试中的高频考点。本文从基础概念、关键机制、工具类、高级技术等多个方面进行了介绍,并提供了丰富的实操示例。希望通过本文的学习,你能够掌握Java并发编程的核心知识,在面试中取得好成绩。同时,在实际工作中,也能够运用这些知识设计和实现高效、稳定的并发系统。
40 0
2025 年 Java 秋招面试必看的 Java 并发编程面试题汇总
文章摘要: 本文系统梳理Java并发编程核心知识点,助力2025年秋招面试。内容涵盖:1)基础概念,包括线程/进程区别、创建线程的3种方式(Thread/Runnable/Callable)、6种线程状态及转换;2)关键机制,对比sleep()与wait()的锁行为差异,解释start()而非run()启动线程的原因;3)工具类与典型应用场景。通过技术原理与代码示例结合的方式,帮助开发者深入理解并发模型、线程同步等核心问题,为高并发系统设计打下坚实基础。(150字)
67 0
Java并发编程之Future与FutureTask
本文深入解析了Future接口及其实现类FutureTask的原理与使用。Future接口定义了获取任务结果、取消任务及查询任务状态的规范,而FutureTask作为其核心实现类,结合了Runnable与Future的功能。文章通过分析FutureTask的成员变量、状态流转、关键方法(如run、set、get、cancel等)的源码,展示了异步任务的执行与结果处理机制。最后,通过示例代码演示了FutureTask的简单用法,帮助读者更直观地理解其工作原理。适合希望深入了解Java异步编程机制的开发者阅读。
Java多线程基础
本文主要讲解多线程相关知识,分为两部分。第一部分涵盖多线程概念(并发与并行、进程与线程)、Java程序运行原理(JVM启动多线程特性)、实现多线程的两种方式(继承Thread类与实现Runnable接口)及其区别。第二部分涉及线程同步(同步锁的应用场景与代码示例)及线程间通信(wait()与notify()方法的使用)。通过多个Demo代码实例,深入浅出地解析多线程的核心知识点,帮助读者掌握其实现与应用技巧。
|
5月前
|
【Java并发】【线程池】带你从0-1入门线程池
欢迎来到我的技术博客!我是一名热爱编程的开发者,梦想是编写高端CRUD应用。2025年我正在沉淀中,博客更新速度加快,期待与你一起成长。 线程池是一种复用线程资源的机制,通过预先创建一定数量的线程并管理其生命周期,避免频繁创建/销毁线程带来的性能开销。它解决了线程创建成本高、资源耗尽风险、响应速度慢和任务执行缺乏管理等问题。
298 60
【Java并发】【线程池】带你从0-1入门线程池
|
3月前
|
【源码】【Java并发】从InheritableThreadLocal和TTL源码的角度来看父子线程传递
本文涉及InheritableThreadLocal和TTL,从源码的角度,分别分析它们是怎么实现父子线程传递的。建议先了解ThreadLocal。
145 4
【源码】【Java并发】从InheritableThreadLocal和TTL源码的角度来看父子线程传递

热门文章

最新文章

AI助理

你好,我是AI助理

可以解答问题、推荐解决方案等

登录插画

登录以查看您的控制台资源

管理云资源
状态一览
快捷访问