JDK源码分析实战系列-PriorityBlockingQueue

前言

可以通过分析PriorityBlockingQueue来了解JUC中的线程安全的队列实现的一些套路，这些套路会在JUC中其他数据结构实现上反复出现，从而可以更合理的了解那些实现机制背后通用的部分。

BlockingQueue

A Queue that additionally supports operations that wait for the queue to become non-empty when retrieving an element, and wait for space to become available in the queue when storing an element.

阻塞队列，这个接口就非常重要，它是定义了阻塞队列需要实现的接口能力，它的子类有ArrayBlockingQueue,LinkedBlockingQueue,PriorityBlockingQueue等。

作为队列的扩展，扩展的核心能力是阻塞能力，这个阻塞能力表示：当队列空了的时候，获取元素的操作需要阻塞，等待队列有存储新的元素进入。这里容易产生一个误解：认为BlockingQueue也包含了当队列满的时候，放入操作阻塞的能力，这一点并不是BlockingQueue的能力要求，这和子类实现的是有界或无界队列有关。

PriorityBlockingQueue

PriorityBlockingQueue的数据结构的实现是和PriorityQueue是一致的，完全可以参考前一篇的文章，这里重点是了解清楚通过锁来保证线程安全的队列的实现方式，知道了这些知识点，再要理解JUC中其他的队列的实现简直轻而易举。

实现线程安全的关键两个属性：

/**
 * Lock used for all public operations
 */
private final ReentrantLock lock;

/**
 * Condition for blocking when empty
 */
private final Condition notEmpty;

各个操作的时候都执行lock.lock();锁住，操作结束执行lock.unlock();解锁。

比如简单的查看元素数量的方法：

public int size() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return size;
    } finally {
        lock.unlock();
    }
}

有了锁，实现线程安全的操作变得简单而不易出错。下面我们看一些关键的操作方法的实现。

offer

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)
            siftUpComparable(n, e, array);
        else
            siftUpUsingComparator(n, e, array, cmp);
        size = n + 1;
       // 唤醒线程
        notEmpty.signal();
    } finally {
        lock.unlock();
    }
    return true;
}

对于一个元素的位置摆放和优先级队列是一致的，完全可以参考前面一篇PriorityQueue。

take

take方法尝试获取队列头节点的元素，如果为空，就阻塞线程等待，直到队列有元素再唤醒继续执行获取动作。这个Condition notEmpty内部机制可以参考前面的文章

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;
}

通过notEmpty，实现了阻塞队列的核心能力，那就是当获取元素的时候队列空了阻塞线程和当队列有元素的时候进行唤醒动作。这一点在其他阻塞队列中也这样实现，另外，因为这是个无界队列，并不会发生队列满的情况，所以就没有在放入元素的时候处理阻塞的逻辑，而那些有界队列就需要处理这种情况，当然，处理起来也非常简单，再来一个标记队列满了的Condition就可以了。

扩容

这种队列需要扩容，经过前面的一篇，我们已经不再陌生，而对于一个线程安全的队列来说，正在扩容的时候只要确保持有锁，挡住外界的读写操作，就不会有问题。如果你也是这么想的，那么从下面的实现代码中就可以学习到一些优化细节和思路。

private void tryGrow(Object[] array, int oldCap) {
   // 解锁操作，这里需要清楚调用这个方法默认是需要保证获得主锁的
    lock.unlock(); // must release and then re-acquire main lock
    Object[] newArray = null;
   // 对allocationSpinLock进行cas更新，调用的地方是用while包住的，所以没有进入这个if的话还会自旋
    if (allocationSpinLock == 0 &&
        UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset,
                                 0, 1)) {
        try {
           // 扩大的容量计算
            int newCap = oldCap + ((oldCap < 64) ?
                                   (oldCap + 2) : // grow faster if small
                                   (oldCap >> 1));
           // MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8
           // oldCap最大就能到MAX_ARRAY_SIZE，如果计算出来的newCap大于MAX_ARRAY_SIZE，那么就判断一下oldCap+1是不是已经超过了MAX_ARRAY_SIZE，如果超过就抛出OutOfMemoryError异常，也就是说最后一次扩容最大只能到MAX_ARRAY_SIZE的容量，下一次就不行了
            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 放开扩容操作限制
            allocationSpinLock = 0;
        }
    }
   // 这个条件成立意味着cas失败或者allocationSpinLock状态为1，表示有线程正在扩容
    if (newArray == null) // back off if another thread is allocating
       // 让出CPU
        Thread.yield();
    lock.lock();
    if (newArray != null && queue == array) {
        queue = newArray;
        System.arraycopy(array, 0, newArray, 0, oldCap);
    }
}

扩容操作并没有像其他方法一样上来就抢锁，而是进行了lock.unlock()操作，再看下去发现它是使用原子cas更新allocationSpinLock来保证没有其他线程可以并发执行扩容的逻辑代码。这样就优化了在扩容时对其他操作的性能影响。

总结

在清楚了PriorityQueue数据结构后对于理解PriorityBlockingQueue的实现机制就很简单了，主要需要理解到ReentrantLock和Condition的作用，因为前面已经详细进入过AQS系列的世界，现在看来从基础的开始看起，一些花里胡哨的东西融会贯通也是容易的。

后面和PriorityBlockingQueue有点关联的是DelayedWorkQueue和ScheduledThreadPoolExecutor。路线图大概是这样的：
UNSAFE->AQS->ReentrantLock+PriorityQueue->PriorityBlockingQueue->DelayedWorkQueue->ScheduledThreadPoolExecutor