Java并发编程中优先级队列PriorityBlockingQueue的用法详解

PriorityBlockingQueue是Java中实现了堆数据结构的线程安全的有界阻塞队列。它可以在多线程场景下安全地进行元素添加、删除和获取操作，而且可以根据元素的优先级进行排序。

一、PriorityBlockingQueue概述

PriorityBlockingQueue类实现了BlockingQueue接口，它是一个线程安全的队列，继承自AbstractQueue类，而AbstractQueue类又实现了Queue接口。PriorityBlockingQueue是一个有界的队列，其容量可以在构造函数中进行指定，若不指定则默认大小为Integer.MAX_VALUE。同时，PriorityBlockingQueue也支持根据元素的优先级进行排序，这是由于PriorityBlockingQueue内部实现了一个堆数据结构。

二、PriorityBlockingQueue源码解析

1.容器

PriorityBlockingQueue内部使用了一个Object类型的数组queue来存储元素，同时使用了一个int类型的变量size来记录元素的数量。下面是PriorityBlockingQueue类中的定义：

private transient Object[] queue;
private transient int size;

2.比较器

PriorityBlockingQueue可以根据元素的优先级进行排序，这是由于PriorityBlockingQueue内部维护了一个小根堆或大根堆。在构造函数中，我们可以选择使用元素自身的比较方式或是自定义比较器来进行元素的排序。若未指定比较器，则PriorityBlockingQueue将使用元素自身的比较方式进行排序。

private final Comparator comparator;

3.构造函数

PriorityBlockingQueue提供了多个构造函数，我们可以选择使用无参构造函数来创建一个默认容量为Integer.MAX_VALUE的PriorityBlockingQueue，或是使用带有初始容量或自定义比较器的构造函数。下面是PriorityBlockingQueue类的两个构造函数：

public PriorityBlockingQueue() {
    this(DEFAULT_INITIAL_CAPACITY, null);
}
public PriorityBlockingQueue(int initialCapacity, Comparator comparator) {
    if (initialCapacity < 1)
        throw new IllegalArgumentException();
    this.queue = new Object[initialCapacity];
    this.comparator = comparator;
}

4.添加元素

PriorityBlockingQueue中添加元素的方法为offer()方法，它会首先检查容量是否足够，如果容量不足则会进行扩容操作，扩容的方式是将原数组长度增加一半。接着，它会将新元素添加到队列的末尾，并通过siftUp()方法将元素上滤到合适的位置，以维护堆的性质。

public boolean offer(E e) {
    if (e == null)
        throw new NullPointerException();
    final ReentrantLock lock = this.lock;
    lock.lock();
    int n, cap;
    Object[] array;
    while ((n = size) >= (cap = (array = queue).length))
        tryGrow(array, cap);
    try {
        Comparator cmp = comparator; 
        if (n == 0) { array[0] = e; } 
        else { siftUp(n, e, array, cmp); } 
        size = n + 1; notEmpty.signal(); 
    } finally { 
        lock.unlock(); 
    } 
    return true; 
}

5.获取元素

PriorityBlockingQueue中获取元素的方法为take()方法，它会首先检查队列是否为空，如果队列为空则会将当前线程阻塞，直到有元素被添加到队列中。接着，它会获取队列的头部元素，并通过siftDown()方法将队列的末尾元素移动到头部，以维护堆的性质。

public E take() throws InterruptedException { 
    final ReentrantLock lock = this.lock; 
    lock.lockInterruptibly(); 
    E result; 
    try { 
        while (size == 0) notEmpty.await(); 
        result = extract(); 
    } finally {
        lock.unlock(); 
    } 
    return result; 
}
private E extract() { 
    final Object[] array = queue; 
    final E result = (E) array[0]; 
    final int n = --size; 
    final E x = (E) array[n]; 
    array[n] = null; 
    if (n != 0) 
    siftDown(0, x, array, comparator); 
    return result; 
}

6.维护堆性质

PriorityBlockingQueue使用小根堆或大根堆来维护元素的优先级，这里我们以小根堆为例。小根堆的特点是父节点的值小于等于左右子节点的值，PriorityBlockingQueue中的堆是通过数组来实现的。当添加元素时，会将新元素添加到队列的末尾，并通过siftUp()方法将元素上滤到合适的位置，以维护堆的性质。当获取元素时，会获取队列的头部元素，并通过siftDown()方法将队列的末尾元素移动到头部，以维护堆的性质。下面是siftUp()和siftDown()方法的代码实现：

private static  
void siftUp(int k, T x, Object[] array, Comparator cmp) { 
    if (cmp != null) 
    siftUpUsingComparator(k, x, array, cmp); 
    else siftUpComparable(k, x, array); 
}
@SuppressWarnings("unchecked") 
private static  
void siftUpUsingComparator(int k, T x, Object[] array, Comparator cmp) { 
    while (k > 0) { 
        int parent = (k - 1) >>> 1; 
        Object e = array[parent]; 
        if (cmp.compare(x, (T) e) >= 0) 
        break; 
        array[k] = e; 
        k = parent; 
    } 
    array[k] = x; 
}
@SuppressWarnings("unchecked") 
private static  
void siftUpComparable(int k, T x, Object[] array) { 
    Comparable key = (Comparable) x; 
    while (k > 0) { 
        int parent = (k - 1) >>> 1; 
        Object e = array[parent]; 
        if (key.compareTo((T) e) >= 0) 
        break; 
        array[k] = e; 
        k = parent; 
    } 
    array[k] = key; 
}
private static  
void siftDown(int k, T x, Object[] array, Comparator cmp) { 
    if (cmp != null) 
    siftDownUsingComparator(k, x, array, cmp); 
    else siftDownComparable(k, x, array); 
}
@SuppressWarnings("unchecked") 
private static  
void siftDownUsingComparator(int k, T x, Object[] array, Comparator cmp) { 
    int half = size >>> 1; 
    while (k < half) { 
        int child = (k << 1) + 1; 
        Object c = array[child]; 
        int right = child + 1; 
        if (right < size && cmp.compare((T) c, (T) array[right]) > 0) 
        c = array[child = right]; 
        if (cmp.compare(x, (T) c) <= 0) 
        break; 
        array[k] = c; 
        k = child; 
    } 
    array[k] = x; 
}
@SuppressWarnings("unchecked") 
private static  
void siftDownComparable(int k, T x, Object[] array) { 
    Comparable key = (Comparable) x; 
    int half = size >>> 1; 
    while (k < half) { 
        int child = (k << 1) + 1; 
        Object c = array[child]; 
        int right = child + 1; 
        if (right < size && ((Comparable) c).compareTo((T) array[right]) > 0) 
        c = array[child = right]; 
        if (key.compareTo((T) c) <= 0) 
        break; 
        array[k] = c; 
        k = child; 
    } 
    array[k] = key; 
}

siftUp()方法和siftDown()方法都使用了siftUpUsingComparator()方法和siftDownUsingComparator()方法，它们是使用Comparator来实现堆的上滤和下滤的。当PriorityBlockingQueue没有指定Comparator时，会使用元素自身的Comparable来实现堆的上滤和下滤。