2023-01-14  阅读(6)
原文作者: HelloWorld_EE 原文地址:https://blog.csdn.net/u010412719/category_6159934_2.html

《Java源码分析》:BlockingQueue之PriorityBlockingQueue

上面两篇博文分别介绍了BlockingQueue、ArrayBlockingQueue和LinkedBlockingQueue。这篇博文就来分析下PriorityBlockingQueue。

1、PriorityBlockingQueue的继承体系结构

        public class PriorityBlockingQueue<E> extends AbstractQueue<E>
            implements BlockingQueue<E>, java.io.Serializable

也是继承了AbstractQueue,实现了BlockingQueue、Serializable接口,与ArrayBlockingQueue、LinkedBlockingQueue的继承结构一模一样。

2、PriorityBlockingQueue的相关属性介绍

1)、默认容量

    private static final int DEFAULT_INITIAL_CAPACITY = 11;

2)、最大容量

     private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;

3)、这个数组代表的是一个平衡二叉堆,即queue[n]的子节点为queue[2n+1]和queue[2*(n+1)]

     private transient Object[] queue;

4)、优先队列中的元素个数

     private transient int size;

5)、比较器,如果为空,则为自然顺序

     private transient Comparator<? super E> comparator;

6)、锁

        private final ReentrantLock lock;
        //为空时,进行阻塞的Condition
        private final Condition notEmpty;
    
        /**
         * Spinlock for allocation, acquired via CAS.
         */
        private transient volatile int allocationSpinLock;

7)、优先队列:主要用于序列化,这是为了兼容之前的版本。只有在序列化和反序列化才非空

       private PriorityQueue<E> q;

3、PriorityBlockingQueue的构造函数介绍

PriorityBlockingQueue类中共有3个构造方法,如下

1)、创建一个默认容量大小为11的PriorityBlockingQueue对象。按照自然顺序进行排序

        public PriorityBlockingQueue() {
            this(DEFAULT_INITIAL_CAPACITY, null);
        }

2)、创建一个指定大小的PriorityBlockingQueue对象。按照自然顺序进行排序

        public PriorityBlockingQueue(int initialCapacity) {
            this(initialCapacity, null);
        }

3)、创建一个指定大小的PriorityBlockingQueue对象。按照给定的Comparator进行排序

        public PriorityBlockingQueue(int initialCapacity,
                                     Comparator<? super E> comparator) {
            if (initialCapacity < 1)
                throw new IllegalArgumentException();
            this.lock = new ReentrantLock();
            this.notEmpty = lock.newCondition();
            this.comparator = comparator;
            this.queue = new Object[initialCapacity];
        }

4、PriorityBlockingQueue中put方法介绍

函数功能:插入一个元素到优先队列中,由于此优先队列是无边界的,因此永远不可能阻塞。

        public void put(E e) {
            offer(e); // never need to block
        }

此put方法由于是无边界的,因此不可能阻塞,里面的里面实现直接调用的是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进行加一操作
                size = n + 1;
                notEmpty.signal();//唤醒正在等待的消费者线程
            } finally {
                lock.unlock();//释放锁
            }
            return true;
        }

offer方法功能:插入一个元素到优先队列中,由于此优先队列是无边界的,因此永远不可能返回false。

offer方法内部实现思路描述如下:

1、首先检查添加的元素是否为null,如果为null,则抛空指针异常。否则进行 2

2、加锁

3、检查数组是否已满,如果已满,则先进行扩容,否则进行 4.

4、然后将元素存储在数组的末尾即可。(由于优先队列底层借助于二叉堆来实现的,因此在插入之后,要进行一定的调整,使之维持二叉堆的特性。具体为:检查比较器是否为null,如果为null,则调用siftUpComparable方法将元素加入到数组末尾并进行相应的调整使之维持二叉堆的特性。如果比较器不为null。则根据指定的比较器进行调整)。

5、该计数器进行加一操作,由于成功的添加了一个元素队列不再为空,因此发射一个非空信号来唤醒正在等待的消费者。

6、释放锁

在offer中涉及到tryGrow函数进行扩容,siftUpComparable和siftUpUsingComparator这两个函数进行数组的插入和调整。下面一一进行介绍。

扩容函数:tryGrow(Object[] array, int oldCap)

函数功能:尝试的进行扩容一个及以上的位置空间(一般情况下为50%),放弃(允许重试)争用(但我们希望这种情况罕见)

        private void tryGrow(Object[] array, int oldCap) {
            lock.unlock(); // must release and then re-acquire main lock
            Object[] newArray = null;
            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) // back off if another thread is allocating
                Thread.yield();
            lock.lock();
            if (newArray != null && queue == array) {
                queue = newArray;
                System.arraycopy(array, 0, newArray, 0, oldCap);
            }
        }

siftUpComparable(int k, T x, Object[] array):此为没有指定构造器时向优先队列中添加元素并调整的函数

源码如下:

        /*
         * @param k the position to fill
         * @param x the item to insert
         * @param array the heap array
         */
        private static <T> void siftUpComparable(int k, T x, Object[] array) {
            Comparable<? super T> key = (Comparable<? super T>) x;
            while (k > 0) {
                int parent = (k - 1) >>> 1;//先求出父节点的位置
                Object e = array[parent];
                //如果父节点的值大于此值,则进行交换,然后继续判断。直至大于父节点的值。
                if (key.compareTo((T) e) >= 0)
                    break;
                array[k] = e;
                k = parent;
            }
            array[k] = key;
        }

