源码阅读(30)：Java中线程安全的QueueDeque结构——概述

Posted 2022-12-13 说好不能打脸

1、概述

如果要将java.util.concurrent工具包中的各种工具类进行详细的功能分类，那么在这个工具包中可以将“队列”性质的工具类专门作为一个独立的功能分类。为了适应高并发的程序工作场景，java.util.concurrent工具提供了丰富用于高并发场景的，线程安全的Queue/Deque结构集合，整体类结构可由下图进行描述：

在上文中我们已经介绍了队列的基本工作特点：允许在队列的head（头）位置取出元素，并且只允许在队列的tail（尾）位置添加元素，也就是说先进入队列的元素会先从队列取出（先进先出FIFO）。除此之外，如上图所示的这些队列都有一些自身的工作特点：

• ArrayBlockingQueue：这是一种内部基于数组的，使用在高并发场景下的阻塞队列，也是一种容量有界的队列。这个队列最显著的工作特性是，存储在队列中的总元素数量有一个最大值，这个队列的使用场景也非常丰富，后文将进行详细讲解。

• LinkedBlockingQueue：这是一种内部基于链表的，使用在高并发场景下的阻塞队列，是一种容量无界的队列。这个队列最显著的工作特点就是他的内部结构是一个链表，这保证了它可以在有界队列和无界队列间非常方便的进行转换。

• ConcurrentLinkedQueue：和LinkedBlockingQueue相比，这也是一种内部基于链表的，可以在有更高性能要求的场景下使用的容量无界的，体现先进先出工作特点的队列。但它不是一种阻塞队列，其内部主要也是使用CAS思想进行实现，通过我们对java parking锁机制分析的相关内容可以知道，CAS思想在大多数情况下比基于parking锁机制实现的工具类，工作性能要高（但也不是绝对的）。

• LinkedTransferQueue：这是一个基于“链表”的，可以在高并发场景下使用的阻塞队列，它是一种无界队列。可以将它看成LinkedBlockingQueue和ConcurrentLinkedQueue两者优点的结合体，既关注集合的读写性能，又维持队列集合的工作特性。

• PriorityBlockingQueue：这是一种内部基于数组的，采用小顶堆结构的，可以在高并发场景下使用的阻塞队列，它也是一种容量无界的队列。这个队列最显著的工作特点是，队列集合中的元素将按照堆树结构进行排序，以保证从该队列取出的元素都是集合中权值最小的元素。

• DelayQueue：这是一种内部依赖PriorityQueue的，采用小顶堆结构的，可以在高并发场景下使用的阻塞队列，它也是一种容量无界的队列。这个队列最显著的工作特点是，队列集合中的元素除了将按照堆树结构进行排序外，这些元素还通过实现java.util.concurrent.Delayed接口，定义一个延迟时间，只有当权值最小的元素的延迟时间小于等于0时，该元素才会被外部调用者获取到（这是一个实现租约协议的很好的基础思想，不过还差了一些要点，本专题后续文章会进行说明）。

• SynchronousQueue：这是一个内部“只能存储”一个元素的阻塞队列（基于），很明显它是一个有界队列。这个队列最显著的工作特点是，一个调用者向该队列放入一个元素后，就会进入阻塞状态，直到另一个调用者将队列中的这个元素取出。同样来说，如果一个调用者需要从该队列中取出一个元素，但队列中有没有元素，那么该调用者也会进去阻塞状态，直到另一个调用者向该队列放入一个元素位置——总结来说，就是向队列放入元素和取出元素的调用者要成对出现。

2、Queue/Deque接口和BlockQueue/BlockingDeque接口

这里我们对什么叫阻塞队列什么叫非阻塞队列，什么叫有界队列什么叫无界队列进行说明。

2.1、什么是有界队列，什么是无界队列

首先有界队列和无界队列的说法是在多线程、高并发场景下对队列容量特点的描述。

• 有界队列：即队列容量有一个固定大小的容量上限，一旦队列中的元素总量达到最大容量时，队列就会对添加操作做容错性处理，如返回false证明操作失败、抛出运行时异常、进入阻塞状态直到操作条件满足要求等——一句话，不再允许元素被添加了。

