JUC集合类 PriorityBlockingQueue源码解析 JDK8

灰太狼 2023-03-03 11:11 189阅读 0赞

### 文章目录 ###

*  前言
 *  成员
 *  构造器
 *   *  原地建堆
 *  入队
 *   *  offer
     *  扩容
 *  出队
 *   *  poll
 *  获取堆顶方法
 *   *  peek
 *  内部删除
 *  迭代器
 *  总结

# 前言 #

PriorityBlockingQueue是一个无界阻塞队列，它的出队方式不再是FIFO，而是优先级高的先出队。其内部实现是最小堆，即堆顶元素是逻辑上最小的那个元素，也是最先出队的那个元素。简单的说，如果`a.compareTo(b) < 0`的话，那么`a`将先出队。

关于最小堆的相关知识，请看[从小顶堆到堆排序——超详细图解][Link 1]这篇文章。

[JUC框架 系列文章目录][JUC_]

# 成员 #

//默认数组的容量
        private static final int DEFAULT_INITIAL_CAPACITY = 11;
    
        //数组的最大容量，减8是因为有的虚拟机实现里数组的前8个字节用来存储别的东西
        private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
    
        //内部数组，逻辑上是一个堆
        private transient Object[] queue;
    
        //元素个数，小于等于queue.length
        private transient int size;
    
        //比较器
        private transient Comparator<? super E> comparator;
    
        //唯一的锁，用来保证并发安全和可见性
        private final ReentrantLock lock;
    
        //队列空时，出队线程将阻塞在这里
        private final Condition notEmpty;
    
        //相当于AQS的state，持有这个state才可以准备新数组以扩容
        private transient volatile int allocationSpinLock;

# 构造器 #

public PriorityBlockingQueue() { 
        this(DEFAULT_INITIAL_CAPACITY, null);
    }
    
    public PriorityBlockingQueue(int initialCapacity) { 
        this(initialCapacity, null);
    }
    
    public PriorityBlockingQueue(int initialCapacity,
                                 Comparator<? super E> comparator) { 
        if (initialCapacity < 1)//检查
            throw new IllegalArgumentException();
    
        this.lock = new ReentrantLock();
    
        this.notEmpty = lock.newCondition();
        //没有notFull,因为这是一个无界队列
        
        this.comparator = comparator;
    
        this.queue = new Object[initialCapacity];
    }

最终调用的都是同一个构造器，按照`initialCapacity`新建数组，因为它是int型的，所以最大是`Integer.MAX_VALUE`。注意，`initialCapacity`并没有和`MAX_ARRAY_SIZE`进行比较，所以完全可能创建出`Integer.MAX_VALUE`大小的数组。

public PriorityBlockingQueue(Collection<? extends E> c) { 
            this.lock = new ReentrantLock();
            this.notEmpty = lock.newCondition();
            boolean heapify = true; // 为true代表数组需要重新建堆
            boolean screen = true;  // 为true代表需要扫描一遍数组，看里面有没有null元素（此类不支持null元素）
            if (c instanceof SortedSet<?>) { 
                SortedSet<? extends E> ss = (SortedSet<? extends E>) c;
                this.comparator = (Comparator<? super E>) ss.comparator();
                heapify = false;//SortedSet的内部数据已经按照升序排序好了，自然也是堆结构的
            }
            else if (c instanceof PriorityBlockingQueue<?>) { 
                PriorityBlockingQueue<? extends E> pq =
                    (PriorityBlockingQueue<? extends E>) c;
                this.comparator = (Comparator<? super E>) pq.comparator();
                screen = false;//PriorityBlockingQueue不可能包含null元素
                if (pq.getClass() == PriorityBlockingQueue.class) // exact match
                    heapify = false;
            }
            Object[] a = c.toArray();
            int n = a.length;
    
            if (a.getClass() != Object[].class)
                a = Arrays.copyOf(a, n, Object[].class);
            if (screen && (n == 1 || this.comparator != null)) { //检查null元素
                for (int i = 0; i < n; ++i)
                    if (a[i] == null)//检查到有null元素，直接抛出异常
                        throw new NullPointerException();
            }
            this.queue = a;
            this.size = n;
            if (heapify)//建堆
                heapify();
        }

