Java Review - 并发编程_ConcurrentLinkedQueue原理&源码剖析

简介: Java Review - 并发编程_ConcurrentLinkedQueue原理&源码剖析


概述

JDK中提供了一系列场景的并发安全队列。总的来说,按照实现方式的不同可分为阻塞队列和非阻塞队列,

  • 阻塞队列使用锁实现
  • 非阻塞队列则使用CAS非阻塞算法实现



ConcurrentLinkedQueue

ConcurrentLinkedQueue是线程安全的无界非阻塞队列,其底层数据结构使用单向链表实现,对于入队和出队操作使用CAS来实现线程安全。

【类图】

ConcurrentLinkedQueue内部的队列使用单向链表方式实现,

其中有两个volatile类型的Node节点分别用来存放队列的首、尾节点。

从下面的无参构造函数可知,默认头、尾节点都是指向item为null的哨兵节点。 新元素会被插入队列末尾,出队时从队列头部获取一个元素。

在Node节点内部则维护一个使用volatile修饰的变量item,用来存放节点的值;next用来存放链表的下一个节点,从而链接为一个单向无界链表。其内部则使用UNSafe工具类提供的CAS算法来保证出入队时操作链表的原子性。


核心方法&源码解读

下面我们介绍ConcurrentLinkedQueue的几个主要方法的实现原理。


offer

在链表末尾添加一个元素

/**
     * Inserts the specified element at the tail of this queue.
     * As the queue is unbounded, this method will never return {@code false}.
     *
     * @return {@code true} (as specified by {@link Queue#offer})
     * @throws NullPointerException if the specified element is null
     */
public boolean offer(E e) {
    //1  e为null则抛出空指针异常
    checkNotNull(e);
   //2 构造Node节点构造函数内部调用unsafe.putObject,后面统一讲
    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成功说明新增节点已经被放入链表,然后设置当前尾节点(包含head,1,3,5.。。个节点为尾节点)
                if (p != t) // hop two nodes at a time
                    casTail(t, newNode);  // Failure is OK.
                return true;
            }
            // Lost CAS race to another thread; re-read next
        }
        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;
    }
}
  • 首先看当一个线程调用offer(item)时的情况。首先代码(1)对传参进行空检查,如果为null则抛出NPE异常,否则执行代码(2)并使用item作为构造函数参数创建一个新的节点,然后代码(3)从队列尾部节点开始循环,打算从队列尾部添加元素,当执行到代码(4)时队列状态如下所示。

这时候节点p、t、head、tail同时指向了item为null的哨兵节点,由于哨兵节点的next节点为null,所以这里q也指向null。

  • q==null则执行代码(5),通过CAS原子操作判断p节点的next节点是否为null,如果为null则使用节点newNode替换p的next节点,然后执行代码(6),这里由于p==t所以没有设置尾部节点,然后退出offer方法,这时候队列的状态如下图所示

(2)上面是一个线程调用offer方法的情况,如果多个线程同时调用,就会存在多个线程同时执行到代码(5)的情况。假设线程A调用offer(item1),线程B调用offer(item2),同时执行到代码(5)p.casNext(null, newNode)。

由于CAS的比较设置操作是原子性的,所以这里假设线程A先执行了比较设置操作,发现当前p的next节点确实是null,则会原子性地更新next节点为 item1,这时候线程B也会判断p的next节点是否为null,结果发现不是null(因为线程A已经设置了p的next节点为 item1),则会跳到代码(3),然后执行到代码(4),这时候的队列分布如下图所示。

根据上面的状态图可知线程B接下来会执行代码(8),然后把q赋给了p,这时候队列状态如下图所示。

然后线程B再次跳转到代码(3)执行,当执行到代码(4)时队列状态如下图所示

由于这时候q==null,所以线程B会执行代码(5),通过CAS操作判断当前p的next节点是否是null,不是则再次循环尝试,是则使用item2替换。假设CAS成功了,那么执行代码(6),由于p!=t,所以设置tail节点为item2,然后退出offer方法。这时候队列分布如下图所示。

分析到现在,就差代码(7)还没走过,其实这一步要在执行poll操作后才会执行。这里先来看一下执行poll操作后可能会存在的一种情况,如下图所示。

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

这里由于q节点不为空并且pq所以执行代码(7),由于ttail所以p被赋值为head,然后重新循环,循环后执行到代码(4),这时候队列状态如下图所示。

这时候由于q==null,所以执行代码(5)进行CAS操作,如果当前没有其他线程执行offer操作,则CAS操作会成功,p的next节点被设置为新增节点。然后执行代码(6),由于p!=t所以设置新节点为队列的尾部节点,现在队列状态如图

需要注意的是,这里自引用的节点会被垃圾回收掉。

