Java 中并发编程怎么使用 PriorityBlockingQueue 实现优先级队列-java教程-PHP中文网

priorityblockingqueue是线程安全的无界优先级队列，基于堆实现，通过reentrantlock保证原子性并支持阻塞操作；而priorityqueue非线程安全，多线程下易导致数据异常。

java 中并发编程怎么使用 priorityblockingqueue 实现优先级队列

PriorityBlockingQueue 是 Java 并发包（java.util.concurrent）中提供的一个线程安全的无界优先级队列，底层基于堆实现，支持按自然顺序或自定义比较器排序。它适合在多线程环境下实现带优先级的任务调度、事件分发等场景。

为什么用 PriorityBlockingQueue 而不是普通 PriorityQueue？

普通 PriorityQueue 不是线程安全的，多线程并发读写会引发数据不一致甚至抛出异常；而 PriorityBlockingQueue 内部通过显式锁（ReentrantLock）保证所有操作原子性，且提供阻塞式插入/获取能力（如 take() 会等待非空），天然适配生产者-消费者模型。

基本用法：定义元素并指定优先级

要让元素按优先级排序，需满足以下任一条件：

元素类实现 Comparable 接口，重写 compareTo() 方法（按自然顺序）
创建队列时传入 Comparator 实例（更灵活，推荐）

例如定义一个任务类：

Alibabacloud Sdk Client Initialization For Java

在 Java 中初始化和管理阿里云 SDK客户端。包括单例模式、线程安全、endpoint 与 region 配置、VPC 终端节点、同步与异步等。

下载

public class Task implements Comparable<task> {
    private final String name;
    private final 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 String.format("Task{name='%s', priority=%d}", name, priority);
    }
}</task>

然后初始化队列并使用：

PriorityBlockingQueue<task> queue = new PriorityBlockingQueue();
queue.offer(new Task("low", 10));
queue.offer(new Task("high", 1));
queue.offer(new Task("medium", 5));

System.out.println(queue.poll()); // Task{name='high', priority=1}
System.out.println(queue.poll()); // Task{name='medium', priority=5}
</task>

在并发环境中的典型应用模式

常见于“多生产者 + 单消费者”或“多线程协作调度”场景。关键点：

生产者调用 offer() 或 put()（不会阻塞，因队列无界）
消费者常用 take() 阻塞等待最高优先级任务，避免忙等
注意：队列不保证相同优先级元素的处理顺序（无 FIFO 保证），如需稳定顺序，可在 compareTo() 中加入时间戳或序列号作为第二排序字段

示例消费者线程：

Thread consumer = new Thread(() -> {
    try {
        while (!Thread.currentThread().isInterrupted()) {
            Task task = queue.take(); // 阻塞直到有任务
            System.out.println("Processing: " + task);
            // 执行任务逻辑...
            Thread.sleep(100);
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
});
consumer.start();