11. PriorityBlockingQueue
2019-03-20 本文已影响0人
shallowinggg
PriorityBlockingQueue类实现了BlockingQueue接口。阅读BlockingQueue文本以获取有关的更多信息。
PriorityBlockingQueue是一个无限的并发队列。它使用与java.util.PriorityQueue类相同的排序规则。你不能将null插入此队列。
插入java.util.PriorityQueue的所有元素必须实现java.lang.Comparable接口。因此，元素根据你在Comparable 中的实现进行优先级排序。
注意，对于具有相同优先级的元素（compare()== 0），不会强制执行任何特定行为。
另请注意，如果你从PriorityBlockingQueue得到一个Iterator，Iterator不保证按优先级顺序迭代元素。
以下是使用PriorityBlockingQueue的示例：
BlockingQueue<String> queue = new PriorityBlockingQueue<String>();
//String implements java.lang.Comparable
queue.put("Value");
String value = queue.take();
源码
PriorityBlockingQueue内部使用了一个以数组为基础的二叉堆，所有的公共操作使用一个锁来进行保护。当对数组进行扩容时，放弃主锁，使用一个简单的自旋锁进行扩容，这样做是为了让扩容和提取元素同步进行。
成员变量
// 默认初始化大小
private static final int DEFAULT_INITIAL_CAPACITY = 11;
/**
* 数组可分配的最大容量。
* 一些虚拟机在数组中分配了对象头，尝试分配更大的容量
* 可能会导致OOM，请求的数组容量超过了允许的上限。
*/
private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
/**
* 优先级队列使用Comparator进行排序，或者通过元素的自然顺序，
* 即实现了Comparable接口。如果没有比较器：对于在堆中的每个结点n，
* 以及它的后代 d，n <= d。
*/
private transient Object[] queue;
// 队列元素数量
private transient int size;
// 比较器，为null代表使用自然顺序排序
private transient Comparator<? super E> comparator;
private final ReentrantLock lock;
private final Condition notEmpty;
/**
* 分配时的自旋锁，通过CAS获得
*/
private transient volatile int allocationSpinLock;
/**
* 只为序列化操作使用
*/
private PriorityQueue<E> q;
构造方法
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];
}
/**
* 如果给定的集合是 SortedSet或者 PriorityQueue， 这个优先级
* 队列根据同样的顺序排序。
*/
public PriorityBlockingQueue(Collection<? extends E> c) {
this.lock = new ReentrantLock();
this.notEmpty = lock.newCondition();
boolean heapify = true; // true 如果不知道二叉堆的顺序
boolean screen = true; // true 如果必须检查null
// 针对 SortedSet和 PriorityQueue处理
if (c instanceof SortedSet<?>) {
SortedSet<? extends E> ss = (SortedSet<? extends E>) c;
this.comparator = (Comparator<? super E>) ss.comparator();
heapify = false;
}
else if (c instanceof PriorityBlockingQueue<?>) {
PriorityBlockingQueue<? extends E> pq =
(PriorityBlockingQueue<? extends E>) c;
this.comparator = (Comparator<? super E>) pq.comparator();
screen = false;
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.
if (a.getClass() != Object[].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)
throw new NullPointerException();
}
this.queue = a;
this.size = n;
if (heapify)
heapify();
}
增加操作
// 因为PriorityBlockingQueue本身拒绝插入null，所以offer也需要抛出NPE，
// 复用offer方法即可
public boolean add(E e) {
return offer(e);
}
// 此队列是无界的，不会返回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;
}
private void tryGrow(Object[] array, int oldCap) {
lock.unlock(); // must release and then re-acquire main lock
Object[] newArray = null;
// 获取自旋锁
if (allocationSpinLock == 0 &&
ALLOCATIONSPINLOCK.compareAndSet(this, 0, 1)) {
try {
// 计算新容量。如果当前容量很小，那么直接扩容一倍多一点，
// 因为此时容量可能会迅速增长，否则扩容50%即可
int newCap = oldCap + ((oldCap < 64) ?
(oldCap + 2) : // grow faster if small
(oldCap >> 1));
// 如果新容量大于最大容量，那么计算当前最小容量（+1），
// 如果依然大于最大容量，抛出OOM
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;
}
}
// CAS竞争自旋锁失败，调度此线程
if (newArray == null) // back off if another thread is allocating
Thread.yield();
// 加锁，拷贝元素
lock.lock();
if (newArray != null && queue == array) {
queue = newArray;
System.arraycopy(array, 0, newArray, 0, oldCap);
}
}
// 此队列是无界的，所以永远不会被阻塞，复用offer方法即可
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 E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return dequeue();
} finally {
lock.unlock();
}
}
private E dequeue() {
int n = size - 1;
// 队列为空返回null
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);
else
siftDownUsingComparator(0, x, array, n, cmp);
size = n;
return result;
}
}
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;
}
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;
}
访问操作
public E peek() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return (size == 0) ? null : (E) queue[0];
} finally {
lock.unlock();
}
}
迭代器
PriorityBlockingQueue类中的迭代器和DelayQueue中的迭代器一样，都不会与原组件保证一致性。
// 调用toArray()方法获取当前的二叉堆
public Iterator<E> iterator() {
return new Itr(toArray());
}
/**
* Snapshot iterator that works off copy of underlying q array.
*/
final class Itr implements Iterator<E> {
final Object[] array; // Array of all elements
int cursor; // index of next element to return
int lastRet; // index of last element, or -1 if no such
Itr(Object[] array) {
lastRet = -1;
this.array = array;
}
public boolean hasNext() {
return cursor < array.length;
}
public E next() {
if (cursor >= array.length)
throw new NoSuchElementException();
return (E)array[lastRet = cursor++];
}
public void remove() {
if (lastRet < 0)
throw new IllegalStateException();
removeEQ(array[lastRet]);
lastRet = -1;
}
}
核心要点
- 必须提供要
Comparator接口或者队列元素实现Comparable接口。 - 可以同时进行扩容和提取元素的操作，不过只能有一个线程进行扩容
- 数组大小小于64时，进行双倍容量的扩展，否则扩容1.5倍
- 使用迭代器访问元素的顺序不会按指定的比较器顺序
- 迭代器不会与原数组保持一致性
下面是关于第四点的一个测试：
private static PriorityBlockingQueue<Integer> queue2 = new PriorityBlockingQueue<>();
@Test
public void test2() {
Random random = new Random(37);
for(int i = 0; i < 10; ++i) {
queue2.offer(random.nextInt(1000));
}
Iterator<Integer> itr = queue2.iterator();
itr.forEachRemaining(System.out::println);
}
输出如下：
18
73
26
123
200
430
562
505
259
230
