JUC PriorityBlockingQueue 分析

红太狼 2022-05-21 13:42 171阅读 0赞

## 基本介绍 ##

`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的存储
        }
    }