并发编程之PriorityBlockingQueue

这几天利用空闲时间学习了一个新的阻塞队列-PriorityBlockingQueue，具有优先级的队列，也就是里面的节点是按照一定的顺序而排列的，当你执行take等方法时候，总是会弹出最大（最小）的节点。那么它是一种什么原理的呢？

What is PriorityBlockingQueue

这里首先讲讲PriorityBlockingQueue所用道的数据结构。

试想，如果自己要设计一个优先级队列，会怎么设计呢？

首先优先级必须是可比较的

极值在边界，这样才能够第一时间拿出优先级的值

那么有以下几种方法：

1）：通过排序，使得里面元素有序，从而能够取出比较后的优先级高的节点

2）：只让极值在边界，从而当遇到优先级高时候，能够立刻取出优先级高（低）的元素

在PriorityBlockingQueue中，是通过大根堆（小根堆）来实现的，那么大（小）根堆具体是啥意思呢？以大根堆为例，它有如下特点：

逻辑结构是二叉树构造

利用数组快速索引的特点，一般使用数组存储

堆分为大根堆（根节点最大）和小根堆（根节点最小），是完全二叉树

大根堆的要求是每个节点的值都不大于其父节点的值，即A[PARENT[i]] >= A[i]

获取顶端节点的时间复杂度为O(1)，建立堆的时间复杂度为n，而调整堆的时间复杂度为O(log n)。

下面主要以小根堆进行介绍。

模拟小根堆的插入删除

一般和大小根堆相关的概念还有就是堆排序，就是利用大小根堆的特性对一组数据进行排序。

这里先给出一张符合小根堆特性的图（注意，存储结构是数组）：

如上图所示，结合小根堆特性，符合以下特性：

- 根节点最小，同时，每科子树也具有相同特点

- 是一棵完全二叉树（具体概念自行获取）

插入新节点

现在假如，需要插入一个新节点，值的大小为1，那么首先需要将1放到数组的尾端，即10的后面一个，放入后树的结构如下：

下一步，就开始了调整操作，即对整棵树进行调整使其符合小根堆特性，根据该图，分为以下几步：（假设数组为a）

1. 找到最后一个非叶子节点即数组序号为4的节点，比较它和它的孩子大小（由数组特性可直接获得两个孩子节点序号），小的顶替它的位置，结果为a[4]和a[9]交换。

2. 再去检验a[3]节点，发现它正常，同理a[2]也正常

3. 检验a[1]，发现a[1]>a[4]，所以a[1]和a[4]交换，再需要向下检验新的a[4]值是否符合规范，最终符合

4. 最后检测根节点a[0]，结果a[0]和a[1]替换，此时需要检测新a[1]节点，发现符合规范，则最终调整完成。

最终结果为：

删除操作

假设，此时需要弹出a[0]节点，即值为1的节点，在PriorityBlockingQueue中，是将最后一个节点a[9]替换a[0]，同时将a[9]=null，然后再对a[0]~a[8]元素来调整来实现的。

即对如下存储的小根堆进行调整：

具体步骤如下：

1. 很明显，上图节点顺序不符合小根堆的特性，所以需要进行调整;

2. 因为第一个非叶子节点即a[3]～a[1]，都是符合小根堆特性（显然）所以直接对a[0]开始调整;

3. a[0] = (a[1]>a[2])? a[2]:a[1]，即和更小者替换，所以a[1]和a[0]替换;

4. 同理，a[1]和a[4]替换;

5. 整棵树符合规范

接下来，将结合PriorityBlockingQueue具体代码进行分析。

PriorityBlockingQueue

在上文中，应该大概的知道了，PriorityBlockingQueue对可比较元素进行优先级顺序的，下面，针对PriorityBlockingQueue的具体关键源码进行分析。

先看看其基本字段以及定义：

/**

* 没有界限的非阻塞队列。太多的话，会报错OutOfMemoryError，不允许null元素。

* 不能插入不可比较的元素。

* Iterator不保证特定的顺序。

* 如果需要有序，请使用： Arrays.sort(pq.toArray())

* 如果你一定要有特定的顺序，那么可以自己实现一个基于大小根堆的二级比较序列。

*/

public class PriorityBlockingQueue extends AbstractQueue

implements BlockingQueue, java.io.Serializable {

//默认初始大小

private static final int DEFAULT_INITIAL_CAPACITY = 11;

//最大值

private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;

//用数组来存储，优先级的队列。

private transient Object[] queue;

//优先级队列大小

private transient int size;

//比较器

private transient Comparator comparator;

//可重入锁

private final ReentrantLock lock;

//非空

private final Condition notEmpty;

//允许自旋的CAS

private transient volatile int allocationSpinLock;

...

}

从上面的基本字段可以看出，PriorityBlockingQueue的基本数据结构是使用堆排序原理，而使用ReentrantLock和Condition来对元素进行访问控制，是线程安全的阻塞工具队列。

add操作

下面看add操作：

public boolean add(E e) {

return offer(e);

}

再看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))

tryGrow(array, cap); //扩容

try {

Comparator cmp = comparator; //获取comparator。

if (cmp == null) //在n的位置插入e。向上调整。

siftUpComparable(n, e, array);

else

siftUpUsingComparator(n, e, array, cmp);

size = n + 1;

notEmpty.signal(); //通知等待中的队列。

} finally {

lock.unlock();

}

return true;

}

上面代码还是比较好理解的，就是加锁，然后调用siftUpComparable进行插入调整。

其中，siftUpComparable和siftUpUsingComparator分别是传入了比较器(Comparator)和没有传入时候的处理结果。

所以接下来直接看siftUpComparable即可：