• 无界队列：即队列容量可以没有一个固定大小的容量上限，或者容量上限是一个很大的理论上限值（例如常量Integer.MAX_VALUE，就大值为2147483647）。由于这种队列理论上没有容量上限，所以理论上调用者可以将任意多的元素添加到集合中，而不会引起添加操作出现容量异常。

这里要注意，无界队列是不是真的无界呢？显然不是的，首先通过上文的定义我们可知一部分无界队列是可以在队列实例化时设置其队列容量上限的，例如LinkedBlockingQueue这个队列默认的容量大小是Integer.MAX_VALUE（相当于无界），但是我们也可以设定LinkedBlockingQueue队列容量为一个特定的值。

其次，既然无界队列可以不设定固定大小的容量上限，换句话说就是无界队列也可以设定固定大小的容量上限，例如以上例举的LinkedBlockingQueue队列，就可以设定一个容量上限，从而变成有界队列。

最后，无界队列也不可能保证其容量无限大，因为JVM可申请的堆内存是有上限的，当超过堆内存容量且JVM无法再申请新的内存空间时，应用程序就会抛出OutofMemoryError异常。

2.2、什么是阻塞队列，什么是非阻塞队列

Queue接口是BlockQueue接口的父级接口，前者定义了一些和队列相关的接口，后者在此基础上又补充了另一些接口功能，Queue接口主要的功能方法如下所示：

// 以下是java.util.Queue接口的主要定义
public interface Queue<E> extends Collection<E> 
  // 这是一种添加操作，如果不违反队列集合的容量限制要求，则立即向集合添加新的元素，并返回true
  // 其它情况则抛出IllegalStateException异常，添加失败
  boolean add(E e);
  
  // 这也是一种添加操作，如果不违反队列集合的容量限制要求，则立即向集合添加新的元素，并返回true
  // 其它添加失败的情况，则返回false（不是抛出异常）
  // add方法和offer方法，在添加操作的边界限制中，也有一些共同的限制，例如如果添加的元素是null，则都会抛出NullPointerException运行时异常
  // 再例如，如果添加的元素类型不符合要求，则会抛出ClassCastException运行时异常
  boolean offer(E e);
  
  // 这是一种移除操作，该操作将从队列集合头部移除操作，移除的元素将被返回给调用者
  // 如果操作时，队列集合中没有元素，则抛出NoSuchElementException异常
  E remove();

  // 这也是一种移除操作，该操作将从队列集合头部移除操作，移除的元素将被返回给调用者
  // 如果操作时，队列集合中没有元素，则返回null。
  E poll();
  
  // 这是一种查询操作，该操作将查询当前队列集合头部的元素（但不会移除），并进行返回
  // 如果操作时，队列集合中没有元素，则抛出NoSuchElementException异常
  E element();
  
  // 这也是一种查询操作，该操作将查询当前队列集合头部的元素（但不会移除），并进行返回
  // 如果操作时，队列集合中没有元素，则返回null。
  E peek();

通过以上源代码，我们可以知道，java.util.Queue接口中主要定义了六个方法，这六个方法可以分为两类：一类是在操作时如果队列集合的容量状态不符合要求，就会抛出异常；另一类是在操作时如果集合的状态不符合要求，则尽可能不抛出异常——使用返回一些特定的值进行替换。通过下表我们可以对这6个方法进行详细分类

操作类型	抛出异常的操作	返回特定值的操作（null或者false）
添加操作	add	offer
删除操作	remove	poll
查询操作	element	peek

这里要特别说明的情况是，抛出异常的方法在抛出诸如NoSuchElementException、IllegalStateException等异常时，其关注的点只是队列集合的容量状态，而其它异常的抛出要求是没有区别的。所以在阅读源代码时会出现的有趣情况是，那些无界队列的实现中，往往可以看到其add方法直接调用了offer方法；那些有界队列中才对offer方法返回false的情况进行了异常抛出处理。

例如我们后文将要详细介绍的ConcurrentLinkedQueue非阻塞队列，其add方法就是直接调用了offer方法——原因很简单，因为无界队列本来就对容器的上限容量没有限制，所以也就不存在添加操作会超界的场景，如下所示：