可见,offer操作中的关键步骤是代码(5),通过原子CAS操作来控制某时只有一个线程可以追加元素到队列末尾。进行CAS竞争失败的线程会通过循环一次次尝试进行CAS操作,直到CAS成功才会返回,也就是通过使用无限循环不断进行CAS尝试方式来替代阻塞算法挂起调用线程。相比阻塞算法,这是使用CPU资源换取阻塞所带来的开销。



add

add操作是在链表末尾添加一个元素,其实在内部调用的还是offer操作’\

/**
     * Inserts the specified element at the tail of this queue.
     * As the queue is unbounded, this method will never throw
     * {@link IllegalStateException} or return {@code false}.
     *
     * @return {@code true} (as specified by {@link Collection#add})
     * @throws NullPointerException if the specified element is null
     */
    public boolean add(E e) {
        return offer(e);
    }



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 当前item有值,则CAS变为null
                if (item != null && p.casItem(item, null)) {
                    // Successful CAS is the linearization point
                    // for item to be removed from this queue.
                    // 5 cas成功则标记当前节点并从链表中移除
                    if (p != h) // hop two nodes at a time
                        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;
            }
        }
    }

poll方法在移除一个元素时,只是简单地使用CAS操作把当前节点的item值设置为null,然后通过重新设置头节点将该元素从队列里面移除,被移除的节点就成了孤立节点,这个节点会在垃圾回收时被回收掉。另外,如果在执行分支中发现头节点被修改了,要跳到外层循环重新获取新的头节点。



peek

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

public E peek() {
      // 1 
        restartFromHead:
        for (;;) {
          // 2 
            for (Node<E> h = head, p = h, q;;) {
                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;
            }
        }
    }

Peek操作的代码结构与poll操作类似,不同之处在于代码(3)中少了castItem操作。

其实这很正常,因为peek只是获取队列头元素值,并不清空其值。根据前面的介绍我们知道第一次执行offer后head指向的是哨兵节点(也就是item为null的节点),那么第一次执行peek时在代码(3)中会发现item==null,然后执行q=p.next,这时候q节点指向的才是队列里面第一个真正的元素,或者如果队列为null则q指向null。


总结:peek操作的代码与poll操作类似,只是前者只获取队列头元素但是并不从队列里将它删除,而后者获取后需要从队列里面将它删除。

另外,在第一次调用peek操作时,会删除哨兵节点,并让队列的head节点指向队列里面第一个元素或者null。


size

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

/**
     * Returns the number of elements in this queue.  If this queue
     * contains more than {@code Integer.MAX_VALUE} elements, returns
     * {@code Integer.MAX_VALUE}.
     *
     * <p>Beware that, unlike in most collections, this method is
     * <em>NOT</em> a constant-time operation. Because of the
     * asynchronous nature of these queues, determining the current
     * number of elements requires an O(n) traversal.
     * Additionally, if elements are added or removed during execution
     * of this method, the returned result may be inaccurate.  Thus,
     * this method is typically not very useful in concurrent
     * applications.
     *
     * @return the number of elements in this queue
     */
    public int size() {
        int count = 0;
        for (Node<E> p = first(); p != null; p = succ(p))
            if (p.item != null)
                // Collection.size() spec says to max out
                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;
            }
        }
    }



remove

如果队列里面存在该元素则删除该元素,如果存在多个则删除第一个,并返回true,否则返回false。

/**
     * Removes a single instance of the specified element from this queue,
     * if it is present.  More formally, removes an element {@code e} such
     * that {@code o.equals(e)}, if this queue contains one or more such
     * elements.
     * Returns {@code true} if this queue contained the specified element
     * (or equivalently, if this queue changed as a result of the call).
     *
     * @param o element to be removed from this queue, if present
     * @return {@code true} if this queue changed as a result of the call
     */
    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;
}



contains

判断队列里面是否含有指定对象,由于是遍历整个队列,所以像size 操作一样结果也不是那么精确,有可能调用该方法时元素还在队列里面,但是遍历过程中其他线程才把该元素删除了,那么就会返回false。

/**
     * Returns {@code true} if this queue contains the specified element.
     * More formally, returns {@code true} if and only if this queue contains
     * at least one element {@code e} such that {@code o.equals(e)}.
     *
     * @param o object to be checked for containment in this queue
     * @return {@code true} if this queue contains the specified element
     */
    public boolean contains(Object o) {
        if (o == null) return false;
        for (Node<E> p = first(); p != null; p = succ(p)) {
            E item = p.item;
            if (item != null && o.equals(item))
                return true;
        }
        return false;
    }

总结

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

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

如下图所示,入队、出队都是操作使用volatile修饰的tail、head节点,要保证在多线程下出入队线程安全,只需要保证这两个Node操作的可见性和原子性即可。由于volatile本身可以保证可见性,所以只需要保证对两个变量操作的原子性即可。

offer操作是在tail后面添加元素,也就是调用tail.casNext方法,而这个方法使用的是CAS操作,只有一个线程会成功,然后失败的线程会循环,重新获取tail,再执行casNext方法。poll操作也通过类似CAS的算法保证出队时移除节点操作的原子性


