1、PriorityBlockingQueue 介绍
PriorityBlockingQueue 是一个优先级队列，它不满足队列的先进先出特点；
PriorityBlockingQueue 会对队列的数据进行排序，排序规则是数据的优先级；
PriorityBlockingQueue是基于二叉堆来实现优先级的，底层采用数组来实现二叉堆；
虽然PriorityBlockingQueue 底层是数组，但该数组是可以扩容的，理论上相当于一个无界
链表，所以在 PriorityBlockingQueue 中生产者线程是不会阻塞的。
2、PriorityBlockingQueue 核心属性介绍
PriorityBlockingQueue 核心属性和构造方法如下：
public class PriorityBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
private static final long serialVersionUID = 5595510919245408276L;
/**
* Default array capacity.
* 数组的初始长度
*/
private static final int DEFAULT_INITIAL_CAPACITY = 11;
/**
*
* 数组的最大长度
* todo 注意：
* 这里之所以 减8 ，则是为了适配各个版本的虚拟机
*/
private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
/**
*
* 存储数据的数组，基于这个数组实现的二叉堆
*/
private transient Object[] queue;
/**
*
* 优先级队列的容量，即数组queue的长度
*/
private transient int size;
/**
*
* 比较器，比较优先级
* todo 注意：
* 若队列存放的数据是类（引用）类型的数据，则该类需要实现比较接口Comparable
* 基于 Comparable做对象之间的比较
*/
private transient Comparator<? super E> comparator;
/**
* 锁，实现阻塞队列的lock锁
*/
private final ReentrantLock lock;
/**
* 关联Lock的Condition
*/
private final Condition notEmpty;
/**
* 因为 PriorityBlockingQueue 是基于二叉堆实现的，而这里的二叉堆是基于数组实现的；
* 数组长度是固定的，如果需要扩容则需要构建一个新数组，如果在锁lock范围内，构建数组的过程中需要迁移数据，
* 此时效率会很低；PriorityBlockingQueue 做了一个事情，它在扩容过程中是不会加锁的；
* PriorityBlockingQueue 在扩容过程中会先释放锁，基于属性 allocationSpinLock 做标记 来避免出现并发
* 扩容的问题。
*/
private transient volatile int allocationSpinLock;
/**
*
* 阻塞优先级队列的原理，用到了普通优先级队列的特性（堆）
*/
private PriorityQueue<E> q;
/**
* 默认构造函数
* 不指定优先级的比较规则，PriorityBlockingQueue保存的数据必须实现接口Comparable
*/
public PriorityBlockingQueue() {
this(DEFAULT_INITIAL_CAPACITY, null);
}
/**
* 带容量的构造函数
* 不指定优先级的比较规则，PriorityBlockingQueue保存的数据必须实现接口Comparable
*/
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];
}
}注意：若 new 创建PriorityBlockingQueue对象时若不指定比较规则（即：Comparator不
指定），则 PriorityBlockingQueue 保存的数据必须是可比较的对象，即
PriorityBlockingQueue存储数据的类型必须实现接口Comparable
3、使用示例
PriorityBlockingQueue使用也很简单，它常用的方法也是在接口BlockingQueue中定义的
那几个存储数据和取数据的方法。
示例代码如下：
public class PriorityBlockingQueueDemo01 {
public static void main(String[] args) throws InterruptedException {
//向PriorityBlockingQueue 存放基础类型数据
PriorityBlockingQueue queue = new PriorityBlockingQueue();
queue.add(3);
queue.add(2);
queue.add(1);
queue.offer(0);
//添加数据，当队列满了之后会自动扩容，所以添加数据不会因为队列满了后线程挂起等待
queue.offer(4,5, TimeUnit.SECONDS);
queue.put(5);
//取数据
System.out.println(queue.remove());//取的第一个数据是1，
//若队列为空，则返回null
System.out.println(queue.poll());
//当队列为空时，消费者线程会阻塞等待5s，5s后若队列还没有数据则返回null
System.out.println(queue.poll(5,TimeUnit.SECONDS));
//若队列为空则一直阻塞
System.out.println(queue.take());
//PriorityBlockingQueue 保存引用类型，1、可比较的引用类型
PriorityBlockingQueue<Apple> que2 = new PriorityBlockingQueue<Apple>();
que2.add(new Apple());
//PriorityBlockingQueue 保存引用类型，1、不可比较的引用类型
PriorityBlockingQueue<Dog> que3 = new PriorityBlockingQueue<Dog>();
que3.add(new Dog());//抛出异常：Dog cannot be cast to java.lang.Comparable
}
}4、常用方法解析
4.1、offer(E e) 方法
在PriorityBlockingQueue中添加数据时add、put方法里面直接调用offer方法，还有
方法 offer(E e, long timeout, TimeUnit unit) 因为 PriorityBlockingQueue 存储数据的数组
是可以自动扩容的，不存在因为队列已满数据放不进而生产者线程挂起等待的情况，所以
在 offer(E e, long timeout, TimeUnit unit) 也是直接调用的方法offer(E e)，所以这里直接
看下方法 offer(E e) 就行了；
offer(E e) 方法代码如下：
/*
* 添加数据
*/
public boolean offer(E e) {
if (e == null)
throw new NullPointerException();
final ReentrantLock lock = this.lock;
lock.lock();
/**
* n: 队列数据个数
* cap: 队列数组长度
*/
int n, cap;
Object[] array;
//若队列数据个数大于队列数组长度，则需要扩容
while ((n = size) >= (cap = (array = queue).length))
//数组动态扩容
//todo 注意：并发环境中，扩容不能并发执行，若有2个线程同时执行到了扩容这一步，若已经有第一个线程正在扩容，则第二个线程
// 不会再去扩容，可能多次执行while、多次进入到方法 tryGrow，但仍然需要等待前面的线程扩容操作结束
// （虽然在该方法中加了锁lock，但在扩容时当前线程释放了锁）
tryGrow(array, cap);
try {
//比较器
Comparator<? super E> cmp = comparator;
//比较数据大小，存储数据，然后判断是否需要进行上移操作，保证平衡位置
if (cmp == null)
//比较器为null
siftUpComparable(n, e, array);
else
//比较器不为null
siftUpUsingComparator(n, e, array, cmp);
size = n + 1;
notEmpty.signal();
} finally {
lock.unlock();
}
return true;
}4.2、tryGrow(Object[] array, int oldCap) 方法
该方法功能是数组动态扩容；
注意：在并发环境中，只能由一个线程能够进行数组扩容，但数组的扩容并不是通过
锁来保证原子性，而是基于CAS+属性allocationSpinLockOffset 来保证扩容操作
的原子性。
tryGrow 方法代码如下：
/**
* 数组扩容
* todo 注意并发下的处里
* 为了提高性能，扩容会先释放当前线程持有的锁，并通过属性allocationSpinLockOffset
* 来保证只有一个线程能进行扩容操作
*/
private void tryGrow(Object[] array, int oldCap) {
/**
* todo : 当前线程释放锁
*/
lock.unlock(); // must release and then re-acquire main lock
//声明新的数组
Object[] newArray = null;
//allocationSpinLock是一个标记，等于0，表示当前没有线程正在扩容，当前线程可以进行扩容
if (allocationSpinLock == 0 &&
//基于CAS方式将 allocationSpinLock 值由0修改为1，表示当前线程正在扩容
UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset,
0, 1)) {
try {
//计算新数组长度，若当前数组长度小于64，则每次扩容原来数组长度的2倍
//若原来数组长度大于等于64，则每次扩容到原来数组长度的1.5倍
int newCap = oldCap + ((oldCap < 64) ?
//这里加2，1）是为了加快扩容数组长度；2）反射时，若有人把数组长度设置为0，若不加2则会出错
(oldCap + 2) : // grow faster if small
(oldCap >> 1));//除以2
//判断数组长度是否达到最大
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;
}
//判断当前数组queue是否被其他线程改变（确保没有并发扩容的问题），若没有被改变且新数组长度大于queue长度，则创建新数组
if (newCap > oldCap && queue == array)
newArray = new Object[newCap];
} finally {
//扩容成功把 allocationSpinLock 修改为0
//为了下一次扩容
allocationSpinLock = 0;
}
}
//有线程正在扩容
if (newArray == null) // back off if another thread is allocating
//退出线程执行，让出CPU时间片，等待一会
Thread.yield();
/**
* 获取锁
* 这里获取锁的线程可能是执行扩容操作的线程，也可能是上边执行yield 的线程，
*/
lock.lock();
//若数组没有被修改，表示当前线程是执行扩容操作的线程，则把新数组赋值给queue，并完成数据的迁移
if (newArray != null && queue == array) {
queue = newArray;
System.arraycopy(array, 0, newArray, 0, oldCap);
}
}
4.3、siftUpComparable(int k, T x, Object[] array) 方法
该方法功能是将数据保存到数组queue中，并通过循环比较将数据x保存到合适的位置，
以保证二叉堆的结构不被破坏，使用默认的比较器来进行数据的比较。
siftUpComparable方法代码如下：
/*
* @param k the position to fill 当前元素个数（即数据x要存储的下标位置）
* @param x the item to insert 数据
* @param array the heap array 堆数组
*
* 将数据保存到数组中，并保证二叉堆结构
*/
private static <T> void siftUpComparable(int k, T x, Object[] array) {
//将插入的元素强转为 Comparable(即 x 类（引用类型）必须实现 Comparable 接口)
Comparable<? super T> key = (Comparable<? super T>) x;
//k>0表示数组 array 中有数据
while (k > 0) {
//找到当前将要存入数据x的父节点（x将被保存到k的位置）位置
int parent = (k - 1) >>> 1;
//获取父节点数据
Object e = array[parent];
//比较子节点数据与其父节点数据的大小，若子节点数据大于父节点数据，则直接结束（最小堆）
//否则，交换子节点与父节点位置的数据，并从当前父节点位置向上比较，这个操作称为“堆节点上浮”
if (key.compareTo((T) e) >= 0)
break;
array[k] = e;
//从当前父节点位置进行下一次比较判断
k = parent;
}
//k==0表示当前 数组array没有数据，可以把x存放到下标为0的位置
array[k] = key;
}
4.4、siftUpUsingComparator(int k, T x, Object[] array,Comparator<? super T> cmp) 方法
该方法功能是将数据保存到数组queue中，并通过循环比较将数据x保存到合适的位置，
以保证二叉堆的结构不被破坏，使用实例化PriorityBlockingQueue时指定的比较器。
siftUpUsingComparator 方法代码如下：
/**
* 自定义比较器的数据存储
* 节点上浮
*
* @param k 当前元素个数，也是数据x将要保存的位置
* @param x 要保存的数据
* @param array 数组
* @param cmp 自定义比较器
* @param <T>
*/
private static <T> void siftUpUsingComparator(int k, T x, Object[] array,
Comparator<? super T> cmp) {
//当前数组中有数据
while (k > 0) {
//计算位置k的父节点位置
int parent = (k - 1) >>> 1;
//获取父节点数据
Object e = array[parent];
//最小堆，父节点数据小于子节点数据
//比较位置k的数据x与父节点数据e的大小，若子节点x比父节点e大，则直接结束
//否则交换父节点与子节点的数据，并从父节点位置开始进行下一次循环（网上继续比较父子节点数据）比较，直到根节点
if (cmp.compare(x, (T) e) >= 0)
break;
array[k] = e;
k = parent;
}
//k==0表示数组中无数据，直接把x保存到0的位置
array[k] = x;
}4.5、poll() 方法
该方法功能是从队列中取数据
注意：PriorityBlockingQueue中其他取数据的功能像 remove()、take()的实现与前边的
ArrayBlockingQueue和LinkedBlockingQueue的实现一样，内部都是调用poll()方法
来进行取数据，这里就不看了
poll 方法代码如下：
4.6、poll(long timeout, TimeUnit unit)
该方法是带有超时时间的取数据，若等待超过超时时间队列还位空，则返回null；
该方法允许线程中断，当线程被中断时会抛出异常，并退出。
4.7、dequeue() 方法
该方法是真正取数据的方法，在取数据后并保证二叉堆结构不被破坏。
dequeue 方法代码如下：
/**
*
* 取数据，但取数据后要保证二叉堆结构不会被破坏
*/
private E dequeue() {
//数组中最后一个数据的下标
int n = size - 1;
//队列中无数据
if (n < 0)
return null;
else {
Object[] array = queue;
//取二叉堆的根节点数据，即最小堆的最小的数据
E result = (E) array[0];
//获取二叉堆中最下层最右侧的数据（即数组最后一个数据），然后把最后一个数据x虚拟的放到根节点位置（即k=0的位置）
//然后再调用 siftDownComparable 或 siftDownUsingComparator 把数据x下沉到合适的位置
E x = (E) array[n];
//删除下标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;
}
}4.8、siftDownComparable(int k, T x, Object[] array,int n) 方法
该方法功能是取堆根节点（即queue中第一个数据）数据后把数组queue最后一个数 x “虚拟” 的（假设）放到堆的跟几点（即queue[0]的位置），然后通过循环遍历比较每个节点及其
子节点数据的大小，将数据x 下沉到合适的位置，以保证二叉堆的结构不被破坏。
siftDownComparable 方法代码如下：
/*
* 循环开始时，是把数据x虚拟的放在根节点（即下标是0的位置），然后从根节点开始与左右子节点比较，遵循最小堆的原则，找到数据x的合适位置
* @param k the position to fill 当前根节点位置（默认为0）
* @param x the item to insert 堆中（数组中）最后一个数据元素
* @param array the heap array 堆数组
* @param n heap size 数组中元素的个数
*/
private static <T> void siftDownComparable(int k, T x, Object[] array,
int n) {
if (n > 0) {//堆（数组）中还有有数据
//将要上浮的数据将至转换成 Comparable（或者拿到最后一个数据的比较器）
Comparable<? super T> key = (Comparable<? super T>)x;
//n除以2，因为二叉堆是一个满二叉树，所以只需要拿k的左右子节点中的小的数据进行比较，
// 所以这里只需要比较一半的数据
int half = n >>> 1; // loop while a non-leaf
while (k < half) {
//计算k左子节点的位置
int child = (k << 1) + 1; // assume left child is least
Object c = array[child];
//k右子节点位置
int right = child + 1;
//得到k左右子节点中较小的数，然后与根节点的数进x行比较
if (right < n &&
((Comparable<? super T>) c).compareTo((T) array[right]) > 0)
c = array[child = right];
//若数据x小于根节点的左右子树节点数据，则表示最大位置的数据x可以作为根节点，直接退出
//否则，将根节点的左右子节点较小的数据作为根节点
if (key.compareTo((T) c) <= 0)
break;
//更新位置k的值
array[k] = c;
//以较小的子节点作为根节点进行下一次循环比较
//最后k就是x的位置
k = child;
}
//循环结束后，所有数据已经满足二叉堆的特点，只差位置k的数据为null，k就是数据x的位置
array[k] = key;
}
}4.9、siftDownUsingComparator(int k, T x, Object[] array,int n,Comparator<? super T> cmp) 方法
该方法功能是取堆根节点（即queue中第一个数据）数据后把数组queue最后一个数 x “虚拟” 的（假设）放到堆的跟几点（即queue[0]的位置），然后通过循环遍历比较每个节点及其
子节点数据的大小，将数据x 下沉到合适的位置，以保证二叉堆的结构不被破坏。
siftDownUsingComparator与 siftDownComparable 唯一的区别是比较数据实用的是实例化
PriorityBlockingQueue时指定的比较器。
siftDownUsingComparator 方法代码如下：
private static <T> void siftDownUsingComparator(int k, T x, Object[] array,
int n,
Comparator<? super T> cmp) {
if (n > 0) {//堆中还有数据
//n除以2，因为二叉堆是一个满二叉树，所以只需要拿k的左右子节点中的小的数据进行比较，
// 所以这里只需要比较一半的数据
int half = n >>> 1;
while (k < half) {
//计算k左子节点的位置
int child = (k << 1) + 1;
Object c = array[child];
//k右子节点位置
int right = child + 1;
//得到k左右子节点中较小的数，然后与根节点的数进x行比较
if (right < n && cmp.compare((T) c, (T) array[right]) > 0)
c = array[child = right];
//若数据x小于根节点的左右子树节点数据，则表示最大位置的数据x可以作为根节点，直接退出
//否则，将根节点的左右子节点较小的数据作为根节点
if (cmp.compare(x, (T) c) <= 0)
break;
//更新位置k的值
array[k] = c;
//以较小的子节点作为根节点进行下一次循环比较
//最后k就是x的位置
k = child;
}
//循环结束后，所有数据已经满足二叉堆的特点，只差位置k的数据为null，k就是数据x的位置
array[k] = x;
}
}