这个构造器有几个地方可能比较难懂，我单独写了一篇[PriorityBlockingQueue构造器解析][PriorityBlockingQueue]进行讲解。

当数组在逻辑上不是最小堆的结构时，需要调用`heapify`建立一个最小堆。

## 原地建堆 ##

private void heapify() { 
            Object[] array = queue;
            int n = size;
            int half = (n >>> 1) - 1;//最后一个非叶子节点的索引
            Comparator<? super E> cmp = comparator;
            if (cmp == null) { 
                //从最后一个非叶子节点，反向层次遍历，以遍历节点为root的子树将被构建成最小堆
                for (int i = half; i >= 0; i--)
                    siftDownComparable(i, (E) array[i], array, n);
            }
            else { 
                //同上
                for (int i = half; i >= 0; i--)
                    siftDownUsingComparator(i, (E) array[i], array, n, cmp);
            }
        }

上面两个函数实际就是最小堆构建过程中的冒泡下移操作。分析之前，我们先来记一下`Comparable`和`Comparator`比较的返回值的含义：

*  `a.compareTo(b) < 0`，不管`compareTo`是怎么实现的，总之，`a.compareTo(b) < 0`代表`a < b`。由于PriorityBlockingQueue是最小堆，所以a会放到数组的前面去。
 *  `cmp.compare(a, b) < 0`，不管`Comparator`是怎么实现的，总之，`cmp.compare(a, b) < 0`代表`a < b`。

private static <T> void siftDownComparable(int k, T x, Object[] array,
                                                   int n) { 
            if (n > 0) { 
                Comparable<? super T> key = (Comparable<? super T>)x;
                int half = n >>> 1;// 这是第一个叶子节点的索引，也就是说当k到达一个叶子节点时，它就不能再下沉了
                while (k < half) { 
                    int child = (k << 1) + 1; // 左孩子的索引
                    Object c = array[child];  // 获得左孩子
                    int right = child + 1;
                    if (right < n &&
                        ((Comparable<? super T>) c).compareTo((T) array[right]) > 0)
                        //如果右孩子存在，而右孩子更小的话
                        c = array[child = right];//更新child和c
                    //如果key小于等于两个孩子的较小值，说明key停留在k索引处，刚好构成了最小堆
                    if (key.compareTo((T) c) <= 0)
                        break;
                    //如果key大于两个孩子的较小值
                    array[k] = c;//孩子较小值上移
                    k = child;//遍历索引k下移
                }
                //退出循环说明key的停留位置已经确定，就是现在的k的值
                array[k] = key;
            }
        }

简单的说，直到找到合适的位置之前（合适的位置指，key停留在k索引处，刚好构成了最小堆），key不停的冒泡下移。只是实际操作中，并不会每次冒泡下移都把key赋值过去，只有最终确定位置后，才把key赋值过去。示意图如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FubGlhbjUyMw_size_16_color_FFFFFF_t_70]

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)
                        c = array[child = right];
                    if (cmp.compare(x, (T) c) <= 0)
                        break;
                    array[k] = c;
                    k = child;
                }
                array[k] = x;
            }
        }

`siftDownUsingComparator`分析完全类似，只是比较时使用的是`Comparator`而已。

# 入队 #

由于PriorityBlockingQueue是无界的，所以入队是不可能因为队列满而阻塞的，但有可能因为内存耗尽而抛出OutOfMemoryError。

## offer ##

public boolean offer(E e) { 
            if (e == null)//不允许null元素
                throw new NullPointerException();
            final ReentrantLock lock = this.lock;
            lock.lock();
            int n, cap;
            Object[] array;
            while ((n = size) >= (cap = (array = queue).length))//数组每个元素都不是null，再加一个就得扩容
                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();//通知一个出队线程（只有出队线程可能阻塞在AQS条件队列里）
            } finally { 
                lock.unlock();
            }
            return true;
        }

元素入队可能会破坏最小堆性质，所以需要调用函数保证最小堆。注意传入的n为元素个数，即下一个节点的索引（现有元素在数组中的索引为`0 ~ n-1`）。

