Java1.8-PriorityBlockingQueue源码学习（九）

一、概述

  有关优先级队列PriorityQueue的学习，我们前面已经学习过了，而PriorityBlockingQueue 同样是优先级队列，并且是一种无界的阻塞队列，该队列使用了和PriorityQueue相同的排序规则，在此基础上提供了阻塞的操作；虽然该队列在逻辑上是无容量限制的，但实际上是有最大容量限制的，超过最大容量有可能会导致OutOfMemoryError。我们再来简单看下优先级队列的一些特性：

  • 和普通的先进先出（FIFO）的队列不同，优先队列每次取出的元素都是队列中优先级最高的，PriorityBlockingQueue默认优先级最高的是元素最小的值，当然也可以按照我们指定的规则来自定义优先级；
  • 队列不允许null元素；
  • 默认情况下该队列根据元素的自然顺序进行排序，或者根据传入的比较器进行排序，但该队列不保证具有相同优先级的元素的排序；
  • 和PriorityQueue很像，PriorityBlockingQueue底层也是通过数组来实现的；默认情况下，PriorityBlockingQueue有默认的队列容量大小，但当队列满了之后，队列会自动扩容，直到最大容量；
  • 队列是基于小顶堆（或者最小堆）来实现的，也就是说小顶堆堆根结点是所有数据中最小的元素，并且堆中每个结点的值总是不大于其孩子结点的值。

用图来简单的对最小堆举个例子：

由于PriorityBlockingQueue对底层存储结构是数组，所以我们特地对元素标了编号，其实这也就是在数组中的下标值，最终存储格式可以看下图：

可以观察下父子之间的编号，也就是在数组中的下标值，可以发现：

如果节点的下标是i，那么i节点对应的子节点在数组中的位置分别是：2i + 1，2i + 2，同时i的父节点的位置是(i -1)/2，根据这个公式，我们可以很方便的找到每个节点的父亲节点和孩子节点。

其中，对堆最主要的操作有两种：上浮和下沉，但前提是堆是有序的，后续我们结合方法再来学习这两个主要操作。

二、PriorityBlockingQueue

接下来我们来学习PriorityBlockingQueue的源码实现，首先我们还是先来看下继承结构。

1. 继承结构
public class PriorityBlockingQueue<E> extends AbstractQueue<E>
    implements BlockingQueue<E>, java.io.Serializable {

可以看到，和其他的阻塞队列差不多，实现了BlockingQueue，然后继承了AbstractQueue，并且支持序列化。接下来来看下属性。

2. 属性
/**
 * 默认队列的容量
 */
private static final int DEFAULT_INITIAL_CAPACITY = 11;

/**
 * 队列的最大容量，超过该容量会导致OutOfMemoryError
 */
private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;

/**
 * 队列底层通过数组来实现
 */
private transient Object[] queue;

/**
 * 队列中元素数量
 */
private transient int size;

/**
 * 比较器
*/
private transient Comparator<? super E> comparator;

/**
 * Lock used for all public operations
 * 用于队列操作的可重入锁
 */
private final ReentrantLock lock;

/**
 * 队列不为空的Condition条件
 */
private final Condition notEmpty;

/**
 * 专为队列扩容时用的锁
 */
private transient volatile int allocationSpinLock;

/**
 * 普通优先级队列PriorityQueue，仅用于序列化
 */
private PriorityQueue<E> q;

可以看到，PriorityBlockingQueue 底层还是通过数组来实现，并且支持设置初始化容量和指定相应的比较器，但该容量只是初始容量，当队列满了之后，数组会进行扩容操作，直到达到最大容量限制，达到最大容量限制后，再添加就会抛出异常，而不是进行阻塞。

3. 构造方法

接下来我们来简单看下该队列的构造方法：

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();
    // 比较器
    this.comparator = comparator;
    // 数组初始化容量
    this.queue = new Object[initialCapacity];
}

前三个构造方法都比较简单，来看下最后一个构造方法：

