前面讲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[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; }
这里提一下PriorityBlockingQueue的序列化。PriorityBlockingQueue内置了一个PriorityQueue对象,序列化会把元素转存到这个PriorityQueue中,然后再进行序列化。
反序列化时也是用PriorityQueue读取,然后再把元素转存回PriorityBlockingQueue自己的队列。
private PriorityQueue q;
下一篇会讲具有DelayedWorkQueue的另外一个功能延迟执行的DelayQueue。