源码阅读(37)：Java中线程安全的QueueDeque结构——PriorityBlockingQueue

Posted 2023-02-05 说好不能打脸

1、概述

PriorityBlockingQueue是一种无界阻塞队列，其内部核心结构和我们前文中已经介绍过的PriorityQueue队列集合类似，都是基于小顶堆树进行工作。本文不会赘述介绍PriorityQueue时已经详解过的内容（《源码阅读(12)：Java中主要的Queue、Deque结构——PriorityQueue集合》），例如小顶堆树的工作原理等。本文将集中精力在几个PriorityBlockingQueue队列集合的核心方法的介绍上，这些方法都是确保PriorityBlockingQueue队列在多线程场景下能正确工作的重要方法。

PriorityBlockingQueue队列本身的使用属于最基本的知识，并不是本文介绍的重点，和之前的规矩类似，给出一个简单的使用实例即可：

// ......
// 创建一个PriorityBlockingQueue对象
PriorityBlockingQueue<Integer> queue = new PriorityBlockingQueue<>(16 , new Comparator<Integer>() 
  @Override
  public int compare(Integer o1, Integer o2) 
    return o1 - o2;
  
);
// 向priorityQueue集合添加数据
queue.add(11);
queue.add(88);
queue.add(8);
queue.add(19);
queue.add(129);
// ......
queue.add(15);
queue.add(198);
queue.add(189);
queue.add(200);
// 从PriorityBlockingQueue集合移除数据
for (int index = 0 ; index < queue.size() ; ) 
  System.out.println("priorityBlockingQueue item = " + queue.poll());

// ......

注意，由于小顶堆的工作特点，所以PriorityBlockingQueue队列集合PriorityQueue队列集合一样，也只是保证数组头部将要去取的数据对象满足权值最小的要求。

2、PriorityBlockingQueue核心结构

2.1、PriorityBlockingQueue主要属性

为了了解PriorityBlockingQueue队列集合如何支持在多线程环境下的正常工作，我们需要首先分析一下这个队列集合具有哪些重要属性，如下所示：

// ......
public class PriorityBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable 
  // ......
  // 该常量用来描述该队列默认的初始化容量
  private static final int DEFAULT_INITIAL_CAPACITY = 11;
  
  // 实际上PriorityBlockingQueue队列本质上还是有界的（只是这个界限非常大），
  // 该常量用来描述该队列支持的最大容量上限
  private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
  /**
   * 就像上文中介绍PriorityQueue时提到的，其内部基于小顶堆树工作（小顶堆树是一种平衡二叉树），
   * 而小顶堆树在java的集合框架中，往往有以数组形式进行表单（一种树结构的降维表达）。
   * 所以PriorityBlockingQueue内部也是使用数组进行数据对象存储。树节点的左右儿子在数组中的索引定位具有以下特点：
   * 如果记当前节点在索引中的存储位置为n，其左儿子的索引位为2*n+1，其右儿子的索引位为2*(n+1)。
   */
  private transient Object[] queue;
  /**
   * 该值记录当前PriorityBlockingQueue队列集合的大小
   * 请注意容量和大小的区别。
   */
  private transient int size;
  /**
   * 用于当前队列集合中数据对象排序的比较器，如果该比较器为null
   * 那么将使用数据对象自带的比较器进行排序比较
   */
  private transient Comparator<? super E> comparator;
  // 通过这个锁对象，控制该队列集合所有公共写方法的线程安全性
  private final ReentrantLock lock = new ReentrantLock();
  // 该控制条件对象基于以上的可重入锁，方便在队列至少有一个数据对象时，唤醒可能处于阻塞状态的消费者线程
  // 或者在队列中没有数据对象时，让消费者线程进入阻塞状态
  @SuppressWarnings("serial") // Classes implementing Condition may be serializable.
  private final Condition notEmpty = lock.newCondition();
  // SpinLock 代表自旋锁，这个变量用于队列集合的扩容过程
  // 它保证扩容过程不会重复进行，且尽可能少的产生性能影响（不好想象？后文将会详细介绍）
  private transient volatile int allocationSpinLock;
  // 该属性仅用于PriorityBlockingQueue对象的序列化和反序列化过程
  // 这是一个巧妙的设计，避免多个JDK版本间进行对象序列化和反序列的过程中，发生兼容性问题
  private PriorityQueue<E> q;
  // ......
 