public PriorityBlockingQueue(Collection<? extends E> c) {
    // 构建可重入锁及对应的Condition条件
    this.lock = new ReentrantLock();
    this.notEmpty = lock.newCondition();
    // 堆是否进行有序化操作，因为原先集合可能不是有序的
    boolean heapify = true; // true if not known to be in heap order
    // 堆是否要扫描null值
    boolean screen = true;  // true if must screen for nulls
    // 如果集合属于有序的SortedSet集合，获取SortedSet的比较器，并将heapify设置为false
    if (c instanceof SortedSet<?>) {
        SortedSet<? extends E> ss = (SortedSet<? extends E>) c;
        this.comparator = (Comparator<? super E>) ss.comparator();
        heapify = false;
    }
    // 否则，如果集合本身就是一个 PriorityBlockingQueue队列
    else if (c instanceof PriorityBlockingQueue<?>) {
        PriorityBlockingQueue<? extends E> pq =
            (PriorityBlockingQueue<? extends E>) c;
        // 获取集合的比较器，并将screen设置为false，因为PriorityBlockingQueue本身就不包含null，不用再次扫描
        this.comparator = (Comparator<? super E>) pq.comparator();
        screen = false;
        // 并且PriorityBlockingQueue本身就有序，也不用再次有序化
        if (pq.getClass() == PriorityBlockingQueue.class) // exact match
            heapify = false;
    }
    // 将集合转化成数组
    Object[] a = c.toArray();
    // 获取数组长度
    int n = a.length;
    // If c.toArray incorrectly doesn't return Object[], copy it.
    // 如果此时a的数据类型不是Object[]，拷贝一份
    if (a.getClass() != Object[].class)
        a = Arrays.copyOf(a, n, Object[].class);
    // 如果需要扫描null并且（要么元素就1个，要么元素可排序）
    if (screen && (n == 1 || this.comparator != null)) {
        for (int i = 0; i < n; ++i)
            if (a[i] == null)
                throw new NullPointerException();
    }
    // 数组赋值
    this.queue = a;
    // 容量赋值
    this.size = n;
    // 有序化操作
    if (heapify)
        heapify();
}

最后一个构造方法有点小复杂，需要先判断是否需要排序，是否需要扫描集合中值为null的元素，然后这两者操作完成之后，如果需要排序，再进行有序化操作：

private void heapify() {
    Object[] array = queue;
    int n = size;
    // 计算非叶子结点元素的最大位置，循环的开始条件（在最后一个非叶子节点处开始，直到根节点）
    int half = (n >>> 1) - 1;
    Comparator<? super E> cmp = comparator;
    if (cmp == null) {
        // 循环进行下沉操作
        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);
    }
}

所谓的有序化，其实就是循环下沉或者上浮操作，这里采用的是循环下沉操作，从非叶子结点元素的最大位置开始依次向上比较调整，直到根节点为止。

4. 方法
4.1 add/put/offer方法

首先，我们来看下入队的几个方法add/put/offer，由于最终都是通过offer方法来实现的，所以我们主要来看下offer方法。

public boolean add(E e) {
    return offer(e);
}

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

public boolean offer(E e, long timeout, TimeUnit unit) {
    return offer(e); // never need to block
}

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);
        // 队列中元素数量+1
        size = n + 1;
        // 元素添加完成，说明队列不为空了，唤醒在notEmpty条件上的线程
        notEmpty.signal();
    } finally {
        lock.unlock();
    }
    // 由于队列无容量限制，所以会一直返回true
    return true;
}

可以看到，添加元素的时候还是挺简单的：

  • 首先对元素进行非空校验，然后获取可重入锁；
  • 然后循环判断队列元素的数量是否大于数组容量，如果大于进行扩容操作；
  • 获取比较器，添加元素，然后进行上浮操作；
  • 最后唤醒notEmpty条件上的线程；

这里会调用扩容方法进行扩容操作，我们来看下这个方法。