private static <T> void siftUpComparable(int k, T x, Object[] array) { 
            Comparable<? super T> key = (Comparable<? super T>) x;
            while (k > 0) { //如果到达root，自然也不用上移了
                int parent = (k - 1) >>> 1;//不管左右孩子，这句都能得到父节点索引
                Object e = array[parent];
                if (key.compareTo((T) e) >= 0)//如果key已经大于等于它的父节点，说明现在已经是最小堆了，
                    break;
                //如果key已经小于它的父节点，说明key需要冒泡上移
                array[k] = e;//把父节点的值弄下来
                k = parent;//没有实际把key上移，只是把k索引上移
            }
            //退出循环说明key的停留位置已经确定，就是现在的k的值
            array[k] = key;
        }

简单的说，直到找到合适的位置之前（合适的位置指，key停留在k索引处，刚好构成了最小堆），key不停的冒泡上移。只是实际操作中，并不会每次冒泡上移都把key赋值过去，只有最终确定位置后，才把key赋值过去。示意图如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FubGlhbjUyMw_size_16_color_FFFFFF_t_70 1]

private static <T> void siftUpUsingComparator(int k, T x, Object[] array,
                                           Comparator<? super T> cmp) { 
            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;
            }
            array[k] = x;
        }

`siftUpUsingComparator`分析完全类似，只是比较时使用的是`Comparator`而已。

## 扩容 ##

从`offer`的流程来看，完全有可能多个线程同时进入`tryGrow`尝试扩容。

private void tryGrow(Object[] array, int oldCap) { 
            lock.unlock(); // 释放锁以允许出队等操作并行
            //准备新数组
            Object[] newArray = null;
            if (allocationSpinLock == 0 &&
                UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset,
                                         0, 1)) { //相当于AQS的独占锁，同时只能有一个线程持有这个“锁”
                try { 
                    //新容量的计算公式：
                    //1. 如果oldCap < 64, 新容量等于2(oldCap + 1)
                    //2. 如果oldCap >=64, 新容量等于1.5oldCap
                    int newCap = oldCap + ((oldCap < 64) ?
                                           (oldCap + 2) : // grow faster if small
                                           (oldCap >> 1));
                    if (newCap - MAX_ARRAY_SIZE > 0) {     // 如果新容量超过了MAX_ARRAY_SIZE
                        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) // 如果没有新建成功，说明别的线程新建成功了
                Thread.yield();  //尽量让出cpu，好让成功的线程先来执行下面的流程
            lock.lock();
            if (newArray != null && queue == array) { 
                queue = newArray;//赋值新数组
                System.arraycopy(array, 0, newArray, 0, oldCap);//旧数组的元素也复制过去
            }
        }

整个函数的中间部分都是在准备一个新数组，但这段时间是释放了lock的。这是因为准备新数组的期间，不应该让出队线程因为获取不到锁而阻塞。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FubGlhbjUyMw_size_16_color_FFFFFF_t_70 2]  
上图展示了扩容线程和出队线程并行执行的过程。  
但对于其他的入队线程来说，如果`Thread.yield()`没有让出CPU的话，那么其他入队线程就只有自旋了（`while ((n = size) >= (cap = (array = queue).length))`）。

而最后的`lock.lock()`也是有必要的：

*  先锁住了，防止别的线程搞破坏。这个目的和正常出队入队加锁的目的一样。
 *  可见性。加锁强制让所有线程看到新数组成员。

`&& queue == array`判断也是很有必要的，因为两个实参一样的调用`tryGrow`的两个线程，线程1新建数组后`allocationSpinLock = 0`，线程2才去CAS修改`allocationSpinLock`，然后线程2自己又会另外新建一个数组。当线程1替换新数组后，线程2的新数组自然不应该替换过去，所以我们通过`&& queue == array`判断。总之，只能有一个线程替换新数组成功。

# 出队 #

因为队列可能没有元素，所以出队线程是可能阻塞在AQS条件队列里的。

## poll ##

只是一次尝试，完全有可能`poll`返回null（队列为空）。

public E poll() { 
            final ReentrantLock lock = this.lock;
            lock.lock();
            try { 
                return dequeue();//可能返回null
            } finally { 
                lock.unlock();
            }
        }

出队过程：

1.  暂存堆顶元素（最小值），函数结束前返回它。
2.  把堆的最后一个节点放到堆顶，然后下沉它，再次使得整个堆变成最小堆。