// ......

PriorityBlockingQueue通过ReentrantLock可重入锁保证多线程场景下队列集合的安全性，这个思路和之前我们讲解过的ArrayBlockingQueue、LinkedBlockingQueue等队列集合保证线程安全的性的思路大同小异。但是allocationSpinLock和q这两个属性幕后的设计思路值得我们在后文进行详细介绍。

2.2、PriorityBlockingQueue主要构造函数

PriorityBlockingQueue队列集合一共有4个构造函数，其中三个都很简单和PriorityQueue中的构造函数类似，如下所示：

// ......
public class PriorityBlockingQueue<E> extends AbstractQueue<E>
    implements BlockingQueue<E>, java.io.Serializable 
  // ......
  // 默认的构造函数，这是队列没有设置公共的比较器
  // 且队列集合初始化容量为11（DEFAULT_INITIAL_CAPACITY常量决定）
  public PriorityBlockingQueue() 
    this(DEFAULT_INITIAL_CAPACITY, null);
  

  // 使用该构造函数初始化对象时，
  // 队列没有设置公共的比较器，但是可以传入一个初始化的容量大小
  // 该初始化容量大小应该大于1，否则会抛出异常
  public PriorityBlockingQueue(int initialCapacity) 
    this(initialCapacity, null);
  

  // 以上两个构造函数，实际上都是对这个构造函数的调用
  // 可以设定公共的比较器，以及设定队列集合的初始化容量大小
  public PriorityBlockingQueue(int initialCapacity, Comparator<? super E> comparator) 
    if (initialCapacity < 1) 
      throw new IllegalArgumentException();
    
    this.comparator = comparator;
    this.queue = new Object[Math.max(1, initialCapacity)];
  
  // ......
 
// ......

以上三个构造函数都很简单，第四个构造函数：PriorityBlockingQueue(Collection<? extends E> c)，则是参考一个外部集合完成PriorityBlockingQueue对象的实例化，这个被参考的集合不能为null：

// ......
public PriorityBlockingQueue(Collection<? extends E> c) 
  // 如果该变量为true，说明经过处理逻辑，并不知道当前集合队列中的数据对象是否是有序的
  // true if not known to be in heap order
  boolean heapify = true; 
  // 如果该变量为true，说明经过处理逻辑，并不能排除当前集合队列中的数据对象没有为null的情况
  // 所以需要进行排查
  // true if must screen for nulls
  boolean screen = true;  
  // 如果条件成立，说明源集合的数据对象是有序排列的
  // 并且尝试获取SortedSet集合中可能存在的排序器
  if (c instanceof SortedSet<?>) 
    SortedSet<? extends E> ss = (SortedSet<? extends E>) c;
    this.comparator = (Comparator<? super E>) ss.comparator();
    heapify = false;
  
  // 如果条件成立，说明源集合本来就是一个PriorityBlockingQueue队列集合
  // 那么源集合中可能存在的排序器，就是新的PriorityBlockingQueue队列集合的排序器
  else if (c instanceof PriorityBlockingQueue<?>) 
    PriorityBlockingQueue<? extends E> pq = (PriorityBlockingQueue<? extends E>) c;
    this.comparator = (Comparator<? super E>) pq.comparator();
    // 不需要进行数据对象为null的排除操作
    screen = false;
    // 这个判定条件成立，说明当前的源集合c完全匹配PriorityBlockingQueue队列集合
    // 这是多余的判定吗？显然不是，例如源集合c如果是PriorityBlockingQueue的子类对象，这个判定结果就为false
    // exact match
    if (pq.getClass() == PriorityBlockingQueue.class) 
      heapify = false;
    
  
  
  // c不能为null，否则这里会报错
  // es记录了源集合c中的数据对象数组。
  Object[] es = c.toArray();
  int n = es.length;
  // 如果es并不是一个一维数组，那么通过这里的操作转换为一维数组
  // If c.toArray incorrectly doesn't return Object[], copy it.
  if (es.getClass() != Object[].class) 
    es = Arrays.copyOf(es, n, Object[].class);
  

  // ==========接下来开始进行数据对象的清理工作
  if (screen && (n == 1 || this.comparator != null)) 
    for (Object e : es) 
      if (e == null) 
        throw new NullPointerException();
      
    
  
  this.queue = ensureNonEmpty(es);
  this.size = n;
  // 如果装入当前PriorityBlockingQueue队列集合的数据对象数组需要被重新排列
  // 则通过该方法进行小顶堆排序
  if (heapify) 
    heapify();
  

