基本介绍

PriorityBlockingQueue JUC中阻塞队列的一种，支持阻塞的优先队列，容量最大可到int的最大值 – 8（Integer.MAX_VALUE – 8），容量可以自动扩展（内部实现）。PriorityBlockingQueue内部使用最小堆的思想来构建，使用数组形式存储元素

构造时建议定义好Comparator的实现并传入，在内部实现里面将使用传入的Comparator来进行元素的比较，若没有定义此比较器，那么内部将默认采用元素的自然序来完成比较（Comparable 的 compareTo，若元素不是Comparable接口的实现类，那么会抛异常）

内部定义的数组通过system.arraycopy来实现扩容

源码分析

需要提前了解清楚最小堆的概念以及它的一些操作，如插入，删除，最小堆调整等等

先看下PriorityBlockingQueue的重要成员变量:

private transient Object[] queue; // 存放数据的数组，用来构建最小堆
private transient int size; // 队列中元素的个数
private transient Comparator<? super E> comparator; // 构造函数传入的比较器，最小堆维护的时候需要用到
private final ReentrantLock lock; // 用于维护多线程并发操作正确性的锁
private final Condition notEmpty; // take操作需要阻塞等待的条件
private transient volatile int allocationSpinLock; // 扩容时为了减少lock锁锁定范围而配合使用的标志（配合CAS实现无锁并发操作）
private PriorityQueue<E> q; // 用于控制序列化和反序列的普通优先队列，不需太关注

与其他阻塞队列分析类似，我们重点分析put和take:

put

因为队列容量最大可接近于int的最大值，所以可以理解为PriorityBlockingQueue是无界队列，因此对于put操作没有满的定义，不需要阻塞等待，只需控制锁即可，put直接调用offer的实现，因此我们重点看offer方法:

同理，因为容量没有上限，因此超时的offer也是直接调用了offer方法

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[] 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(); // 尝试唤醒一个take操作阻塞等待的线程
    } finally {
        lock.unlock(); // 释放锁
    }
    return true;
}

看下tryGrow方法，为了尽量减少lock锁的范围代码，先解锁，用CAS控制来计算新的容量以及创建新数组，然后在调用system.arraycopy之前，再获取锁:

private void tryGrow(Object[] array, int oldCap) {
    lock.unlock(); // must release and then re-acquire main lock
    Object[] newArray = null;
    // 为了减少lock锁的锁定范围，用CAS控制来计算新的容量以及分配新的数组
    if (allocationSpinLock == 0 &&
        UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset,
                                 0, 1)) {
        try {
            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) // 其他并发操作的线程正在分配新的数组，让出CPU资源
        Thread.yield();
    lock.lock();
    if (newArray != null && queue == array) { // 重新获取锁且完成新数组分配的那个线程，进行copy数组元素的操作
        queue = newArray;
        System.arraycopy(array, 0, newArray, 0, oldCap);
    }
}

siftUpUsingComparator方法完成元素新增以及最小堆的重新调整（自下而上调整）

private static <T> void siftUpUsingComparator(int k, T x, Object[] array,
                                   Comparator<? super T> cmp) {
    // 循环从最下层往上，若节点元素比x小，则小的元素上移
    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;
    }
    // 最后处理新增元素x的填入
    array[k] = x;
}

siftUpComparable实现原理类似，只是使用Comparable的compare来完成比较，这里就不贴这个方法的代码了

take

接下来我们分析take方法，不同于put，需要判断队列是否为空，为空则需要阻塞等待

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

重点逻辑在dequeue方法里面:

poll以及超时的poll调用的也是dequeue，这两个方法也不贴代码，自行看下源码即可

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

根据最小堆的移除，每次都是从堆顶元素中移除，移除动作需要对最小堆进行重新调整，以形成新的最小堆，逻辑在siftDownX方法里面，我们重点看下有定义比较器的方法siftDownUsingComparator:

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) // left和right取最小的
                c = array[child = right];
            if (cmp.compare(x, (T) c) <= 0) // 当前节点与x比较，x小，证明找到x的存放位置，跳出
                break;
            array[k] = c; // 小的元素上移，继续往下走
            k = child;
        }
        array[k] = x; // 最后处理x的存储
    }
}