private E dequeue() { 
            int n = size - 1;//最后一个节点的索引
            if (n < 0)
                return null;//如果队列为空，返回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);//因为放到堆顶，所以从0索引开始下沉
                else
                    siftDownUsingComparator(0, x, array, n, cmp);
                size = n;
                return result;//返回堆顶
            }
        }

下沉函数之前已经讲过。

# 获取堆顶方法 #

## peek ##

public E peek() { 
            final ReentrantLock lock = this.lock;
            lock.lock();
            try { 
                return (size == 0) ? null : (E) queue[0];
            } finally { 
                lock.unlock();
            }
        }

此函数也必须加锁，不然遇到出队过程中的中间过程。比如下图的第1步。当然，第1步肯定是正确的新堆顶。所以，本函数重点还是在于可见性，不然你可能看不到下图的第1步的结果。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FubGlhbjUyMw_size_16_color_FFFFFF_t_70]

# 内部删除 #

public boolean remove(Object o) { 
            final ReentrantLock lock = this.lock;
            lock.lock();
            try { 
                int i = indexOf(o);
                if (i == -1)//如果队列中没有这个相等元素
                    return false;
                //如果队列中有这个相等元素
                removeAt(i);
                return true;
            } finally { 
                lock.unlock();
            }
        }
    
        private int indexOf(Object o) { 
            if (o != null) { 
                Object[] array = queue;
                int n = size;
                for (int i = 0; i < n; i++)//从头找到第一个遇到相等的元素的索引
                    if (o.equals(array[i]))
                        return i;
            }
            return -1;
        }

重点还是在于内部删除对于最小堆的影响。

private void removeAt(int i) { 
            Object[] array = queue;
            int n = size - 1;
            if (n == i) // 如果刚好删除的是最后一个叶子节点，那么不会影响最小堆的性质，直接删除即可
                array[i] = null;
            else { 
                E moved = (E) array[n];//即将把最后一个叶子节点移动到 删除处，先暂存起来
                array[n] = null;
                Comparator<? super E> cmp = comparator;
                if (cmp == null)
                    siftDownComparable(i, moved, array, n);//将其移动到i处，可能需要下沉
                else
                    siftDownUsingComparator(i, moved, array, n, cmp);
                if (array[i] == moved) { //这种情况说明移动过去后，根本没有下沉（如果有下沉，i处肯定会变成一个比moved小的数）
                    if (cmp == null)
                        siftUpComparable(i, moved, array);//上移
                    else
                        siftUpUsingComparator(i, moved, array, cmp);
                }
            }
            size = n;
        }

前半段代码和删除顶点的逻辑一样，只不过删除顶点时被删除节点的索引为0，`removeAt(int i)`删除的是`i`索引节点。总之，就是把最后一个节点放到删除处`i`，然后再使用下沉函数。

调用下沉函数后，节点不一定会下移。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FubGlhbjUyMw_size_16_color_FFFFFF_t_70 3]

如上图，值为8的节点移动到`i`处后，会下沉，但不必上移。

*  如果`if (array[i] == moved)`不成立，说明节点下沉了。不用管被删除节点的值是什么（所以图中是问号），现在被删除节点有父节点parent和左右孩子left、right，在删除前`parent <= left && parent <= right`是肯定成立的，现在`moved`下沉了说明`moved`肯定大于`min(left, right)`，那么`parent < moved`肯定成立。
    
     *  因为`moved`肯定大于`min(left, right)`，所以需要执行下沉函数（`siftDownXXX`），执行完毕后，以`i`索引为root的子树则已经是最小堆了。
     *  因为`parent < moved`成立，所以moved的加入对`i`索引往上的层次没有影响。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FubGlhbjUyMw_size_16_color_FFFFFF_t_70 4]  
如上图，值为6的节点移动到`i`处后，不需要下沉，但需要上移。

*  如果`if (array[i] == moved)`成立，说明节点没有下沉，节点移动过去，以索引`i`为root的子树马上就成为了一个最小堆了。但移动过去的节点跟上层节点的关系还没确定，因为没有下沉，则`parent < moved`就不能说一定成立了，比如上图这种情况。
    
     *  如果不是上图这种情况，上移函数（`siftUpXXX`）也不会把节点上移（此时，节点根本不需要就形成了最小堆）。
     *  只有执行了上移函数，整个堆的最小堆性质才得以保证。

