上一篇文章剖析了 SynchronousQueue 的相关源码，那这篇文章接着看另外一个特殊的阻塞队列 —— PriorityBlockingQueue
简介
PriorityBlockingQueue 从字面意思可以知道是有优先级的阻塞队列。无独有偶，与 PriorityBlockingQueue 类相似的还有 PriorityQueue 类。实际上，其内部就是复用了 PriorityQueue 类，加上 CAS 锁，实现了阻塞的接口形成了与 PriorityQueue 规则一样、线程安全的阻塞队列。
底层是用数组存储元素，通过二叉堆数据结构对元素排序。
ps.不清楚的 PriorityQueue 类的源码，可参考我另外一篇拙作：Java 队列之 PriorityQueue 源码分析
public static void main(String[] args) {
// 第一种：使用 PriorityQueue 默认的比较器，对象排序是升序
// PriorityBlockingQueue<Integer> priorityBlockingQueue = new PriorityBlockingQueue();
// 第二种：自定义比较器
// PriorityBlockingQueue<Integer> priorityBlockingQueue = new PriorityBlockingQueue(20,new Comparator<Integer>() {
// @Override
// public int compare(Integer o1, Integer o2) {
// // 降序
// return o2-o1;
// }
// });
// 第三种：其实和第二种方式一样，只不过使用Lambda表达式更加优雅
// 如果只是想实现降序或者升序，可以使用Comparator.reverseOrder()或Comparator.naturalOrder()
PriorityBlockingQueue<Integer> priorityBlockingQueue = new PriorityBlockingQueue(20, Comparator.naturalOrder());
priorityBlockingQueue.add(6);
priorityBlockingQueue.add(10);
priorityBlockingQueue.add(2);
priorityBlockingQueue.add(8);
priorityBlockingQueue.add(9);
priorityBlockingQueue.add(1);
priorityBlockingQueue.add(3);
while(!priorityBlockingQueue.isEmpty()){
System.out.println(priorityBlockingQueue.poll());
}
}
输出结果：
1
2
3
6
8
9
10
以上的使用示例，除了创建 PriorityBlockingQueue 对象和创建 PriorityQueue 对象有差别，其它基本一样。实际上，PriorityBlockingQueue 类很多方法的实现与 PriorityQueue 类方法的实现基本一模一样，只不过在查找、删除、添加元素的时候会使用到 ReentrantLock 锁保证线程安全。
接下来，咱们就可以剖析源码，由于大部分源码实现与 PriorityQueue 相似，因此这篇文章我只对不同点重点说明，其它的就只添加注释。
类注释
- 和 PriorityQueue 类的规则一样，PriorityBlockingQueue 的队列是无界限的，直到发生OOM就会无法执行添加操作
- 对象如果没有实现 Comparable 接口，则必须实现 Comparator 接口，否则会抛异常
- 实现了 Collection 和 Iterator 接口，但无法使用 iterator 方法迭代队列的元素。如果需要迭代可以考虑使用
Arrays.sort方法 - 队列内的元素不能是null
成员变量
// 默认初始化大小
private static final int DEFAULT_INITIAL_CAPACITY = 11;
// 数组最大容量
private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
// 用数组实现的二叉堆
// 父节点的位置是n，那么左孩子的位置是2*n+1，右孩子的位置是2*（n+1）
private transient Object[] queue;
// 队列的元素数量
private transient int size;
// 比较器
private transient Comparator<? super E> comparator;
// 可重入锁
private final ReentrantLock lock;
//等待获取元素的条件
private final Condition notEmpty;
// 自旋锁
private transient volatile int allocationSpinLock;
// 优先队列：主要用于序列化，这是为了兼容之前的版本。只有在序列化和反序列化才非空
private PriorityQueue<E> q;
构造函数
// 使用默认队列容量
public PriorityBlockingQueue() {
this(DEFAULT_INITIAL_CAPACITY, null);
}
// 指定队列元素容量
public PriorityBlockingQueue(int initialCapacity) {
this(initialCapacity, null);
}
// 指定队列元素容量和比较器
public PriorityBlockingQueue(int initialCapacity, Comparator<? super E> comparator) {
if (initialCapacity < 1)
throw new IllegalArgumentException();
// 初始化锁
this.lock = new ReentrantLock();
this.notEmpty = lock.newCondition();
this.comparator = comparator;
this.queue = new Object[initialCapacity];
}
// 根据Collection集合初始化优先阻塞队列
public PriorityBlockingQueue(Collection<? extends E> c) {
this.lock = new ReentrantLock();
this.notEmpty = lock.newCondition();
boolean heapify = true; // true if not known to be in heap order
boolean screen = true; // true if must screen for nulls
if (c instanceof SortedSet<?>) {
SortedSet<? extends E> ss = (SortedSet<? extends E>) c;
this.comparator = (Comparator<? super E>) ss.comparator();
heapify = false;
} else if (c instanceof PriorityBlockingQueue<?>) {
PriorityBlockingQueue<? extends E> pq =
(PriorityBlockingQueue<? extends E>) c;
this.comparator = (Comparator<? super E>) pq.comparator();
screen = false;
if (pq.getClass() == PriorityBlockingQueue.class) // exact match
heapify = false;
}
Object[] a = c.toArray();
int n = a.length;
if (a.getClass() != Object[].class)
a = Arrays.copyOf(a, n, Object[].class);
if (screen && (n == 1 || this.comparator != null)) {
for (int i = 0; i < n; ++i)
if (a[i] == null)
throw new NullPointerException();
}
this.queue = a;
this.size = n;
if (heapify)
// 将集合转化为符合二叉堆的数组
heapify();
}
添加元素
// 添加数组元素
public boolean add(E e) {
return offer(e);
}
public boolean offer(E e) {
// 添加的元素不能为null
if (e == null)
throw new NullPointerException();
final ReentrantLock lock = this.lock;
// 上锁
lock.lock();
int n, cap;
Object[] array;
// 如果size大于等于了数组的长度，则进行扩容操作
while ((n = size) >= (cap = (array = queue).length))
tryGrow(array, cap);
try {
Comparator<? super E> cmp = comparator;
if (cmp == null)
siftUpComparable(n, e, array);
else
siftUpUsingComparator(n, e, array, cmp);
// 队列的元素数量加1
size = n + 1;
// 唤醒等待线程
notEmpty.signal();
} finally {
// 释放锁
lock.unlock();
}
return true;
}
private static <T> void siftUpComparable(int k, T x, Object[] array) {
Comparable<? super T> key = (Comparable<? super T>) 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 <T> void siftUpUsingComparator(int k, T x, Object[] array, Comparator<? super T> 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;
}
查找元素
public E poll() {
final ReentrantLock lock = this.lock;
// 加锁
lock.lock();
try {
return dequeue();
} finally {
lock.unlock();
}
}
// 取出元素的时候，加上超时时间。如果队列没元素而且未达到超时时间会一直阻塞
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
E result;
try {
// 队列为空而且未到超时时间则一直阻塞
while ((result = dequeue()) == null && nanos > 0)
nanos = notEmpty.awaitNanos(nanos);
} finally {
// 释放锁
lock.unlock();
}
return result;
}
// 读取二叉堆根元素，如果二叉堆为空则等待阻塞
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
E result;
try {
// 如果队列为空则阻塞
while ((result = dequeue()) == null)
notEmpty.await();
} finally {
lock.unlock();
}
return result;
}
// 出队
private E dequeue() {
int n = size - 1;
if (n < 0)
return null;
else {
Object[] array = queue;
E result = (E) array[0];
E x = (E) array[n];
array[n] = null;
Comparator<? super E> cmp = comparator;
if (cmp == null)
siftDownComparable(0, x, array, n);
else
siftDownUsingComparator(0, x, array, n, cmp);
size = n;
return result;
}
}
// 向array数组中位置为k的地方插入元素x
private static <T> void siftDownComparable(int k, T x, Object[] array, int n) {
if (n > 0) {
Comparable<? super T> key = (Comparable<? super T>) x;
// 保证循环到最下面的非叶子节点
int half = n >>> 1;
while (k < half) {
// 取k节点的左子节点位置
int child = (k << 1) + 1;
// 取k节点的左子节点元素c
Object c = array[child];
//取k节点的右子节点位置
int right = child + 1;
// right < size说明k节点是有两个节点，则比较左右节点的大小
// 当左节点大于右节点时，将右节点的值赋给左节点c，即找出k的最小子节点
if (right < n && ((Comparable<? super T>) 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;
}
}
// 与 siftDownComparable 方法流程相似
private static <T> void siftDownUsingComparator(int k, T x, Object[] array, int n, Comparator<? super T> cmp) {
if (n > 0) {
int half = n >>> 1;
while (k < half) {
int child = (k << 1) + 1;
Object c = array[child];
int right = child + 1;
if (right < n && 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;
}
}
删除元素
// 实际上调用poll方法将元素移除
public E remove() {
E x = poll();
if (x != null)
return x;
else
throw new NoSuchElementException();
}
// 移除指定元素
public boolean remove(Object o) {
final ReentrantLock lock = this.lock;
// 加锁
lock.lock();
try {
// 根据元素查找数组对应的下标
int i = indexOf(o);
if (i == -1)
return false;
// 删除元素
removeAt(i);
return true;
} finally {
// 释放锁
lock.unlock();
}
}
// 移除指定位置的元素
private void removeAt(int i) {
Object[] array = queue;
int n = size - 1;
// 如果要删除的节点在队列尾部，直接将队列尾部的元素设置为null
if (n == i) // removed last element
array[i] = null;
else {
// 取出最后一个元素，然后将其位置的元素设置为null
E moved = (E) array[n];
array[n] = null;
Comparator<? super E> cmp = comparator;
if (cmp == null)
siftDownComparable(i, moved, array, n);
else
siftDownUsingComparator(i, moved, array, n, cmp);
if (array[i] == moved) {
if (cmp == null)
siftUpComparable(i, moved, array);
else
siftUpUsingComparator(i, moved, array, cmp);
}
}
size = n;
}
扩容
private void tryGrow(Object[] array, int oldCap) {
lock.unlock(); // must release and then re-acquire main lock
Object[] newArray = null;
if (allocationSpinLock == 0 && UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset, 0, 1)) {
try {
// 如果长度小于64，则添加到原来的2倍，如果大于64，则增加为原来的一半
int newCap = oldCap + ((oldCap < 64) ? (oldCap + 2) : (oldCap >> 1));
if (newCap - MAX_ARRAY_SIZE > 0) { // possible overflow
int minCap = oldCap + 1;
if (minCap < 0 || minCap > MAX_ARRAY_SIZE)
throw new OutOfMemoryError();
newCap = MAX_ARRAY_SIZE;
}
if (newCap > oldCap && queue == array)
// 创建扩容后的数组
newArray = new Object[newCap];
} finally {
allocationSpinLock = 0;
}
}
// 如果newArray为null则代表有其它线程正在扩容，调用yield方法放弃CPU资源
if (newArray == null)
Thread.yield();
lock.lock();
// 数组复制
if (newArray != null && queue == array) {
queue = newArray;
System.arraycopy(array, 0, newArray, 0, oldCap);
}
}
总结：
PriorityBlockingQueue 方法的实现与 PriorityQueue 方法实现十分相似，只不过是在添加、查找、删除元素的时候使用ReentrantLock锁保证线程安全。但与 PriorityQueue 不同的是，PriorityBlockingQueue 实现了 BlockingQueue 接口，所以具有阻塞队列的方法，如take，加上超时的poll方法。值得注意的是，PriorityBlockingQueue 队列并不是有序的，只是保证内部二叉堆头元素是堆内最小值，每次取的时候只取堆头元素，如果没看过源码的话很容易误以为是有序的。
如果觉得源码剖析不错的话，麻烦点个赞哈！对于文章有哪里不清楚或者有误的地方，欢迎在评论区留言~
网友评论