// ......

2.3、PriorityBlockingQueue扩容过程

在讲解PriorityQueue队列对，我们就已经介绍过扩容过程，PriorityBlockingQueue队列集合和PriorityQueue队列集合的扩容过程都是类似的。不过由于PriorityBlockingQueue被设计工作在多线程场景下，所以其扩容操作针对这样的工作场景做了有针对性的优化，我们先来看一下PriorityBlockingQueue队列集合中关于扩容操作部分的代码：

// ......
/**
 * Tries to grow array to accommodate at least one more element
 * (but normally expand by about 50%), giving up (allowing retry)
 * on contention (which we expect to be rare). Call only while
 * holding lock.
 * 注意：官方注释中的允许重试，并不在该方法本身的处理逻辑内，
 * 而是在调用者的while()循环中
 */
private void tryGrow(Object[] array, int oldCap) 
  // tryGrow方法主要由offer(E)方法进行调用
  // 调用tryGrow方法的第一个操作，就是释放当前线程获取的锁操作权
  // 改用CAS思想进行扩容操作
  // must release and then re-acquire main lock
  lock.unlock();
  // 该变量将决定当前线程是否进行了实际的扩容操作
  Object[] newArray = null;
  // 从JDK9+开始，该判断条件变成了现有的语句，实际上和之前版本中使用UNSAFE.compareAndSwapInt()方法的目的一致：
  // 原子性的变更allocationSpinLock属性的值为1，保证成功设置allocationSpinLock为1的线程
  // 能进行真正的扩容操作
  if (allocationSpinLock == 0 && ALLOCATIONSPINLOCK.compareAndSet(this, 0, 1)) 
    try 
      // 实际的扩容逻辑和PriorityQueue集合队列的扩容逻辑一致：
      // 如果原始容量小于64，那么就进行双倍扩容（实际上是双倍容量+2）
      // 如果原始容量大于64，那么进行50%的扩容。
      int newCap = oldCap + ((oldCap < 64) ?
                              (oldCap + 2) : // grow faster if small
                              (oldCap >> 1));
      // 如果条件成立，说明扩容后，新的容量已经超过了最大允许的容量上限
      // 那么就以最大允许的容量作为新的容量
      if (newCap - MAX_ARRAY_SIZE > 0) 
        int minCap = oldCap + 1;
        if (minCap < 0 || minCap > MAX_ARRAY_SIZE) 
          throw new OutOfMemoryError();
        
        newCap = MAX_ARRAY_SIZE;
      
      // 基于扩容后的新的容量，初始化一个数组
      // 这个数组将在随后的操作中替换掉当前队列集合正在使用的queue数组
      if (newCap > oldCap && queue == array) 
        newArray = new Object[newCap];
      
     finally 
      // 操作完成后，变更allocationSpinLock属性为0
      // 以便进行下一次扩容操作。
      allocationSpinLock = 0;
     
  
  // 如果条件成立，说明当前操作线程没有获取到进行扩容的实际操作权
  // 这时让当前线程让出CPU资源（传统意义上讲的降低优先级）
  // 这样保证完成实际扩容操作的线程，能够随后抢占到锁权限
  // back off if another thread is allocating
  if (newArray == null) 
    Thread.yield();
  
  lock.lock();
  // 进行实际的扩容操作——进行数组拷贝
  if (newArray != null && queue == array) 
    queue = newArray;
    System.arraycopy(array, 0, newArray, 0, oldCap);
  