public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
        implements Queue<E>, java.io.Serializable 
  // ......
  public boolean add(E e) 
    return offer(e);
  
  // ......

堆内存不足的情况不在这里考虑，堆内存不足的情况自然会由JVM报告OutOfMemoryError。

再例如我们后文将要详细介绍的有界队列ArrayBlockingQueue，其内部的add方法就会offer方法返回值进行了特别判定，如下源代码段落所示：

public abstract class AbstractQueue<E> extends AbstractCollection<E> implements Queue<E> 
  // ......
  public boolean add(E e) 
    if (offer(e)) 
      return true;
    
    else 
      throw new IllegalStateException("Queue full");
    
  
  // ......


public class ArrayBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable 
  // ......
  public boolean add(E e) 
    return super.add(e);
  
  // ......

另外，我们可以发现java.util.Queue接口中主要定义了六个方法都是“实时”处理，也就是说调用者对这6个方法的调用都不会引起调用者所在线程的阻塞——无论调用者是向集合进行添加操作还是移除操作，又或者查询操作。这种工作特性的队列称为非阻塞队列。

BlockQueue接口中定义的一些方法和Queue接口中定义的部分方法重复，但是还新增了一些方法，那些两个接口重复的方法这里就不再进行赘述了，我们主要来分析一下那些在BlockQueue接口中新增加的方法定义，如下所示：

public interface BlockingQueue<E> extends Queue<E> 
  // 该方法将在有界队列的场景下，试图在给定的限制时间内，将一个元素添加到队列。
  // 如果超过限制时间前，仍然没有操作成功，则调用者所在线程会进入阻塞状态。
  // 如果操作成功，将返回true；其它情况返回false（例如超过限制时间）
  boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException;
  // 该方法将试图从队列头部取出元素，如果没有可以取得的元素，则调用者所在线程进入阻塞状态
  E take() throws InterruptedException;
  // 该方法试图在给定的限制时间内，将队列头的元素进行移除。
  // 如果超过限制时间前，仍然没有操作成功，则调用者所在线程会进入阻塞状态。
  // 如果操作成功，则返回被移除的元素；其它情况返回null（例如超过限制时间，队列中仍然没有任何可以移除的元素）
  E poll(long timeout, TimeUnit unit) throws InterruptedException;
  // 该方法从队列集合中移除指定的元素
  boolean remove(Object o);
  // 该方法从队列集合中“移动”所有元素到指定集合中，“移动”操作的意义在于这些元素将从原来的队列集合中删除
  // 注意：如果指定额元素移除集合和移入集合是同一个集合，则会抛出IllegalArgumentException异常。
  int drainTo(Collection<? super E> c);
 

这里请注意offer(E e, long timeout, TimeUnit unit)方法，该方法仅限于有界队列的场景，也就是说如果是无界队列则该方法的工作意义和offer(E e)方法的工作意义相同，所以读者可以发现无界队列的源代码实现中，这两个方法的实现逻辑相同。如下是PriorityBlockingQueue队列中，两个方法的实现：

public class PriorityBlockingQueue<E> extends AbstractQueue<E> 
  implements BlockingQueue<E>, java.io.Serializable 
  // ......
  public boolean offer(E e) 
    // ......
  
  
  // 该方法直接调用offer(E e)方法
  public boolean offer(E e, long timeout, TimeUnit unit) 
    return offer(e); // never need to block
  
  // ......

由此，我们可以给阻塞队列一个通俗的定义，即实现了java.util.concurrent.BlockingQueue接口的队列中有一组方法功能，当调用者通过这组方法对队列进行的读写操作不满足操作条件时，调用者所在线程将进入阻塞状态，直到操作条件被满足或者超过限制时间。

3、后文说明

实现了java.util.concurrent.BlockingQueue接口的队列是java线程安全的集合体系中非常重要的一组集合类型，后续文章我们将花费较多的篇幅对这些具体实现类进行详细介绍，包括但不限于：ArrayBlockingQueue、LinkedBlockingQueue、DelayQueue、PriorityBlockingQueue、LinkedTransferQueue、SynchronousQueue。

（接下文《源码阅读(31)：Java中线程安全的Queue、Deque结构——ArrayBlockingQueue阻塞队列》）