《java.util.concurrent 包源码阅读》19 PriorityBlockingQueue

前面讲ScheduledThreadPoolExecutor曾经重点讲到了DelayedWorkQueue，这里说的PriorityBlockingQueue其实是DelayedWorkQueue的简化版本，实现了按序排列元素的功能。也就是说PriorityBlockingQueue是维护一个按序排列的队列，排序的方法可以通过指定Comparator来比较元素的大小，或者元素类型本身实现了Comparable接口。

因此PriorityBlockingQueue也是使用基于数组的二叉堆来实现的。

首先还是看看offer方法：

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)
// 没有comparator情况下，元素类型必须实现Comparable接口
// 使用compare方法进行比较，然后插入元素到堆中
siftUpComparable(n, e, array);
else
// 制定comparator的情况下，插入元素使用comparator
// 比较元素，然后插入元素到堆中
siftUpUsingComparator(n, e, array, cmp);
size = n + 1;
notEmpty.signal();
} finally {
lock.unlock();
}
return true;
}

这里看一下tryGrow，在对数组进行扩容时释放了主锁的，因为分配空间本身是不需要主锁的，只有更新数组时才会要主锁。

这样可以提高并发执行的性能，减少阻塞。

private void tryGrow(Object[] array, int oldCap) {
// 扩容数组时，释放主锁，这样其他取走元素的操作就可以正常
// 操作了。这里使用一个简单的allocationSpinLock作为锁，
// 它的值为1表示锁正在被使用，为0表示锁为被占用。
// 在获取该锁时，用的CAS操作，而释放时，因为锁已经占用，
// 直接赋值为0即可。
// 分配空间本身是用不到主锁的，只有更新数组的时候才需要。
lock.unlock();
Object[] newArray = null;
if (allocationSpinLock == 0 &&
UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset,
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;
}
}
// 其他线程正在占用allocationSpinLock，调用yield告诉线程调度
// 如果其他线程需要CPU，可以先拿去，我过会再执行，否则我继续执行。
if (newArray == null) // back off if another thread is allocating
Thread.yield();
// 因为要返回，再次获取主锁，而且后面可能要更新数组也需要主锁
lock.lock();
// 如果分配新空间成功，而且原先的队列没有被其他的线程更新过
// 就更新数组。这里不需要使用CAS，因为这个时候已经占用主锁了
if (newArray != null && queue == array) {
queue = newArray;
System.arraycopy(array, 0, newArray, 0, oldCap);
}

再看取走元素的核心方法extract（poll方法也是使用这个方法从堆中拿走元素）

private E extract() {
E result;
int n = size - 1;
if (n < 0)
result = null;
else {
// 取走第一个元素
Object[] array = queue;
result = (E) array[0];
E x = (E) array
;
array
= 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;
}

这里提一下PriorityBlockingQueue的序列化。PriorityBlockingQueue内置了一个PriorityQueue对象，序列化会把元素转存到这个PriorityQueue中，然后再进行序列化。

反序列化时也是用PriorityQueue读取，然后再把元素转存回PriorityBlockingQueue自己的队列。

private PriorityQueue q;

下一篇会讲具有DelayedWorkQueue的另外一个功能延迟执行的DelayQueue。