4.2 tryGrow方法
private void tryGrow(Object[] array, int oldCap) {
    // 扩容前必须先释放锁，然后再重新获取
    lock.unlock(); // must release and then re-acquire main lock
    Object[] newArray = null;
    // 如果allocationSpinLock == 0，通过CAS方法设置为1，表示同一时刻，只有一个线程可以扩容操作
    if (allocationSpinLock == 0 &&
        UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset,
                                 0, 1)) {
        try {
            // 这里扩容的时候做了一点小处理，后续详细说
            int newCap = oldCap + ((oldCap < 64) ?
                                   (oldCap + 2) : // grow faster if small
                                   (oldCap >> 1));
            // 该模块是为了保证数组的容量不超过最大容量，如果超过了，提示OutOfMemoryError错误
            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，表示当前线程操作完成
            allocationSpinLock = 0;
        }
    }
    // 如果其他线程对队列进行了操作，则放弃扩容，并退出
    if (newArray == null) // back off if another thread is allocating
        Thread.yield();
    // 再次加锁 
    lock.lock();
    // 再次判断queue == array是否相等，相等的话，进行元素的复制
    if (newArray != null && queue == array) {
        queue = newArray;
        System.arraycopy(array, 0, newArray, 0, oldCap);
    }
}

  可以看到扩容方法中，使用了allocationSpinLock这个变量，在扩容前，该方法会释放锁，那么这时候就可能有其他的线程进行操作，为了保证扩容时的线程安全，所以添加了这个变量。在扩容的时候，会通过CAS操作把这个值设置为1，表示只有一个线程可以执行扩容操作，扩容操作完成后，会重置该值为1。

而要判断其他线程是否对队列元素进行了修改的话，是通过 queue == array来判断，如果其他线程对队列进行了修改，那么就会放弃扩容，因此在offer中会看到有一个 while 循环来判断是否真正需要扩容。

另外，针对扩容的容量这点代码，来简单说下：

int newCap = oldCap + ((oldCap < 64) ?
           (oldCap + 2) : // grow faster if small
           (oldCap >> 1));

如果原先队列的容量小于64，则扩充相当于1倍；如果容量大于64，则扩容原先容量的50%。也就是说，如果队列原先容量很小的话，那就多扩充些，如果容量大一点的话，那就少扩充些。

借用PriorityQueue 的说法，就是：Double size if small; else grow by 50%

4.3 siftUpComparable方法

然后这里还会调用到上浮的操作：没有比较器的siftUpComparable和有比较器的siftUpUsingComparator，因为两者操作相似，这里我们直接来看 siftUpComparable 方法即可：

private static <T> void siftUpComparable(int k, T x, Object[] array) {
    // 首先获取元素本身默认的比较器对象
    Comparable<? super T> key = (Comparable<? super T>) x;
    // 这里的k表示元素要插入的位置，第一次循环的时候表示的array[size]位置，也就是数组中最后一个元素 
    // 的下一个元素的位置，这里会循环操作
    while (k > 0) {
        // 找到该元素的父元素的位置，parent =  (thisNode-1)/2，这里使用了无符号右移
        int parent = (k - 1) >>> 1;
        // 获取父元素
        Object e = array[parent];
        // 如果当前元素大于等于父元素，直接退出循环，说明元素不需要上浮
        if (key.compareTo((T) e) >= 0)
            break;
        // 如果当前元素小于父元素，将父元素进行下沉到k所在的节点处
        array[k] = e;
        // 然后重置k，从父元素位置再次向上接着进行判断
        k = parent;
    }
    // 找到最终的位置k，将元素保存在这
    array[k] = key;
}

上浮的操作其实很简单，我们在前面学习 PriorityQueue 的时候就已经介绍过了，这里只需要看一下流程操作图即可：

4.4 poll方法

接下来，我们来看下出队的几个方法，首先来看下poll方法。poll方法本身没什么好说的，主要是该方法会调用出队的主要方法 dequeue，我们主要就是来看下这个方法：

public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return dequeue();
    } finally {
        lock.unlock();
    }
}

private E dequeue() {
    // 数组中元素数量-1
    int n = size - 1;
    // 如果数组中没有元素，直接返回null
    if (n < 0)
        return null;
    else {
        Object[] array = queue;
        // 获取数组的第一个元素，也就是要出队的元素
        E result = (E) array[0];
        // 获取数组的最后一个元素 array[size - 1]
        E x = (E) array[n];
        // 将数组的最后一个元素设置为null
        array[n] = null;
        Comparator<? super E> cmp = comparator;
        if (cmp == null)
            // 进行元素的下沉操作
            siftDownComparable(0, x, array, n);
        else
            siftDownUsingComparator(0, x, array, n, cmp);
        // 数组中数量 - 1
        size = n;
        // 返回数组原先的第一个元素
        return result;
    }
}

