JAVA源码学习之集合-PriorityBlockingQueue

类的描述

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

使用与类PriorityQueue相同的排序规则（堆排序）并提供阻塞检索操作的无界阻塞队列。虽然此队列在逻辑上是无界的，但由于资源耗尽（导致OutOfMemoryError），尝试添加可能会失败。此类不允许空元素。依赖于自然顺序的优先级队列也不允许插入不可比较的对象（这样做会导致ClassCastException）。

这个类和它的迭代器实现所有的可选方法的Collection和Iterator接口。方法iterator()提供的迭代器不能保证遍历的priorityblockingqueue元素在任何特定的顺序。如果你需要有序遍历，考虑使用Arrays.sort(pq.toArray())。同时，drainTo方法可以用来去除部分或全部在另一个集合中的元素，将它们的优先顺序。

此类上的操作不能保证元素具有同等优先级的顺序。如果需要强制执行排序，可以定义自定义类或比较器，这些类或比较器使用辅助键断开主优先级值中的关联。例如，这里有一个类，它将先进先出的连接中断应用于可比较的元素。要使用它，您需要插入一个新的FIFOEntry（anEntry），而不是一个普通的entry对象

class FIFOEntry<E extends Comparable<? super E>>implements Comparable<FIFOEntry<E>> {static final AtomicLong seq = new AtomicLong(0);final long seqNum;final E entry;public FIFOEntry(E entry) {seqNum = seq.getAndIncrement();this.entry = entry;}public E getEntry() { return entry; }public int compareTo(FIFOEntry<E> other) {int res = entry.compareTo(other.entry);if (res == 0 && other.entry != this.entry)res = (seqNum < other.seqNum ? -1 : 1);return res;}}

常量、变量、静态内部类

/*** Default array capacity.*/private static final int DEFAULT_INITIAL_CAPACITY = 11;/*** 要分配的最大数组大小。有些虚拟机在数组中保留一些头字。尝试分配较大的数组可能会导致 * OutOfMemoryError：请求的数组大小超过VM限制*/private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;// 队列中实际存储元素的对像，表示为平衡二进制堆：节点[n]的两个节点是节点[2*n+1]和节点[2*（n+1）]。private transient Object[] queue;// 队列的元素数量private transient int size;//比较器，如果优先级队列使用元素的自然顺序，则为null。private transient Comparator<? super E> comparator;//private final ReentrantLock lock;/*** Condition for blocking when empty*/private final Condition notEmpty;// 用于分配的自旋锁，通过CAS获取, (不是很明白，继续往下看)private transient volatile int allocationSpinLock;//仅用于序列化的普通PriorityQueue，用于保持与此类早期版本的兼容性。仅在序列化/反序列化期间为非nullprivate PriorityQueue<E> q;

猜想：

 通过 Object[] queue 来存储元素，然后排序算法和PriorityQueue是一样的使用的的堆排序

通过  ReentrantLock lock 来实现线程安全

虽然是无界的，但是又一个最大容量值 Integer.MAX_VALUE-8,  默认容量是11

 // 用于分配的自旋锁，通过CAS获取, (不是很明白，继续往下看)
    private transient volatile int allocationSpinLock; 这个不知道是用于什么，往下看

构造方法

