PriorityQueue / PriorityBlockingQueue 堆源码
概述
PriorityQueue 和 PriorityBlockingQueue 是基于二叉堆（Binary Heap）实现的优先级队列。它们按照元素的自然顺序或 Comparator 定义的顺序进行排序，保证每次取出的元素都是当前队列中的最小（或最大）元素。
PriorityQueue 是非线程安全的，PriorityBlockingQueue 则是线程安全的版本，通过 ReentrantLock 实现并发控制，适用于多线程环境下的优先级调度场景。
两者都保证了 O(log n) 的入队和出队时间复杂度。
本文基于 OpenJDK 21 源码，从二叉堆的数组结构开始，逐层深入到上浮、下沉、扩容等核心操作。
本文基于 OpenJDK 21 源码分析，关键实现会对比 JDK 8 到 JDK 21 的版本差异。
核心源码解析
① PriorityQueue 的二叉堆
PriorityQueue 使用数组实现二叉堆：
java
public class PriorityQueue<E> extends AbstractQueue<E>
implements java.io.Serializable {
private static final int DEFAULT_INITIAL_CAPACITY = 11;
transient Object[] queue; // 非 private 以便嵌套类访问
private int size = 0;
private final Comparator<? super E> comparator;
private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
}核心特性：
- 底层是
Object[]数组，索引从 0 开始 - 默认小顶堆：
queue[0]始终是最小元素 - 节点关系：
queue[n]的子节点是queue[2n+1]（左）和queue[2n+2]（右） - 父节点：
queue[n]的父节点是queue[(n-1) >>> 1]
② siftUp(int, E) 的上浮
add() 和 offer() 时调用，在数组尾部插入元素并上浮到正确位置：
java
public boolean offer(E e) {
if (e == null)
throw new NullPointerException();
modCount++;
int i = size;
if (i >= queue.length)
grow(i + 1); // 扩容
size = i + 1;
if (i == 0)
queue[0] = e; // 第一个元素
else
siftUp(i, e); // 上浮
return true;
}siftUp() 的实现：
java
private void siftUp(int k, E x) {
if (comparator != null)
siftUpUsingComparator(k, x, queue, comparator);
else
siftUpComparable(k, x, queue);
}
private static <T> void siftUpComparable(int k, T x, Object[] es) {
Comparable<? super T> key = (Comparable<? super T>) x;
while (k > 0) {
int parent = (k - 1) >>> 1; // 无符号右移 = 整除 2
Object e = es[parent];
if (key.compareTo((T) e) >= 0)
break; // 已满足堆性质
es[k] = e; // 父节点下移
k = parent;
}
es[k] = key; // 找到插入位置
}- 从插入位置
k开始，比较x与其父节点 - 若
x < 父节点→ 父节点下移，继续向上比较 - 若
x ≥ 父节点→ 找到正确位置，退出循环
③ siftDown(int, E) 的下沉
poll() 和 remove() 时调用，将最后一个元素放到根节点并下沉调整：
java
private static <T> void siftDownComparable(int k, T x, Object[] es, int n) {
Comparable<? super T> key = (Comparable<? super T>) x;
int half = n >>> 1; // 非叶子节点数量
while (k < half) {
int child = (k << 1) + 1; // 左子节点
Object c = es[child];
int right = child + 1; // 右子节点
if (right < n &&
((Comparable<? super T>) c).compareTo((T) es[right]) > 0)
c = es[child = right]; // 取两个子节点中较小的
if (key.compareTo((T) c) <= 0)
break; // 已满足堆性质
es[k] = c; // 子节点上移
k = child;
}
es[k] = key;
}n >>> 1是第一个叶子节点的索引- 从根节点开始，比较子节点中较小的那个
- 若
key > 子节点→ 子节点上移，继续下沉 - 若
key ≤ 子节点→ 找到正确位置
④ grow(int) 的扩容
java
private void grow(int minCapacity) {
int oldCapacity = queue.length;
// 小容量（<64）翻倍+2，大容量（≥64）1.5 倍
int newCapacity = oldCapacity + ((oldCapacity < 64) ?
(oldCapacity + 2) :
(oldCapacity >> 1));
// 溢出检查
if (newCapacity - MAX_ARRAY_SIZE > 0)
newCapacity = hugeCapacity(minCapacity);
queue = Arrays.copyOf(queue, newCapacity);
}扩容策略：
oldCapacity < 64：new = old + (old + 2)，近似 2 倍 + 2oldCapacity ≥ 64：new = old + (old >> 1)，1.5 倍
这种策略在队列较小时快速增长，避免频繁扩容；队列较大时平缓增长，节省内存。
⑤ poll() 的根取出
java
public E poll() {
final Object[] es;
final E result;
if ((result = (E) ((es = queue)[0])) != null) {
modCount++;
final int n = --size;
es[0] = es[n]; // 最后一个元素放到根节点
es[n] = null; // 清除原位置
if (n > 0)
siftDown(0, es[0]); // 下沉
}
return result;
}流程：
- 取出
queue[0]作为返回值（最小元素） - 将最后一个元素
queue[n]移到根节点位置 - 最后一个位置置
null辅助 GC - 从根节点开始
siftDown()下沉调整堆结构
⑥ PriorityBlockingQueue 的 ReentrantLock
java
public class PriorityBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
private final ReentrantLock lock;
private final Condition notEmpty;
private PriorityQueue<E> q; // 委托给 PriorityQueue
private transient volatile int allocationSpinLock;
public PriorityBlockingQueue() {
this(DEFAULT_INITIAL_CAPACITY, null);
}
public PriorityBlockingQueue(int initialCapacity, Comparator<? super E> comparator) {
this.lock = new ReentrantLock();
this.notEmpty = lock.newCondition();
this.comparator = comparator;
this.q = new PriorityQueue<E>(initialCapacity, comparator);
}
}put(E) 实现：
java
public void put(E e) {
offer(e); // 无界队列，不阻塞
}
public boolean offer(E e) {
if (e == null) throw new NullPointerException();
final ReentrantLock lock = this.lock;
lock.lock();
int n, cap;
Object[] es;
while ((n = size) >= (cap = (es = queue).length))
tryGrow(es, cap); // 扩容（只有一个线程执行）
try {
Comparator<? super E> cmp = comparator;
if (cmp == null)
siftUpComparable(n, e, es);
else
siftUpUsingComparator(n, e, es, cmp);
size = n + 1;
notEmpty.signal(); // 通知等待的 take()
} finally {
lock.unlock();
}
return true;
}take() 实现（阻塞获取）：
java
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
E result;
try {
while ((result = dequeue()) == null)
notEmpty.await(); // 队列为空时等待
} finally {
lock.unlock();
}
return result;
}⑦ PriorityBlockingQueue 的数组自旋增长
扩容时为了减少锁持有时间，PriorityBlockingQueue 使用 allocationSpinLock 进行自旋：
java
private void tryGrow(Object[] array, int oldCap) {
lock.unlock(); // 先释放锁，允许其他线程继续操作
Object[] newArray = null;
if (allocationSpinLock == 0 &&
U.compareAndSwapInt(this, ALLOCATIONSPINLOCK, 0, 1)) {
// 只有一个线程能进入
try {
int newCap = oldCap + ((oldCap < 64) ?
(oldCap + 2) :
(oldCap >> 1));
if (newCap - MAX_ARRAY_SIZE > 0) {
int minCap = oldCap + 1;
if (minCap < 0 || minCap > MAX_ARRAY_SIZE)
throw new OutOfMemoryError();
newCap = MAX_ARRAY_SIZE;
}
if (newCap > oldCap && queue == array)
newArray = new Object[newCap];
} finally {
allocationSpinLock = 0;
}
}
if (newArray == null) // 其他线程让出 CPU
Thread.yield();
lock.lock(); // 重新获取锁
if (newArray != null && queue == array) {
queue = newArray;
System.arraycopy(array, 0, newArray, 0, oldCap);
}
}核心优化：
- 先释放
lock，允许其他线程并发操作已有数据 - CAS 竞争
allocationSpinLock，只有一个线程分配新数组 - 非竞争的线程
Thread.yield()等待 - 重新获取锁后完成拷贝
总结
PriorityQueue / PriorityBlockingQueue 的二叉堆实现提供了高效的优先级队列操作：
| 操作 | 时间复杂度 | 说明 |
|---|---|---|
offer(E) | O(log n) | 尾部插入 + siftUp 上浮 |
poll() | O(log n) | 根取出 + siftDown 下沉 |
peek() | O(1) | 直接返回 queue[0] |
size() | O(1) | 直接返回 size |
核心设计要点：
- 数组实现二叉树：通过下标计算父子关系，省去指针开销
- 默认小顶堆：
queue[0]始终最小 - 动态扩容：小容量快速翻倍，大容量平缓增长
- 线程安全：
PriorityBlockingQueue通过ReentrantLock+allocationSpinLock保证并发