PriorityBlockingQueue 的优先级能力不是来自链表扫描，而是数组维护的二叉最小堆。每次入队会执行 siftUp 向上调整，出队则把堆顶取走后，将数组末尾元素挪到根节点并执行 siftDown 向下调整，因此插入和删除都保持 O(log n) 的时间复杂度。这个特性在多路归并排序里非常关键：假设有 100 个已经有序的分片，只需要在堆里维护 100 个候选元素，每次弹出一个最小值，再补入对应分片的下一个元素，就能输出全局有序序列，而不需要把全部数据加载进内存。

与 PriorityQueue 不同，PriorityBlockingQueue 通过可重入锁 ReentrantLock 和条件队列实现线程安全。take 操作在队列为空时会等待，直到生产者放入元素并发出信号；offer 和 put 因为队列无界，理论上不会阻塞，但内部仍然加锁。这个语义让它在生产者消费者模型中非常顺手，尤其是多个数据源并行写、多个消费者负责归并处理的场景。

一、底层特性与阻塞语义

PriorityBlockingQueue 的默认初始容量是 11，当元素数量超过容量时会触发扩容。如果旧容量小于 64，会翻倍增长；大于等于 64 时则每次增加旧容量的一半。这种扩容策略在短时间涌入大量元素时仍然会带来内存压力和数组复制开销，所以我们在生产环境中通常会配合有界控制来使用，而不是直接依赖它的无界特性。

阻塞语义主要体现在 take 方法上。当队列为空时，调用 take 的线程会进入条件队列等待，并释放锁，直到其他线程调用 put 或 offer 放入元素并触发 notEmpty 信号。put 方法在 PriorityBlockingQueue 中并不会因为容量满而阻塞，因为队列本身是无界的，它只是在内部执行一次加锁后的入队操作。这种设计适合优先级任务调度，但用于大数据排序时必须额外处理生产速度，否则内存可能被打满。

下面这段代码展示了自定义比较器下的最大堆行为。虽然 PriorityBlockingQueue 通常表现为最小堆，但比较器反转后可以变成最大堆，适合需要先处理高分任务或高优先级日志的场景。

import java.util.concurrent.PriorityBlockingQueue;
public class QueueBasicDemo {
    public static void main(String[] args) throws InterruptedException {
        PriorityBlockingQueue<Integer> queue = new PriorityBlockingQueue<>(11, (a, b) -> b - a);
        queue.put(3);
        queue.put(1);
        queue.put(2);
        System.out.println(queue.take()); // 输出 3，自定义比较器实现最大堆
        System.out.println(queue.take()); // 输出 2
        System.out.println(queue.take()); // 输出 1
    }
}

二、大数据排序实战：从分片到多路归并

大数据排序的经典思路是外部排序：先将待排序数据切分成若干个能放入内存的块，逐块排序后写回磁盘形成有序分片，再对所有分片进行多路归并。PriorityBlockingQueue 在多路归并阶段非常适合充当最小堆，堆中只保存每个分片当前的候选元素，每次弹出全局最小元素，并从该元素所属分片补充下一个元素。

归并过程中，堆的大小始终等于分片数量，而不是总数据量。即使分片数量达到几百个，堆操作的开销也只是 log(分片数) 级别。相比把所有分片数据一次性读入内存后排序，这种方式可以稳定控制内存占用，并且能够随时中断或继续。下面是一个模拟有序分片归并的完整示例。

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.PriorityBlockingQueue;
public class ExternalMergeSorter {
    public static List<String> mergeSortedShards(List<List<String>> shards) throws InterruptedException {
        PriorityBlockingQueue<ShardHead> heap = new PriorityBlockingQueue<>(
                11, Comparator.comparing(ShardHead::getValue));
        int[] indexes = new int[shards.size()];
        for (int i = 0; i < shards.size(); i++) {
            if (indexes[i] < shards.get(i).size()) {
                heap.put(new ShardHead(i, shards.get(i).get(0)));
            }
        }
        List<String> result = new ArrayList<>();
        while (!heap.isEmpty()) {
            ShardHead head = heap.take();
            int shardId = head.getShardId();
            result.add(head.getValue());
            indexes[shardId]++;
            if (indexes[shardId] < shards.get(shardId).size()) {
                heap.put(new ShardHead(shardId, shards.get(shardId).get(indexes[shardId])));
            }
        }
        return result;
    }
    private static class ShardHead {
        private final int shardId;
        private final String value;
        ShardHead(int shardId, String value) {
            this.shardId = shardId;
            this.value = value;
        }
        int getShardId() { return shardId; }
        String getValue() { return value; }
    }
}

