当前位置：首页 > news >正文

Java并发包中的PriorityBlockingQueue解析

news 2026/5/12 17:21:10

PriorityBlockingQueue<E>是 Java 并发包（java.util.concurrent）中提供的一个线程安全的、无界、优先级队列。它的核心思想是：

每次取出的元素，都是当前队列中“优先级最高”的那个元素（即最小值，依据自然排序或自定义比较器）。

一、关键特性总结

特性	说明
线程安全	所有公共操作都通过`ReentrantLock`加锁，支持多线程并发访问。
无界（逻辑上）	理论上可以无限添加元素（但受 JVM 内存限制，可能抛`OutOfMemoryError`）。
不允许 null 元素	插入`null`会抛`NullPointerException`。
基于堆（Heap）实现	底层使用数组表示的二叉堆（最小堆），保证`queue[0]`是优先级最高的元素。
阻塞式取操作	提供`take()`、`poll(timeout)`等方法，在队列为空时可阻塞等待。
不保证迭代顺序	`iterator()`不按优先级顺序遍历！如需有序，必须用`Arrays.sort(toArray())`。
插入/删除时间复杂度	`O(log n)`，因为要维护堆结构。

二、核心机制解析

1.底层数据结构：二叉最小堆

使用Object[] queue存储元素。
对于任意节点i：
- 左孩子：2*i + 1
- 右孩子：2*i + 2
- 父节点：(i - 1) / 2
堆性质：父节点 ≤ 子节点→ 根节点（queue[0]）是最小值（最高优先级）。

2.扩容机制（tryGrow）

当数组满时，自动扩容：
- 小容量（<64）：增长较快（+ oldCap + 2）
- 大容量：增长 50%（oldCap >> 1）
特殊设计：扩容时不持有主锁（lock），而是用CAS 自旋锁（allocationSpinLock）避免阻塞消费者。
- 目的：防止生产者扩容时长时间持有锁，导致消费者“饿死”。

3.堆调整操作

siftUp：插入新元素后，从底部向上调整（冒泡到合适位置）。
siftDown：删除根节点后，把最后一个元素放到根，再向下调整。
分为两种版本：
- siftUpComparable/siftDownComparable：使用元素自身的compareTo()
- siftUpUsingComparator/siftDownUsingComparator：使用外部Comparator

4.构造函数逻辑

如果传入的是SortedSet或PriorityQueue，直接复用其排序规则（无需重新建堆）。
否则，对传入集合调用heapify()从底向上建堆（时间复杂度 O(n)）。

5.阻塞与非阻塞操作

方法	行为
`offer(e)`	立即插入，返回`true`（永不阻塞，因无界）
`put(e)`	同`offer`，语义上“可能阻塞”，但实际不会
`take()`	队列空时阻塞，直到有元素
`poll()`	队列空时立即返回 null
`poll(timeout, unit)`	队列空时最多等待 timeout 时间

6.关于迭代器（重要！）

Iterator<E>it=pq.iterator();// ❌ 不保证按优先级顺序遍历！

原因：堆的数组存储不是排序数组，只是满足堆性质。

正确做法：如需有序遍历，必须：

Object[]arr=pq.toArray();Arrays.sort(arr);// 或使用 Comparator

三、FIFOEntry 示例：解决“优先级相同时的公平性”

当多个元素优先级相同（compareTo == 0），默认不保证谁先出队。

解决方案：引入“插入顺序”作为第二排序键。

classFIFOEntry<EextendsComparable<?superE>>implementsComparable<FIFOEntry<E>>{staticfinalAtomicLongseq=newAtomicLong(0);finallongseqNum;// 插入序号finalEentry;publicintcompareTo(FIFOEntry<E>other){intres=entry.compareTo(other.entry);if(res==0)res=Long.compare(seqNum,other.seqNum);// 先插入的先出returnres;}}