函数功能:将元素x插入到数组array[k]位置上,并进行相应的调整,使之保持二叉堆的特性。调整的思想就是二叉堆添加元素的思想。

二叉堆的添加元素的思想如下

首先把要添加的元素加到数组的末尾,然后和它的父节点(位置为当前位置减去1再除以2取整(k-1)/2,比如第4个元素的父节点位置是1,第7个元素的父节点位置是3)比较,如果新元素比父节点元素大则交换这两个元素,然后再和新位置的父节点比较,直到它的父节点不再比它小,或者已经到达顶端,即第1的位置。

相信对二叉堆添加元素的思想理解了,理解上面的代码是没有问题的。

siftUpUsingComparator(int k, T x, Object[] array):此为指定了构造器时向优先队列中添加元素并调整的函数

函数功能:将元素x插入到array[k]处,利用指定的比较器,思想与siftUpComparable类似。

siftUpUsingComparator方法的源码如下:

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

以上就是关于PriorityBlockingQueue中put及其相关函数的一个分析,比较简单哈。主要的难点可能是一个二叉堆的调整。

4、PriorityBlockingQueue中take方法介绍

最后看下,take方法的实现.

源码如下:

        public E take() throws InterruptedException {
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();//加锁
            E result;
            try {
                while ( (result = dequeue()) == null) //如果出队列的元素为null,则说明此队列为空
                    notEmpty.await();
            } finally {
                lock.unlock();
            }
            return result;
        }

PriorityBlockingQueue类中put方法的思想主要是一个二叉堆添加元素。PriorityBlockingQueue类中的take方法的思想主要是关于一个二叉堆删除头结点位置的元素。

二叉堆删除元素调整过程如下

删除元素的过程类似,只不过添加元素是“向上冒”,而删除元素是“向下沉”:删除位置1的元素,把最后一个元素移到最前面,然后和它的两个子节点比较,如果两个子节点中较小的节点小于该节点,就将它们交换,直到两个子节点都比此顶点大。

理解了二叉堆删除元素的思想,理解下面dequeue方法中的代码就相当容易了。

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

dequeue方法的功能:取出优先队列中的第一个元素,并进行相应的调整,调整的方法下面会介绍。

siftDownComparable/siftDownUsingComparator

函数功能:将元素x插入到array[k]处,并进行相应的调整,使之保持为最小堆

源码如下:(注释相当详细,这里就不再介绍)

        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;           // loop while a non-leaf
                //在数组中,位置k的节点的子节点的下标肯定小于half
                while (k < half) {
                    //左子节点的下标
                    int child = (k << 1) + 1; // assume left child is least
                    Object c = array[child];
                    //右子节点的下标
                    int right = child + 1;
                    //先比较左右子节点谁小,始终保持c为两子节点中最小的。
                    if (right < n &&
                        ((Comparable<? super T>) c).compareTo((T) array[right]) > 0)
                        c = array[child = right];
                    //如果key值大于子节点的值,则向后沉。否则不变
                    if (key.compareTo((T) c) <= 0)
                        break;
                    array[k] = c;
                    k = child;
                }
                array[k] = key;
            }
        }
    
        //函数功能:将元素x插入到array[k]处,并自己指定的比较器进行相应的调整,使之保持为最小堆
    
        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;
            }
        }

小结

以上就是关于PriorityBlockingQueue的一个简单介绍,在了解了二叉堆的一个添加元素和删除元素的相关思想之后,理解这个内部实现还是相当简单的哈。

关于PriorityBlockingQueue类我们只需要记住的是底层是基于一个二叉堆的实现。关于二叉堆的添加元素和删除的思想也是我们需要掌握的哈。


Java 面试宝典是大明哥全力打造的 Java 精品面试题,它是一份靠谱、强大、详细、经典的 Java 后端面试宝典。它不仅仅只是一道道面试题,而是一套完整的 Java 知识体系,一套你 Java 知识点的扫盲贴。

它的内容包括:

  • 大厂真题:Java 面试宝典里面的题目都是最近几年的高频的大厂面试真题。
  • 原创内容:Java 面试宝典内容全部都是大明哥原创,内容全面且通俗易懂,回答部分可以直接作为面试回答内容。
  • 持续更新:一次购买,永久有效。大明哥会持续更新 3+ 年,累计更新 1000+,宝典会不断迭代更新,保证最新、最全面。
  • 覆盖全面:本宝典累计更新 1000+,从 Java 入门到 Java 架构的高频面试题,实现 360° 全覆盖。
  • 不止面试:内容包含面试题解析、内容详解、知识扩展,它不仅仅只是一份面试题,更是一套完整的 Java 知识体系。
  • 宝典详情:https://www.yuque.com/chenssy/sike-java/xvlo920axlp7sf4k
  • 宝典总览:https://www.yuque.com/chenssy/sike-java/yogsehzntzgp4ly1
  • 宝典进展:https://www.yuque.com/chenssy/sike-java/en9ned7loo47z5aw

目前 Java 面试宝典累计更新 400+ 道,总字数 42w+。大明哥还在持续更新中,下图是大明哥在 2024-12 月份的更新情况:

想了解详情的小伙伴,扫描下面二维码加大明哥微信【daming091】咨询

同时,大明哥也整理一套目前市面最常见的热点面试题。微信搜[大明哥聊 Java]或扫描下方二维码关注大明哥的原创公众号[大明哥聊 Java] ,回复【面试题】 即可免费领取。

阅读全文