Java并发编程中的优先级队列——PriorityBlockingQueue源码解读

- 日理万妓 2023-09-27 19:20 7阅读 0赞

PriorityBlockingQueue是Java中实现了堆数据结构的线程安全的有界阻塞队列。它可以在多线程场景下安全地进行元素添加、删除和获取操作，而且可以根据元素的优先级进行排序。本篇博客将会深入解读PriorityBlockingQueue的源码实现。

### 一、PriorityBlockingQueue概述 ###

PriorityBlockingQueue类实现了BlockingQueue接口，它是一个线程安全的队列，继承自AbstractQueue类，而AbstractQueue类又实现了Queue接口。PriorityBlockingQueue是一个有界的队列，其容量可以在构造函数中进行指定，若不指定则默认大小为Integer.MAX\_VALUE。同时，PriorityBlockingQueue也支持根据元素的优先级进行排序，这是由于PriorityBlockingQueue内部实现了一个堆数据结构。

### 二、PriorityBlockingQueue源码解析 ###

1.  容器

PriorityBlockingQueue内部使用了一个Object类型的数组queue来存储元素，同时使用了一个int类型的变量size来记录元素的数量。下面是PriorityBlockingQueue类中的定义：

private transient Object[] queue;
    private transient int size;
    复制代码

1.  比较器

PriorityBlockingQueue可以根据元素的优先级进行排序，这是由于PriorityBlockingQueue内部维护了一个小根堆或大根堆。在构造函数中，我们可以选择使用元素自身的比较方式或是自定义比较器来进行元素的排序。若未指定比较器，则PriorityBlockingQueue将使用元素自身的比较方式进行排序。

private final Comparator<? super E> comparator;
    复制代码

1.  构造函数

PriorityBlockingQueue提供了多个构造函数，我们可以选择使用无参构造函数来创建一个默认容量为Integer.MAX\_VALUE的PriorityBlockingQueue，或是使用带有初始容量或自定义比较器的构造函数。下面是PriorityBlockingQueue类的两个构造函数：

public PriorityBlockingQueue() {
        this(DEFAULT_INITIAL_CAPACITY, null);
    }
    
    public PriorityBlockingQueue(int initialCapacity, Comparator<? super E> comparator) {
        if (initialCapacity < 1)
            throw new IllegalArgumentException();
        this.queue = new Object[initialCapacity];
        this.comparator = comparator;
    }
    复制代码

1.  添加元素

PriorityBlockingQueue中添加元素的方法为offer()方法，它会首先检查容量是否足够，如果容量不足则会进行扩容操作，扩容的方式是将原数组长度增加一半。接着，它会将新元素添加到队列的末尾，并通过siftUp()方法将元素上滤到合适的位置，以维护堆的性质。

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 (n == 0) { array[0] = e; } 
            else { siftUp(n, e, array, cmp); } 
            size = n + 1; notEmpty.signal(); 
        } finally { 
            lock.unlock(); 
        } 
        return true; 
    }
    复制代码

1.  获取元素

PriorityBlockingQueue中获取元素的方法为take()方法，它会首先检查队列是否为空，如果队列为空则会将当前线程阻塞，直到有元素被添加到队列中。接着，它会获取队列的头部元素，并通过siftDown()方法将队列的末尾元素移动到头部，以维护堆的性质。

public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); E result; try { while (size == 0) notEmpty.await(); result = extract(); } finally { lock.unlock(); } return result; }
    
    private E extract() { final Object[] array = queue; final E result = (E) array[0]; final int n = --size; final E x = (E) array[n]; array[n] = null; if (n != 0) siftDown(0, x, array, comparator); return result; }
    复制代码

1.  维护堆性质

PriorityBlockingQueue使用小根堆或大根堆来维护元素的优先级，这里我们以小根堆为例。小根堆的特点是父节点的值小于等于左右子节点的值，PriorityBlockingQueue中的堆是通过数组来实现的。当添加元素时，会将新元素添加到队列的末尾，并通过siftUp()方法将元素上滤到合适的位置，以维护堆的性质。当获取元素时，会获取队列的头部元素，并通过siftDown()方法将队列的末尾元素移动到头部，以维护堆的性质。下面是siftUp()和siftDown()方法的代码实现：

private static <T> void siftUp(int k, T x, Object[] array, Comparator<? super T> cmp) { if (cmp != null) siftUpUsingComparator(k, x, array, cmp); else siftUpComparable(k, x, array); }
    
    @SuppressWarnings("unchecked") 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; }
    
    @SuppressWarnings("unchecked") 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; }
    
    private static <T> void siftDown(int k, T x, Object[] array, Comparator<? super T> cmp) { if (cmp != null) siftDownUsingComparator(k, x, array, cmp); else siftDownComparable(k, x, array); }
    
    @SuppressWarnings("unchecked") private static <T> void siftDownUsingComparator(int k, T x, Object[] array, Comparator<? super T> cmp) { int half = size >>> 1; while (k < half) { int child = (k << 1) + 1; Object c = array[child]; int right = child + 1; if (right < size && 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; }
    
    @SuppressWarnings("unchecked") private static <T> void siftDownComparable(int k, T x, Object[] array) { Comparable<? super T> key = (Comparable<? super T>) x; int half = size >>> 1; while (k < half) { int child = (k << 1) + 1; Object c = array[child]; int right = child + 1; if (right < size && ((Comparable<? super T>) c).compareTo((T) array[right]) > 0) c = array[child = right]; if (key.compareTo((T) c) <= 0) break; array[k] = c; k = child; } array[k] = key; }
    复制代码

siftUp()方法和siftDown()方法都使用了siftUpUsingComparator()方法和siftDownUsingComparator()方法，它们是使用Comparator来实现堆的上滤和下滤的。当PriorityBlockingQueue没有指定Comparator时，会使用元素自身的Comparable来实现堆的上滤和下滤。

1.  总结

PriorityBlockingQueue是Java并发包中的一个线程安全的、支持优先级队列的类，它使用小根堆或大根堆来维护元素的优先级。PriorityBlockingQueue的内部实现使用了ReentrantLock和Condition来实现线程安全的操作，同时使用了数组来实现堆。PriorityBlockingQueue的核心方法包括添加元素的offer()方法，获取元素的take()方法，以及维护堆性质的siftUp()方法和siftDown()方法。PriorityBlockingQueue的使用方式与普通队列类似，但可以根据元素的优先级进行排序和处理。

作者：juer  
链接：https://juejin.cn/post/7229492628873674807