详解Java并发编程中的优先级队列PriorityBlockingQueue
PriorityBlockingQueue是Java中实现了堆数据结构的线程安全的有界阻塞队列。它可以在多线程场景下安全地进行元素添加、删除和获取操作，而且可以根据元素的优先级进行排序。本篇博客将会深入解读PriorityBlockingQueue的源码实现。
一、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<? super E> comparator;
3.构造函数
PriorityBlockingQueue提供了多个构造函数，我们可以选择使用无参构造函数来创建一个默认容量为Integer.MAX_VALUE的PriorityBlockingQueue，或是使用带有初始容量或自定义比较器的构造函数。下面是PriorityBlockingQueue类的两个构造函数：
public PriorityBlockingQueue() {
this(DEFAULT_INITIAL_CAPACITY, null);
}
public PriorityBlockingQueue(int initialCapacity, Comparator<? super E> 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<? super E> 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 <T>
void siftUp(int k, T x, Object[] array, Comparator<? super T> cmp) {
if (cmp != null)
siftUpUsingComparator(k, x, array, cmp);
else siftUpComparable(k, x, array);
}
@SuppressWarnings("unchecked")
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;
}
@SuppressWarnings("unchecked")
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 siftDown(int k, T x, Object[] array, Comparator<? super T> cmp) {
if (cmp != null)
siftDownUsingComparator(k, x, array, cmp);
else siftDownComparable(k, x, array);
}
@SuppressWarnings("unchecked")
private static <T>
void siftDownUsingComparator(int k, T x, Object[] array, Comparator<? super T> 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 <T>
void siftDownComparable(int k, T x, Object[] array) {
Comparable<? super T> key = (Comparable<? super T>) 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<? 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;
}siftUp()方法和siftDown()方法都使用了siftUpUsingComparator()方法和siftDownUsingComparator()方法，它们是使用Comparator来实现堆的上滤和下滤的。当PriorityBlockingQueue没有指定Comparator时，会使用元素自身的Comparable来实现堆的上滤和下滤。
总结
PriorityBlockingQueue是Java并发包中的一个线程安全的、支持优先级队列的类，它使用小根堆或大根堆来维护元素的优先级。PriorityBlockingQueue的内部实现使用了ReentrantLock和Condition来实现线程安全的操作，同时使用了数组来实现堆。PriorityBlockingQueue的核心方法包括添加元素的offer()方法，获取元素的take()方法，以及维护堆性质的siftUp()方法和siftDown()方法。PriorityBlockingQueue的使用方式与普通队列类似，但可以根据元素的优先级进行排序和处理。
以上就是详解Java并发编程中的优先级队列PriorityBlockingQueue的详细内容，更多关于Java PriorityBlockingQueue的资料请关注脚本之家其它相关文章！
相关文章
这篇文章主要介绍了SpringBoot Redis批量存取数据的操作，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教2021-08-08
这篇文章主要介绍了SpringBoot如何实现接口版本控制，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教2021-10-10
这篇文章主要介绍了Java泛型机制必要性及原理解析,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下2020-05-05
这篇文章主要介绍了Dubbo服务如何优雅的校验参数,Dubbo框架本身是支持参数校验的，同时也是基于JSR303去实现的，今天通过示例代码介绍下详细实现过程，需要的朋友可以参考下2022-03-03
这篇文章主要为大家分享了最有价值的6道线程池面试题，涵盖内容全面，包括数据结构和算法相关的题目、经典面试编程题等，对hashCode方法的设计、垃圾收集的堆和代进行剖析，感兴趣的小伙伴们可以参考一下2021-06-06
这篇文章主要给大家介绍了关于JavaGUI界面实现页面跳转的相关资料, GUI是指图形用户界面,指采用图形方式显示的计算机操作用户界面,需要的朋友可以参考下2023-07-07
本篇文章主要介绍了数据库的一些常用方法及一些基本操作，需要的朋友可以参考下2017-04-04
Spring Boot 中嵌入式 Servlet 容器自动配置原理解析
这篇文章主要介绍了Spring Boot 中嵌入式 Servlet 容器自动配置原理解析，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教2021-11-11
Spring是一款广泛应用于企业级应用程序开发的Java框架，其 IOC 和 DI 特性可以有效地管理应用程序中的对象，提高了应用程序的可维护性和可扩展性，那你知道spring有哪些方式将bean放入容器嘛，今天就给大家总结一下2023-07-07
文章介绍了如何在Navicat中创建存储过程,并在Spring Boot项目中调用存储过程,存储过程创建步骤包括选择函数类型、自定义函数名、添加参数等,在Spring Boot中调用存储过程时,可以通过JdbcTemplate或MyBatis等工具进行2024-11-11
最新评论