//默认构造， 数组长度为 DEFAULT_INITIAL_CAPACITY， 排序方法未指定
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) {this.lock = new ReentrantLock();this.notEmpty = lock.newCondition();boolean heapify = true; // 如果不知道排序， 为trueboolean screen = true;  // 如果必须筛选空值，则为trueif (c instanceof SortedSet<?>) { //如果集合C属于SortSet, 本身是有排序的SortedSet<? extends E> ss = (SortedSet<? extends E>) c;//当前的比较方法是c的比较方法this.comparator = (Comparator<? super E>) ss.comparator();heapify = false;}else if (c instanceof PriorityBlockingQueue<?>) { //如果集合C属于PriorityBlockingQueue, PriorityBlockingQueue<? extends E> pq =(PriorityBlockingQueue<? extends E>) c;//比较方法是c的比较方法this.comparator = (Comparator<? super E>) pq.comparator();//因为PriorityBlockingQueue不允许元素为空，所以 screen = false;screen = false;if (pq.getClass() == PriorityBlockingQueue.class) // exact matchheapify = false;}Object[] a = c.toArray();int n = a.length;if (c.getClass() != java.util.ArrayList.class)a = Arrays.copyOf(a, n, Object[].class);if (screen && (n == 1 || this.comparator != 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(); //使用堆排序}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);}}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-leafwhile (k < half) {int child = (k << 1) + 1; // assume left child is leastObject c = array[child];int right = child + 1;if (right < n &&((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;}}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;}}

关于构造器有疑问的地方请看这篇文档

(1条消息) JDK8 PriorityBlockingQueue(Collection＜? extends E＞ c)构造器 源码解析_anlian523的博客-CSDN博客https://blog.csdn.net/anlian523/article/details/107825751

常用方法

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)) //如果size > = length。 扩容tryGrow(array, cap);try {Comparator<? super E> cmp = comparator;if (cmp == null)siftUpComparable(n, e, array); //元素本身的排序算法执行堆排序elsesiftUpUsingComparator(n, e, array, cmp); // 指定比较方法的堆排序size = n + 1;notEmpty.signal();} finally {lock.unlock();}return true;}private void tryGrow(Object[] array, int oldCap) {lock.unlock(); // must release and then re-acquire main lockObject[] newArray = null;//使用volatile和UNSAFE来保证线程安全，相当于，AQS独占锁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 overflowint 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();lock.lock();if (newArray != null && queue == array) {queue = newArray;System.arraycopy(array, 0, newArray, 0, oldCap);}}

add

public boolean add(E e) {return offer(e);}//AbstractQueue的addAll， 最终使用的add方法public boolean addAll(Collection<? extends E> c) {if (c == null)throw new NullPointerException();if (c == this)throw new IllegalArgumentException();boolean modified = false;for (E e : c)if (add(e))modified = true;return modified;}

poll

 public E poll() {final ReentrantLock lock = this.lock;lock.lock();try {return dequeue();} finally {lock.unlock();}}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); 。//堆排序elsesiftDownUsingComparator(0, x, array, n, cmp); //堆排序size = n;return result;}}//等待一定时间的pollpublic 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); //等待nanos时间} finally {lock.unlock();}return result;}

task

  public E take() throws InterruptedException {final ReentrantLock lock = this.lock;lock.lockInterruptibly();E result;try {while ( (result = dequeue()) == null)notEmpty.await(); //notEmpty进入等待对立，一直等到他被唤醒，或者lock被中断} finally {lock.unlock();}return result;}

remove

    //AbstractQueue的removepublic E remove() {E x = poll(); //调用pollif (x != null)return x;elsethrow new NoSuchElementException();}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) //移除最后last一条元素array[i] = null;else {E moved = (E) array[n];array[n] = null; //最后元素置为nullComparator<? super E> cmp = comparator;//将moved替换到i的位置，进行下沉if (cmp == null)siftDownComparable(i, moved, array, n);  elsesiftDownUsingComparator(i, moved, array, n, cmp);if (array[i] == moved) { //这种请看说明根本没有下沉，如果下沉肯定会比当前值小if (cmp == null)siftUpComparable(i, moved, array);elsesiftUpUsingComparator(i, moved, array, cmp);}}size = n;}

peek

 public E peek() {final ReentrantLock lock = this.lock;lock.lock(); //防止其他线程进行了操作导致当前队列头元素变了try {return (size == 0) ? null : (E) queue[0];} finally {lock.unlock();}}

element

 public E element() {E x = peek();if (x != null)return x;elsethrow new NoSuchElementException();}

结论

1. 该队列是通过堆排序算法来保证优先级的，如果未指定排序方法，用元素自身的排序进行比较，

2. 通过 ReentrantLock 来进行实现阻塞

3. 元素不允许为空

关于其中的堆排序，上移和下沉，请参考下面相关文章

(1条消息) JUC集合类 PriorityBlockingQueue源码解析 JDK8_anlian523的博客-CSDN博客

(1条消息) 从小顶堆到堆排序——超详细图解——Python3实现_anlian523的博客-CSDN博客_堆排序小顶堆

上一章