在真实的大数据排序任务中，shards 可以来自磁盘上的有序小文件，读取逻辑封装成分片迭代器。生产者线程负责从各分片读取头部数据并放入 PriorityBlockingQueue，消费者线程调用 take 取出当前最小值写入最终结果文件。由于 take 会阻塞，当所有分片都暂时没有数据时，消费者不会空转，这是它相比普通 PriorityQueue 加轮询方案更优雅的地方。

三、生产消费调优：有界化、批量消费与锁策略

无界是 PriorityBlockingQueue 最大的使用风险。排序任务如果因为上游数据倾斜或消费者处理缓慢，生产者又持续写入，队列会不断扩容，最终可能触发 OutOfMemoryError。实际项目中通常会封装一个带许可证的有界优先阻塞队列，用 Semaphore 控制最大元素数量。put 操作先获取许可，若队列满则阻塞；take 操作释放许可，让生产者继续放入。

import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.Semaphore;
public class BoundedPriorityBlockingQueue<E> {
    private final PriorityBlockingQueue<E> queue;
    private final Semaphore permits;
    public BoundedPriorityBlockingQueue(int capacity) {
        this.queue = new PriorityBlockingQueue<>();
        this.permits = new Semaphore(capacity);
    }
    public void put(E element) throws InterruptedException {
        permits.acquire();
        queue.put(element);
    }
    public E take() throws InterruptedException {
        E element = queue.take();
        permits.release();
        return element;
    }
    public int size() {
        return queue.size();
    }
}

批量消费可以使用 drainTo 方法。它在内部循环调用 poll，因此会按照优先级顺序移除元素，但通常一次性移走当前所有元素，不像 take 那样阻塞等待新数据。如果消费端还需要稳定处理时间，建议在 drainTo 之后判断返回数量，再决定是否继续等待下一轮。相比于反复调用 poll，drainTo 减少了加锁次数，适合积压数据较多时快速清理。

PriorityBlockingQueue 内部使用 ReentrantLock，但构造函数没有暴露公平性配置，默认采用非公平锁。非公平锁吞吐量更高，但在高并发、多消费者场景下，某些线程可能长时间拿不到锁。对于延迟敏感的任务调度系统，如果公平性要求很高，可能需要在外层引入额外的调度机制，或者评估其他并发队列实现。优先级排序场景通常更看重吞吐，默认非公平锁在多数情况下是可以接受的。

四、常见误区与适用边界

第一个误区是认为 PriorityBlockingQueue 能直接完成大数据排序。实际上它只负责归并阶段，前置的分片排序不能省略。如果原始数据完全无序，直接写入 PriorityBlockingQueue 并不会避免内存压力，堆中仍然需要保存全量数据。正确做法是先分块读入、块内排序、落盘，再进入多路归并。

第二个误区是以为无界队列就是安全队列。无界只代表 put 不会阻塞，不代表内存不会耗尽。在互联网长尾流量、日志暴涨或下游消费停滞时，无界优先队列会迅速积累元素，最终拖垮整个 JVM。有界包装和监控队列深度是必须配套的措施。

PriorityBlockingQueue 适合单机多线程环境下的优先级任务调度、广告竞价排序、日志聚合和多路归并。对于超高吞吐、分布式排序场景，它受限于单机锁竞争和内存容量，更适合作为局部组件，而不是全局排序方案。理解它的边界，比掌握它的 API 更重要。