private static void siftUpComparable(int k, T x, Object[] array) {

Comparable key = (Comparable) x; //获取x的k。保存key。

while (k > 0) { //k>0时候

int parent = (k - 1) >>> 1; //获取它的parent节点的index。

Object e = array[parent]; //获取parent节点值。

if (key.compareTo((T) e) >= 0) //符合根堆特性

break;

array[k] = e; //把k的位置，赋值为e，也就是把

k = parent; //把k复制为parent。

}

array[k] = key; //最后把k的位置赋值为key。

}

上面的方法siftUpComparable 是利用默认比较器向上调整过程，为啥会有向上调整呢？

因为如上文带图分析一样，当是插入节点时候，会在最后一个节点插入，再调整，因为引起不一致因素是最后一个节点，所以就一步一步向前调整，所以会有siftUpComparable ，当然，后文分析take方法时，会有siftDownComparable 方法。

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; //返回结果

}

take操作，就是一个阻塞式的出队操作，接下来看dequeue方法：

private E dequeue() {

int n = size - 1;

if (n

return null; //小于0返回null

else {

Object[] array = queue;

E result = (E) array[0]; //首先获得队头元素，也就是数组下标为0的元素。

E x = (E) array[n]; //获取最后一个元素赋值为x

array[n] = null; //把最后元素置为null。

Comparator cmp = comparator;

if (cmp == null) //看是否传入了comparator，没有就用默认的comparator。

siftDownComparable(0, x, array, n); //把x插入0的位置，重新调整以便堆。往下调整

else

siftDownUsingComparator(0, x, array, n, cmp);

size = n; //更改size

return result; //返回结果

}

}

整个过程与上文有图分析一直，把最后一个元素插入到第一个元素，再进行调整，下面看siftDownComparable方法：

private static void siftDownComparable(int k, T x, Object[] array,

int n) {

if (n > 0) { //首先得保证里面有东西

Comparable key = (Comparable)x; //首先获取x的Comparable类型

int half = n >>> 1; //从一半开始，也就是size/2

while (k

int child = (k

Object c = array[child];

int right = child + 1; //再获的k的右孩子节点

if (right

((Comparable) c).compareTo((T) array[right]) > 0) //如果不符合Comparable的特性。

c = array[child = right];

if (key.compareTo((T) c)

break;

array[k] = c; //把k的位置用c来表示。

k = child; //扩大为自己的左孩子大小。

}

array[k] = key; //当k>key，就是x应该在的key，那么直接把key赋值给array[k]

}

}

siftDownComparable 方法的意思就是向下调整，从根节点比较，一步一步比较知道符合堆的规范。

tryGrow操作

扩容操作，其实这个操作自我感觉方法写的很有意思，它会先释放锁，再去获取锁，先看代码：

private void tryGrow(Object[] array, int oldCap) {

lock.unlock(); // must release and then re-acquire main lock 一定要先释放锁，在获取锁。

Object[] newArray = null; //定义新数组

if (allocationSpinLock == 0 && //检测allocationSpinLock

UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset,0, 1)) { //allocationSpinLockOffset是用来在扩容队列时候做cas的，目的是保证只有一个线程可以进行扩容。

try {

int newCap = oldCap + ((oldCap

(oldCap + 2) : //扩容大小，最少2倍。

(oldCap >> 1));

if (newCap - MAX_ARRAY_SIZE > 0) { //检测是否超过最大值。

int minCap = oldCap + 1;

if (minCap MAX_ARRAY_SIZE)

throw new OutOfMemoryError();

newCap = MAX_ARRAY_SIZE;

}

if (newCap > oldCap && queue == array)

newArray = new Object[newCap]; //实例化数组。

} finally {

allocationSpinLock = 0; //最后重新赋值allocationSpinLock

}

}

if (newArray == null) //如果有桶等级的线程在等待，可以叫它们先干活，当前线程休息下

Thread.yield();

lock.lock(); //加锁

if (newArray != null && queue == array) {

queue = newArray; //指向新的引用

System.arraycopy(array, 0, newArray, 0, oldCap); //copy数组。

}

}

如上代码所示，在扩容操作时候，首先会释放锁，再以CAS去改变allocationSpinLock的值的方式去获取锁。

而最终，在try代码块中，只做了一件事，就是设置newArray的值，而在最后真真的数组copy阶段，则是需要重新获取锁。

如上述代码line 24行操作，以及line 27的拷贝操作。

那么为什么要进行先释放锁，在获取锁呢？

一方面是为了性能，因为扩容时候是需要花时间的，如果这些操作时候还占用锁那么其他线程在这个时候是不能进行出队操作的，也不能进行入队操作，这大大降低了并发性。

所以在扩容前释放锁，这允许其他出队线程可以进行出队操作，但是由于释放了锁，所以也允许在扩容时候进行入队操作，这就会导致多个线程进行扩容会出现问题，所以这里使用了一个spinlock用cas控制只有一个线程可以进行扩容，失败的线程调用Thread.yield()让出cpu。最终，多线程并发下最终会确定一个newArray，再加锁进行copy操作。

最后，全类没有一个字段用volatile字段修饰，可以好好理解下copy这里，即copy元素数据到新数组为啥放到获取锁（lock）后面那?原因应该是因为可见性问题，因为queue并没有被volatile修饰。另外有可能在扩容时候进行了出队操作，如果直接拷贝可能看到的数组元素不是最新的。而通过调用Lock后，获取的数组则是最新的，并且在释放锁前 数组内容不会变化。

而对于里面其他的一些内容如Iterator，则是比较好理解的，因为存储结构是数组，所以是基于数组的迭代操作，这里就不赘述了。