// ......

扩容操作的发生条件是一种极端场景——集合内用于存储数据对象的queue数组不够用了，且都是由队列的生产者线程发起扩容操作。这种场景下扩容操作者通过整个对象共享的可重入锁获取了操作权限，但实际上扩容操作只对数据添加操作有影响，对PriorityBlockingQueue队列集合的数据读取并没有影响。

为了使得扩容操作的同时，其它消费者线程能继续从队列中取出数据对象，所以扩容操作释放了可重入锁的占用状态。但这又引来一个新问题，既是可能有多个生产者线程同时调用扩容请求，而扩容请求又不能重复操作，否则会造成queue数组大小（队列容量上限）数值错误。

为了避免出现的新问题，PriorityBlockingQueue队列集合改用CAS原理控制多个生产者线程同时调用扩容操作——保证同一时间只有一个扩容请求得到实际操作，其余扩容操作保持自旋，直到扩容操作结束。

2.4、PriorityBlockingQueue序列化和反序列过程

PriorityBlockingQueue队列集合的序列化和反序列化过程也是一对很有趣的操作过程，从JDK 1.5+的版本到JDK 14的版本，虽然PriorityBlockingQueue队列集合的基本属性和和核心处理思想都没有太大变化，但是为了保证序列化后的信息能够被多个版本很好的兼容读取（反序列化），PriorityBlockingQueue队列集合中使用了一个取代的PriorityQueue对象，来实现这样的兼容性目标。

• 序列化过程

// ......
private void writeObject(java.io.ObjectOutputStream s) throws java.io.IOException 
  // 序列化操作也需要获得操作权限
  lock.lock();
  try 
    // 初始化一个PriorityQueue队列集合，帮助完成序列化过程
    q = new PriorityQueue<E>(Math.max(size, 1), comparator);
    // 将当前集合中的数据对象添加到这个PriorityQueue队列集合中
    q.addAll(this);
    // 然后将q属性引用的PriorityQueue队列集合进行序列化
    // 这样就达到了使用PriorityQueue保证兼容性
    s.defaultWriteObject();
   finally 
    // 序列化完成后 q 属性就没有用了，设为null
    // 最后释放操作权限
    q = null;
    lock.unlock();
  

// ......

• 反序列化过程

// ......
private void readObject(java.io.ObjectInputStream s) throws java.io.IOException, ClassNotFoundException 
  try 
    // 使用该方法从序列化信息中还原PriorityBlockingQueue队列集合中的基本属性
    // 其中就包括了q属性代表的PriorityQueue对象
    s.defaultReadObject();
    int sz = q.size();
    SharedSecrets.getJavaObjectInputStreamAccess().checkArray(s, Object[].class, sz);
    // 为新的PriorityBlockingQueue对象初始化queue数组
    this.queue = new Object[Math.max(1, sz)];
    // 为新的PriorityBlockingQueue对象初始化可能的比较器
    comparator = q.comparator();
    // 将PriorityQueue队列集合中的数据对象赋值到新的PriorityBlockingQueue队列集合中
    addAll(q);
   finally 
    // 反序列化过程结束后，q属性就完成了它的工作目标
    // 设置为null
    q = null;
  

// ......

2.5、PriorityBlockingQueue的典型操作方法

PriorityBlockingQueue中的多种典型操作方法，其核心逻辑思路已经在介绍PriorityQueue队列集合时进行了详细说明，这里不再进行赘述。