PriorityBlockingQueue 是二叉堆结构的无界优先级阻塞队列，使用显示锁 Lock 保证线程安全，是一个线程安全的 PriorityQueue。元素的优先级顺序通过 Comparator 实现，内部不可添加不可比较的值。相较于我们前几章所讲的Queue，PriorityBlockingQueue 算是一个老牌的队列了，从JDK1.5加入JUC行列，如果我们需要对队列进行优先级排序，PriorityBlockingQueue 将是一个不错的选择。
数据结构及核心参数
//默认容量
private static final int DEFAULT_INITIAL_CAPACITY = 11;
//数组最大长度
private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
//内部存储元素的数组，基于二叉堆实现
private transient Object[] queue;
//Lock used for all public operations
private final ReentrantLock lock;
//Condition for blocking when empty
private final Condition notEmpty;
/**初始0为可获取状态，用于控制扩容操作*/
private transient volatile int allocationSpinLock;
//比较器
private transient Comparator<? super E> comparator;
//内部PriorityQueue引用，用于兼容序列化
private PriorityQueue<E> q;
二叉堆
PriorityBlockingQueue（后称PBQ）的数据由内部的一个 Object 数组(queue)存储，这个数组本质上是一个二叉堆：
二叉堆
二叉堆是一种特殊的堆，二叉堆是完全二叉树或者是近似完全二叉树。二叉堆满足堆特性：父节点的键值总是与任何一个子节点的键值保持固定的序关系，且每个节点的子节点也都是一个二叉堆。
当父节点的键值总是大于或等于任何一个子节点的键值时为最大堆。 当父节点的键值总是小于或等于任何一个子节点的键值时为最小堆。
二叉堆一般用数组来表示。如果根节点在数组中的位置是1，第n个位置的子节点分别在 2*n 和 2*n+1。因此，第1个位置的子节点在2和3，第2个位置的子节点在4和5。以此类推。这种基于1的数组存储方式便于寻找父节点和子节点。如果存储数组的下标基于0，那么下标为i的节点的子节点是 2*n + 1 与 2*(n+1)；其父节点的下标是 (n − 1)/2，PriorityBlockingQueue 中使用的就是基于0下标的二叉堆。
源码解析
如果你理解了我们前几章的内容，会发现 PBQ 的入队/出队操作都很简单，没有太复杂的算法，无非就是对二叉堆的插入/删除操作。
offer(E e)
//入队
public boolean offer(E e) {
if (e == null)
throw new NullPointerException();
final ReentrantLock lock = this.lock;
lock.lock();
int n, cap;
Object[] array;
while ((n = size) >= (cap = (array = queue).length))
//队列扩容
tryGrow(array, cap);
try {
Comparator<? super E> cmp = comparator;
//找到合适位置插入元素
if (cmp == null)
siftUpComparable(n, e, array);
else
siftUpUsingComparator(n, e, array, cmp);
size = n + 1;
notEmpty.signal();
} finally {
lock.unlock();
}
return true;
}
说明：首先对队列加锁，判断队列是否需要扩容，如果需要调用tryGrow()方法进行扩容；然后调用siftUpComparable()方法找到合适位置插入元素；更新队列元素数size，唤醒等待notEmpty的线程，最后别忘了解除锁定unlock。
- PBQ 的扩容说明：当队列元素数大于等于数组的长度时，会触发扩容操作，扩容是由单线程完成的。如果数组长度 cap 小于64，扩容长度为 2*(cap+1)；否则扩容长度为原来的1.5倍(1.5*cap)，
tryGrow()源码如下：
private void tryGrow(Object[] array, int oldCap) {
lock.unlock(); // must release and then re-acquire main lock
Object[] newArray = null;
if (allocationSpinLock == 0 &&
UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset,
0, 1)) {
try {
//计算新的数组容量（1.5*cap或2*(cap+1)）
int newCap = oldCap + ((oldCap < 64) ?
(oldCap + 2) : // grow faster if small
(oldCap >> 1));
if (newCap - MAX_ARRAY_SIZE > 0) { // possible overflow
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) // back off if another thread is allocating
Thread.yield();
lock.lock();
if (newArray != null && queue == array) {
queue = newArray;
System.arraycopy(array, 0, newArray, 0, oldCap);
}
}
-
siftUpComparable()和siftUpUsingComparator()，由于队列中的元素都是有优先级（基于comparator排序）的，所以如果有新元素进来不会像其他队列一样直接放在队尾，而是通过这两个方法找到新增元素在队列中的排序位置然后插入，源码如下：
//在k位置插入元素x，从父节点开始向上找到合适位置，保持二元堆的性质不变
private static <T> void siftUpComparable(int k, T x, Object[] array) {
Comparable<? super T> key = (Comparable<? super T>) x;
while (k > 0) {
//从父节点开始向上查找，并保持二叉堆性质
int parent = (k - 1) >>> 1;
Object e = array[parent];
if (key.compareTo((T) e) >= 0)
break;//找到合适位置，跳出循环
array[k] = e;
k = parent;
}
array[k] = key;
}
//自定义Comparator版本的siftUpComparable
private static <T> void siftUpUsingComparator(int k, T x, Object[] array,
Comparator<? super T> cmp) {
while (k > 0) {
//从父节点开始向上查找，并保持二叉堆性质
int parent = (k - 1) >>> 1;
Object e = array[parent];
if (cmp.compare(x, (T) e) >= 0)
break;//找到合适位置，跳出循环
array[k] = e;
k = parent;
}
array[k] = x;
}
poll()
//出列
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return dequeue();
} finally {
lock.unlock();
}
}
/**
* Mechanics for poll(). Call only while holding lock.
*/
private E dequeue() {
int n = size - 1;
if (n < 0)
return null;
else {
Object[] array = queue;
E result = (E) array[0];//第一个元素，即出列元素
E x = (E) array[n];//最后一个元素
array[n] = null;
Comparator<? super E> cmp = comparator;
//重构二叉堆
if (cmp == null)
siftDownComparable(0, x, array, n);
else
siftDownUsingComparator(0, x, array, n, cmp);
size = n;
return result;
}
}
说明： 在出列操作时，首先移除数组的最后一个元素，然后调用siftDownComparable或siftDownUsingComparator方法进行二叉堆的重组，最后返回队列的第一个元素。
PBQ的出列本质上是删除二叉堆的根节点，然后，把堆存储的最后那个节点移到填在根节点处，再从上而下调整父节点与它的子节点。
siftDownComparable和siftDownUsingComparator源码如下：
//在k位置插入元素x，从子节点开始向下调整节点位置，保持二叉堆的性质不变
private static <T> void siftDownComparable(int k, T x, Object[] array,
int n) {
if (n > 0) {
Comparable<? super T> key = (Comparable<? super T>)x;
//获取最后一个节点的父节点
int half = n >>> 1; // loop while a non-leaf
while (k < half) {
//从左叶子节点向下调整
int child = (k << 1) + 1; // assume left child is least
Object c = array[child];
int right = child + 1;
if (right < n &&
((Comparable<? super T>) c).compareTo((T) array[right]) > 0)
c = array[child = right];
if (key.compareTo((T) c) <= 0)
break;
array[k] = c;
k = child;
}
array[k] = key;
}
}
private static <T> void siftDownUsingComparator(int k, T x, Object[] array,
int n,
Comparator<? super T> cmp) {
if (n > 0) {
int half = n >>> 1;
while (k < half) {
int child = (k << 1) + 1;
Object c = array[child];
int right = child + 1;
if (right < n && cmp.compare((T) c, (T) array[right]) > 0)
c = array[child = right];
if (cmp.compare(x, (T) c) <= 0)
break;
array[k] = c;
k = child;
}
array[k] = x;
}
}
小结
本章重点：理解 PriorityBlockingQueue 的优先级策略实现，二叉堆数据结构。
网友评论