其实，出队的操作流程也很简单，获取第一个元素出队，然后将最后一个元素移动到第一个元素位置（并不是真的放到第一个位置），然后进行下沉操作即可。这里主要的方法是下沉方法siftDownComparable和siftDownUsingComparator，同样，我们来看下 siftDownComparable方法。

4.5 siftDownComparable方法
 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表示数组中原先 最后一个元素要放置的位置
        while (k < half) {
            // 计算k位置的左孩子的位置 => 2 * k + 1
            int child = (k << 1) + 1; // assume left child is least
            // 获取k位置的左孩子的值
            Object c = array[child];
            // 获取k位置的右孩子的位置
            int right = child + 1;
            // 获取左右孩子中值较小的值 
            if (right < n &&
                ((Comparable<? super T>) c).compareTo((T) array[right]) > 0)
                c = array[child = right];
            // 如果x 比左右孩子都小，不用下沉了，跳出循环
            if (key.compareTo((T) c) <= 0)
                break;
            // c节点下沉
            array[k] = c;
            // 重置k值
            k = child;
        }
        // 替换k位置处的值
        array[k] = key;
    }
}

由于PriorityQueue中也已经介绍过了，所以这里也就不多介绍了，来简单看下操作图即可：

4.6 poll(long, TimeUnit)方法

这个是超时的poll方法，表示获取并移除队列的队头元素，如果没有获取到对应的值，等待相应的超时时间。

public E poll(long timeout, TimeUnit unit) throws InterruptedException {
    // 获取超时时间，纳秒
    long nanos = unit.toNanos(timeout);
    final ReentrantLock lock = this.lock;
    // 可中断锁
    lock.lockInterruptibly();
    E result;
    try {
        // 如果没有获取到元素，并且没有超时，进行等待
        while ( (result = dequeue()) == null && nanos > 0)
            nanos = notEmpty.awaitNanos(nanos);
    } finally {
        lock.unlock();
    }
    return result;
}
4.7 take方法

take是出队的阻塞方法，表示获取并移除队列的队头元素，如果没有获取到元素，一直等待：

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;
}
4.8 peek方法

peek方法只获取队头元素，不移除元素：

public E peek() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        // 获取队头元素
        return (size == 0) ? null : (E) queue[0];
    } finally {
        lock.unlock();
    }
}
4.9 writeObject方法

从上面的操作我们可以看到，PriorityBlockingQueue中的属性private PriorityQueue<E> q;一直没有用到，因为该属性在该对象中只有一个用处就是用于序列化的，我们来看下序列化方法writeObject：

private void writeObject(java.io.ObjectOutputStream s)
    throws java.io.IOException {
    lock.lock();
    try {
        // avoid zero capacity argument
        q = new PriorityQueue<E>(Math.max(size, 1), comparator);
        q.addAll(this);
        s.defaultWriteObject();
    } finally {
        q = null;
        lock.unlock();
    }
}

之所以这样做，是为了避免过多的null值被序列化。

5. 总结

到这里一些主要的方法都学习过了，其他一些方法，大家有兴趣的可以自己了解下，或者参考原先学习PriorityQueue时的内容，而有关PriorityQueue原先学习的链接地址是：Java1.8-PriorityQueue源码解析。不过方法中最主要的还是队列的上浮和下沉两个操作。

本文参考除了包含官方文档之外，还包括：
Java 并发 --- 阻塞队列之PriorityBlockingQueuey源码分析 - csdn.net

另外本文画图工具：ProcessOn

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 194,242评论 5赞 459
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 81,769评论 2赞 371
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 141,484评论 0赞 319
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 52,133评论 1赞 263
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 61,007评论 4赞 355
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 46,080评论 1赞 272
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 36,496评论 3赞 381
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 35,190评论 0赞 253
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 39,464评论 1赞 290
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 34,549评论 2赞 309
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 36,330评论 1赞 326
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 32,205评论 3赞 312
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 37,567评论 3赞 298
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 28,889评论 0赞 17
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 30,160评论 1赞 250
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 41,475评论 2赞 341
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 40,650评论 2赞 335