PriorityBlockingQueue是Java并发包java.util.concurrent下的一个线程安全阻塞队列，它支持按照元素的优先级进行排序，内部基于二叉堆结构实现，默认情况下元素按照自然顺序升序排列，也可以通过自定义比较器指定排序规则。它不允许插入null元素，并且是 unbounded 队列，理论上可以一直添加元素直到内存耗尽。
PriorityBlockingQueue核心特性
- 线程安全：所有入队出队操作都通过锁机制保证线程安全，适合多线程并发场景
- 优先级排序：元素会按照指定规则排序，队头元素永远是优先级最高（或最低，取决于排序规则）的元素
- 阻塞特性：当队列为空时，获取元素的操作会阻塞等待，直到队列中有新元素加入
- 无界队列：没有固定容量限制，插入操作不会因为队列满而阻塞
基本操作方法
构造方法
PriorityBlockingQueue提供了多个构造方法，常用的有以下几种：
// 默认构造方法，初始容量为11，按照元素自然顺序排序 PriorityBlockingQueue<Integer> queue1 = new PriorityBlockingQueue<>(); // 指定初始容量的构造方法 PriorityBlockingQueue<Integer> queue2 = new PriorityBlockingQueue<>(20); // 指定初始容量和自定义比较器的构造方法 PriorityBlockingQueue<Integer> queue3 = new PriorityBlockingQueue<>(20, (a, b) -> b - a);
常用API说明
| 方法名 | 作用 | 特性说明 |
|---|---|---|
| add(E e) | 添加元素到队列 | 成功返回true，插入null会抛NullPointerException |
| offer(E e) | 添加元素到队列 | 成功返回true，插入null会抛NullPointerException |
| put(E e) | 添加元素到队列 | 由于是无界队列，该方法不会阻塞，效果和add一致 |
| peek() | 获取队头元素但不移除 | 队列为空返回null |
| poll() | 获取并移除队头元素 | 队列为空返回null |
| take() | 获取并移除队头元素 | 队列为空时会阻塞等待，直到有元素可用 |
| size() | 获取队列元素数量 | 由于遍历过程需要加锁，该方法开销比普通集合大 |
自定义元素优先级规则
如果要存储自定义类型的元素，需要让元素实现Comparable接口重写compareTo方法，或者在构造PriorityBlockingQueue时传入自定义Comparator比较器，否则会抛出ClassCastException。
方式一：元素实现Comparable接口
以任务类为例，假设任务有优先级字段，优先级数值越小优先级越高：
import java.util.concurrent.PriorityBlockingQueue;
// 自定义任务类，实现Comparable接口
class Task implements Comparable<Task> {
private String name;
private int priority; // 优先级，数值越小优先级越高
public Task(String name, int priority) {
this.name = name;
this.priority = priority;
}
// 重写compareTo方法，按照优先级升序排列
@Override
public int compareTo(Task o) {
return Integer.compare(this.priority, o.priority);
}
@Override
public String toString() {
return "Task{name='" + name + "', priority=" + priority + "}";
}
}
public class PriorityBlockingQueueDemo {
public static void main(String[] args) {
// 创建优先级阻塞队列
PriorityBlockingQueue<Task> queue = new PriorityBlockingQueue<>();
// 添加不同优先级的任务
queue.add(new Task("低优先级任务", 3));
queue.add(new Task("高优先级任务", 1));
queue.add(new Task("中优先级任务", 2));
// 依次取出任务，验证优先级排序
while (!queue.isEmpty()) {
System.out.println(queue.poll());
}
}
}
方式二：传入自定义Comparator比较器
如果不想修改自定义类的代码，可以在构造队列时传入比较器：
import java.util.Comparator;
import java.util.concurrent.PriorityBlockingQueue;
class Task {
private String name;
private int priority;
public Task(String name, int priority) {
this.name = name;
this.priority = priority;
}
public int getPriority() {
return priority;
}
@Override
public String toString() {
return "Task{name='" + name + "', priority=" + priority + "}";
}
}
public class PriorityBlockingQueueDemo2 {
public static void main(String[] args) {
// 传入自定义比较器，按照优先级升序排列
PriorityBlockingQueue<Task> queue = new PriorityBlockingQueue<>(10, Comparator.comparingInt(Task::getPriority));
queue.add(new Task("低优先级任务", 3));
queue.add(new Task("高优先级任务", 1));
queue.add(new Task("中优先级任务", 2));
while (!queue.isEmpty()) {
System.out.println(queue.poll());
}
}
}
多线程场景使用示例
PriorityBlockingQueue的阻塞特性非常适合生产者消费者模型，下面的示例模拟多个生产者提交不同优先级的任务，单个消费者按照优先级处理任务：
import java.util.concurrent.PriorityBlockingQueue;
class Task implements Comparable<Task> {
private String name;
private int priority;
public Task(String name, int priority) {
this.name = name;
this.priority = priority;
}
@Override
public int compareTo(Task o) {
return Integer.compare(this.priority, o.priority);
}
@Override
public String toString() {
return "Task{name='" + name + "', priority=" + priority + "}";
}
}
public class MultiThreadDemo {
public static void main(String[] args) {
PriorityBlockingQueue<Task> queue = new PriorityBlockingQueue<>();
// 启动3个生产者线程，提交不同优先级的任务
for (int i = 0; i < 3; i++) {
int producerId = i;
new Thread(() -> {
for (int j = 0; j < 3; j++) {
int priority = (int) (Math.random() * 5) + 1;
Task task = new Task("生产者" + producerId + "的任务" + j, priority);
queue.put(task);
System.out.println("生产者" + producerId + "提交了任务：" + task);
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}).start();
}
// 启动1个消费者线程，处理任务
new Thread(() -> {
while (true) {
try {
// take方法会阻塞，直到队列中有任务
Task task = queue.take();
System.out.println("消费者处理了任务：" + task);
Thread.sleep(200);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
}
}).start();
}
}
注意事项
- PriorityBlockingQueue的迭代器不保证按照优先级顺序遍历元素，如果需要按顺序遍历，可以先取出所有元素再排序
- 自定义比较规则时要保证比较逻辑的一致性，避免出现比较结果矛盾的情况，否则可能导致堆结构异常
- 虽然是无界队列，但插入元素时如果内存不足还是会抛出OutOfMemoryError
- 批量添加元素时，如果使用addAll方法，会逐个添加元素并调整堆结构，效率比逐个添加高，但依然要保证所有元素都符合排序规则
PriorityBlockingQueueJava优先级队列线程安全队列compareTo方法阻塞队列修改时间：2026-07-17 21:30:22