# 迭代器 #

PriorityBlockingQueue的迭代器也是弱一致性的，而且它弱得都有点离谱，因为在迭代器对象初始化的时候，就复制了一个新数组出来。也就是说，从初始化的时间节点之后，元素被从PriorityBlockingQueue中删除了迭代器也不管，新元素加入了PriorityBlockingQueue迭代器也不管。

public Iterator<E> iterator() { 
        return new Itr(toArray());
    }
    
    public Object[] toArray() { 
        final ReentrantLock lock = this.lock;
        lock.lock();
        try { 
            //拷贝queue数组的前面[0,size)部分元素，因为后面的元素都是null
            //新数组大小为size，元素都是浅拷贝
            return Arrays.copyOf(queue, size);
        } finally { 
            //解锁
            lock.unlock();
        }
    }

创建迭代器时，浅拷贝一个新数组给迭代器，颇有点“CopyOnWrite”的意思。

final class Itr implements Iterator<E> { 
            final Object[] array; // 创建迭代器时，队列内部数组的快照
            int cursor;           // 下一次next()即将返回的索引
            int lastRet;          // 上一次next()返回的索引，初始时或lastRet已经被删除时，它为-1
    
            Itr(Object[] array) { 
                lastRet = -1;
                this.array = array;
            }
    
            public boolean hasNext() { 
                return cursor < array.length;
            }
    
            public E next() { 
                if (cursor >= array.length)
                    throw new NoSuchElementException();
                lastRet = cursor;
                return (E)array[cursor++];
            }
    
            public void remove() { //用lastRet来支持remove函数
                if (lastRet < 0)
                    throw new IllegalStateException();
                removeEQ(array[lastRet]);//注意传入的是数组元素
                lastRet = -1;
            }
        }

在迭代器的remove函数中，不可以直接调用`removeAt`，而是要先去检查该数组元素是否还在PriorityBlockingQueue的内部数组中，如果还在，才能去删除它。毕竟迭代器的内部数组只是一个快照。

void removeEQ(Object o) { 
            final ReentrantLock lock = this.lock;
            lock.lock();
            try { 
                Object[] array = queue;
                for (int i = 0, n = size; i < n; i++) { 
                    if (o == array[i]) { //如果PriorityBlockingQueue内部数组中确实还有该元素
                        removeAt(i);//则删除它
                        break;
                    }
                }
            } finally { 
                lock.unlock();
            }
        }

# 总结 #

*  使用最小堆来实现优先级出队，但内部成员是数组，只是逻辑是个堆。
 *  两个元素`a b`，如果`a`的优先级更高，那么肯定`a.compareTo(b) < 0`。
 *  对于PriorityBlockingQueue来说，只需要时刻知道队列哪个元素最小，其他元素的顺序并不重要。所以使用堆这种数据结构再合适不过。
 *  元素比较方式有两种，`Comparable`和`Comparator`。但元素不一定支持`Comparable`，所以PriorityBlockingQueue的声明不能写成`PriorityBlockingQueue<E extends Comparable<E>>`这样的泛型自限定。
 *  堆的内部操作只有两种：冒泡上移 和 冒泡下沉。
 *  PriorityBlockingQueue不允许存储null元素。

[Link 1]: https://blog.csdn.net/anlian523/article/details/107727736
[JUC_]: https://blog.csdn.net/anlian523/article/details/106598910
[PriorityBlockingQueue]: https://blog.csdn.net/anlian523/article/details/107825751
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FubGlhbjUyMw_size_16_color_FFFFFF_t_70]: /images/20230208/38916a2af40a4f5e9c923c1603913587.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FubGlhbjUyMw_size_16_color_FFFFFF_t_70 1]: /images/20230208/6f9e9b4ba7964868b7f4fe9c7c839cae.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FubGlhbjUyMw_size_16_color_FFFFFF_t_70 2]: /images/20230208/342277eb3337489c86db7edd6962fc10.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FubGlhbjUyMw_size_16_color_FFFFFF_t_70 3]: /images/20230208/d48b3a5037844aa695d1c4ad19ad4c1a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FubGlhbjUyMw_size_16_color_FFFFFF_t_70 4]: /images/20230208/27690bd05475415e9634e64b078817b1.png