相关文章
|
3天前
|
Java
Java并发编程:深入理解线程池
【4月更文挑战第30天】本文将深入探讨Java并发编程中的一个重要主题——线程池。我们将从线程池的基本概念入手,了解其工作原理和优势,然后详细介绍如何使用Java的Executor框架创建和管理线程池。最后,我们将讨论一些高级主题,如自定义线程工厂和拒绝策略。通过本文的学习,你将能够更好地理解和使用Java的线程池,提高你的并发编程能力。
|
1天前
|
并行计算 安全 Java
Java 并发编程的探索之旅
【5月更文挑战第2天】 在多线程的世界里,程序的行为变得错综复杂。本文将带您走进 Java 并发编程的核心概念,通过深入分析并发工具的使用和原理,帮助您构建高效、安全且响应迅速的应用程序。我们将探讨线程的基本知识,同步机制,以及高级并发工具如 Executors、Futures 和 Streams。通过理论与实践相结合的方式,为开发者提供一份清晰、实用的并发编程指南。
6 2
|
2天前
|
存储 安全 Java
深入理解Java并发编程:线程安全与性能优化
【5月更文挑战第1天】本文将深入探讨Java并发编程的核心概念,包括线程安全和性能优化。我们将详细分析线程安全问题的根源,以及如何通过合理的设计和编码实践来避免常见的并发问题。同时,我们还将探讨如何在保证线程安全的前提下,提高程序的并发性能,包括使用高效的同步机制、减少锁的竞争以及利用现代硬件的并行能力等技术手段。
|
2天前
|
安全 Java 程序员
Java并发编程:理解并应用ReentrantLock
【4月更文挑战第30天】 在多线程的世界中,高效且安全地管理共享资源是至关重要的。本文深入探讨了Java中的一种强大同步工具——ReentrantLock。我们将从其设计原理出发,通过实例演示其在解决并发问题中的实际应用,以及如何比传统的synchronized关键字提供更灵活的锁定机制。文章还将讨论在使用ReentrantLock时可能遇到的一些挑战和最佳实践,帮助开发者避免常见陷阱,提高程序性能和稳定性。
|
2天前
|
缓存 Java 调度
Java并发编程:深入理解线程池
【4月更文挑战第30天】 在Java并发编程中,线程池是一种重要的工具,它可以帮助我们有效地管理线程,提高系统性能。本文将深入探讨Java线程池的工作原理,如何使用它,以及如何根据实际需求选择合适的线程池策略。
|
2天前
|
Java
Java并发编程:深入理解线程池
【4月更文挑战第30天】 本文将深入探讨Java中的线程池,解析其原理、使用场景以及如何合理地利用线程池提高程序性能。我们将从线程池的基本概念出发,介绍其内部工作机制,然后通过实例演示如何创建和使用线程池。最后,我们将讨论线程池的优缺点以及在实际应用中需要注意的问题。
|
3天前
|
存储 安全 Java
【亮剑】Java并发编程中的四个关键字:ThreadLocal、Volatile、Synchronized和Atomic
【4月更文挑战第30天】Java并发编程涉及`ThreadLocal`、`Volatile`、`Synchronized`和`Atomic`四个关键机制。`ThreadLocal`为每个线程提供独立变量副本;`Volatile`确保变量可见性,但不保证原子性;`Synchronized`实现同步锁,保证单线程执行;`Atomic`类利用CAS实现无锁并发控制。理解其原理有助于编写高效线程安全代码。根据业务场景选择合适机制至关重要。
|
Java
Java并发编程笔记之FutureTask源码分析
FutureTask可用于异步获取执行结果或取消执行任务的场景。通过传入Runnable或者Callable的任务给FutureTask,直接调用其run方法或者放入线程池执行,之后可以在外部通过FutureTask的get方法异步获取执行结果,因此,FutureTask非常适合用于耗时的计算,主线程可以在完成自己的任务后,再去获取结果。
4259 0
|
Java 调度 API
Java并发编程笔记之Timer源码分析
timer在JDK里面,是很早的一个API了。具有延时的,并具有周期性的任务,在newScheduledThreadPool出来之前我们一般会用Timer和TimerTask来做,但是Timer存在一些缺陷,为什么这么说呢?   Timer只创建唯一的线程来执行所有Timer任务。
2966 0
|
Java
Java并发编程笔记之Semaphore信号量源码分析
JUC 中 Semaphore 的使用与原理分析,Semaphore 也是 Java 中的一个同步器,与 CountDownLatch 和 CycleBarrier 不同在于它内部的计数器是递增的,那么,Semaphore 的内部实现是怎样的呢?   Semaphore 信号量也是Java 中一个同步容器,与CountDownLatch 和 CyclicBarrier 不同之处在于它内部的计数器是递增的。
4250 0