PriorityBlockingQueue 是线程安全的无界优先队列，基于堆实现，支持自然序或自定义排序，不保证公平性，适合优先级任务分发；需封装调度器、复合排序防饿死、监控背压防OOM。
PriorityBlockingQueue 是 Java 并发包（java.util.concurrent）中一个线程安全的无界优先队列，底层基于堆实现，支持按自然顺序或自定义 Comparator 排序。它本身**不保证公平性**，也不支持阻塞式插入（如满时等待），但提供阻塞式获取（take()），非常适合构建“生产者-消费者”模式下的**优先级任务分发模型**。
一、核心设计原则：任务建模 + 线程安全调度
高并发下优先级任务分发的关键不是单纯用队列，而是围绕三个要素组织：
-
可比较的任务对象：任务类需实现
Comparable或配合外部Comparator，确保优先级逻辑明确（例如数值越小优先级越高，或用枚举定义等级）； - 单点调度器封装：避免多个线程直接操作队列，用一个调度器类统一管理入队、出队、拒绝策略和监控；
-
消费者线程池 + 阻塞拉取：使用固定线程池持续调用
queue.take()，天然支持空闲等待与唤醒，无需轮询或 sleep。
二、任务定义示例：支持动态优先级与时间兜底
真实场景中，仅靠静态优先级易导致低优任务饿死。建议在任务中融合“业务优先级 + 提交时间”做复合排序：
public class PriorityTask implements Comparable<PriorityTask> {
private final int businessPriority; // 1=最高，10=最低
private final long submitTime; // 提交时间戳，用于防止低优任务永久等待
private final Runnable action;
public PriorityTask(int businessPriority, Runnable action) {
this.businessPriority = Math.max(1, Math.min(10, businessPriority));
this.submitTime = System.nanoTime();
this.action = action;
}
@Override
public int compareTo(PriorityTask o) {
int cmp = Integer.compare(this.businessPriority, o.businessPriority);
if (cmp != 0) return cmp;
// 同优先级时，先提交的先执行（FIFO）
return Long.compare(this.submitTime, o.submitTime);
}
public void execute() { action.run(); }
}
三、调度器封装：线程安全 + 可观测 + 可控拒绝
直接暴露 PriorityBlockingQueue 容易引发并发误用。推荐封装成带监控和拒绝策略的调度器：
立即学习“Java免费学习笔记（深入）”；
public class PriorityTaskDispatcher {
private final PriorityBlockingQueue<PriorityTask> queue;
private final ExecutorService workerPool;
private final RejectedExecutionHandler rejectionHandler;
public PriorityTaskDispatcher(int corePoolSize, RejectedExecutionHandler handler) {
this.queue = new PriorityBlockingQueue<>();
this.workerPool = new ThreadPoolExecutor(
corePoolSize, corePoolSize,
0L, TimeUnit.MILLISECONDS,
new SynchronousQueue<>(),
r -> new Thread(r, "priority-task-worker-" + r.hashCode()),
handler
);
this.rejectionHandler = handler != null ? handler : (r, executor) -> {
System.err.println("Task rejected: " + r);
};
// 启动常驻消费者线程（数量 = corePoolSize）
for (int i = 0; i < corePoolSize; i++) {
workerPool.submit(() -> {
try {
while (!Thread.currentThread().isInterrupted()) {
PriorityTask task = queue.take(); // 阻塞直到有任务
task.execute();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
}
}
public boolean submit(PriorityTask task) {
if (task == null) return false;
// offer() 总成功（无界队列），无需处理失败
return queue.offer(task);
}
public long getQueueSize() { return queue.size(); }
public void shutdown() {
workerPool.shutdown();
try {
if (!workerPool.awaitTermination(10, TimeUnit.SECONDS)) {
workerPool.shutdownNow();
}
} catch (InterruptedException e) {
workerPool.shutdownNow();
Thread.currentThread().interrupt();
}
}
}
四、关键注意事项与避坑指南
PriorityBlockingQueue 在高并发优先级调度中容易被低估的细节：
-
非公平性不可忽略：多个线程同时
take()，JVM 不保证谁先拿到下一个任务——若需严格 FIFO+优先级，应只用单个消费者线程，或改用DelayQueue+ 时间戳模拟； - 修改正在排队的任务属性无效：队列排序在入队时确定，后续修改任务字段（如 priority）不会触发重排序，必须重新入队；
-
无界 ≠ 无限内存：虽不抛
IllegalStateException，但积压过多任务会 OOM，务必配合监控（如定期打印queue.size()）和背压策略（如自定义RejectedExecutionHandler降级或限流）； -
Comparator 必须满足一致性：若
a.equals(b)为 true，compare(a,b)必须返回 0；否则可能导致重复任务或排序异常。
