java中PriorityBlockingQueue的入队知识点总结

下面是对 "java中PriorityBlockingQueue的入队知识点总结" 的详细讲解。

PriorityBlockingQueue的概述

PriorityBlockingQueue 是 java.util.concurrent 包中的一个类，它是一个具有优先级的无界阻塞队列，可以用来实现生产者-消费者模式中的队列。

PriorityBlockingQueue 使用数组实现，内部采用堆结构排序，可以根据元素自然排序或者通过构造函数指定的 Comparator 排序。它有以下主要特点：

内部采用堆结构实现排序，具有很好的性能。
可以通过两种方式排序：自然排序和指定比较器排序。
内部使用 ReentrantLock（可重入锁）来保证线程安全。

PriorityBlockingQueue的入队操作

PriorityBlockingQueue 的入队方法是 offer() 或 put()，两种方法的区别在于，当队列已满时，offer() 方法会返回 false，而 put() 方法则会阻塞。

offer()方法

offer() 方法可以向队尾插入一个元素，如果队列已满，则返回 false。

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();
    } finally {
        lock.unlock();
    }
    return true;
}

tryGrow() 方法的作用是扩容。如果队列的长度已经等于容量，说明队列已满，此时需要扩容以容纳新的元素。

扩容的方法是通过创建一个新数组 newArray 来实现的，然后将原数组 queue 的元素复制到新数组中，最后将 newArray 赋值给 queue。

private void tryGrow(Object[] array, int oldCap) {
    lock.unlock();
    Object[] newArray = null;
    if (allocationSpinLock == 0 &&
        UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset, 0, 1)) {
        try {
            int newCap = oldCap + ((oldCap < 64) ?
                                   (oldCap + 2) :
                                   (oldCap >> 1));
            if (newCap >= MAX_ARRAY_SIZE)
                throw new OutOfMemoryError();
            int newAlloc = newCap - oldCap;
            newArray = new Object[newCap];
            System.arraycopy(array, 0, newArray, 0, Math.min(array.length, newAlloc));
        } finally {
            allocationSpinLock = 0;
        }
    }
    if (newArray == null)
        Thread.yield();
    lock.lock();
    if (queue == array) {
        queue = newArray;
        System.arraycopy(array, 0, newArray, 0, oldCap);
    }
}

while 循环的作用是，如果队列已满，就不断尝试扩容，直到新增元素可以被添加到队列中。

如果队列已满，但是扩容失败（例如在高并发的情况下，多个线程同时扩容），此时会让出 CPU 时间，等待其他线程释放锁，再重新尝试添加元素。

然后，调用 siftUpComparable() 或 siftUpUsingComparator() 方法来对新增元素进行排序。如果没有指定比较器（即使用默认的自然排序），则调用 siftUpComparable() 方法。

private void siftUpComparable(int k, E x, Object[] array) {
    Comparable<? super E> key = (Comparable<? super E>) x;
    while (k > 0) {
        int parent = (k - 1) >>> 1;
        Object e = array[parent];
        if (key.compareTo((E) e) >= 0)
            break;
        array[k] = e;
        k = parent;
    }
    array[k] = key;
}

while 循环的条件是：如果新增元素比其父节点更小，则将父节点下移，继续比较，直到满足堆的性质。

最后，更新队列的元素数量，发出 notEmpty 的信号，表示队列已非空，并返回 true 。

put()方法

put() 方法将一个元素插入队列中，如果队列已满，则一直等待，直到队列有空闲位置。

public void put(E e) throws InterruptedException {
    offer(e); // 向队列尾部插入一个元素
    if (Thread.interrupted()) // 判断线程是否被打断，如果被打断，就抛出异常
        throw new InterruptedException();
}

在 put() 方法中，调用了 offer() 方法向队列尾部插入元素。如果队列已满，就不断尝试扩容或者等待，直到新增元素可以被添加到队列中。

最后，如果线程被中断，就抛出 InterruptedException 异常。

示例说明

下面以两个示例分别说明 PriorityBlockingQueue 的入队操作。

示例一

PriorityBlockingQueue<Integer> queue = new PriorityBlockingQueue<>(16);

for (int i = 0; i < 20; i++) {
    queue.offer(i);
    System.out.println("add " + i + ", size: " + queue.size());
}

结果输出：

add 0, size: 1
add 1, size: 2
add 2, size: 3
add 3, size: 4
add 4, size: 5
add 5, size: 6
add 6, size: 7
add 7, size: 8
add 8, size: 9
add 9, size: 10
add 10, size: 11
add 11, size: 12
add 12, size: 13
add 13, size: 14
add 14, size: 15
add 15, size: 16
add 16, size: 17
add 17, size: 18
add 18, size: 19
add 19, size: 20

在 PriorityBlockingQueue 的入队操作中，如果队列已满，就会进行扩容。在这个示例中，队列的初始容量是 16，当添加第 8 个元素时，队列已满，触发了扩容操作。从输出结果可以看出，每插入一个元素，队列的大小就会增加 1 。

示例二

PriorityBlockingQueue<Integer> queue = new PriorityBlockingQueue<>(16);

for (int i = 0; i < 20; i++) {
    queue.put(i);
    System.out.println("add " + i + ", size: " + queue.size());
}

System.out.println("--------------------");

for (int i = 20; i < 30; i++) {
    queue.put(i);
    System.out.println("add " + i + ", size: " + queue.size());
}