前言
PriorityBlockingQueue是一个支持优先级排序的无界阻塞队列.
实现思路
无并发情况
针对没有并发情况下,其实就是我们熟悉的
PriorityQueue, 对此请参考我的另外一篇博客 PriorityQueue 源码解析 .
并发情况
PriorityBlockingQueue是PriorityQueue的一个升级版, 可以处理多线程情况.
PriorityBlockingQueue是通过重入锁ReentrantLock和一个ConditionnotEmpty来实现的, 在操作过程需要用加锁来解决, 当队列为空需要消费元素时可以使用notEmpty.await()来使得该线程休眠等待, 另外由于该BlockingQueue是无界的, 所以并没有看到notFull.
源码
理解了基本思路后, 其实就是在
PriorityQueuePriorityQueue 源码解析 中加上了锁的操作.
属性
/**
* Default array capacity.
* 默认的数组空间
*/
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<? super E> comparator;
/**
* 重入锁
*/
private final ReentrantLock lock;
/**
* 当队列为空时需要用到
*/
private final Condition notEmpty;
/**
* Spinlock for allocation, acquired via CAS.
*
* 用来申请空间 注意是volatile
*/
private transient volatile int allocationSpinLock;
构造方法
public PriorityBlockingQueue() {
this(DEFAULT_INITIAL_CAPACITY, null);
}
/**
* Creates a {@code PriorityBlockingQueue} with the specified
* initial capacity that orders its elements according to their
* {@linkplain Comparable natural ordering}.
*
* @param initialCapacity the initial capacity for this priority queue
* @throws IllegalArgumentException if {@code initialCapacity} is less
* than 1
*/
public PriorityBlockingQueue(int initialCapacity) {
this(initialCapacity, null);
}
/**
*
* 如果初始化空间小于1 抛出运行时异常IllegalArgumentException
*
* @param initialCapacity the initial capacity for this priority queue
* @param comparator the comparator that will be used to order this
* priority queue. If {@code null}, the {@linkplain Comparable
* natural ordering} of the elements will be used.
* @throws IllegalArgumentException if {@code initialCapacity} is less
* than 1
*/
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; // 如果为null则使用对象的natural ordering
this.queue = new Object[initialCapacity];
}
辅助函数
private static <T> void siftUpComparable(int k, T x, Object[] array) {
Comparable<? super T> key = (Comparable<? super T>) x;
while (k > 0) {
int parent = (k - 1) >>> 1;
Object e = array[parent];
if (key.compareTo((T) e) >= 0)
break;
array[k] = e;
k = parent;
}
array[k] = key;
}
private static <T> void siftUpUsingComparator(int k, T x, Object[] array,
Comparator<? super T> cmp) {
while (k > 0) {
int parent = (k - 1) >>> 1;
Object e = array[parent];
if (cmp.compare(x, (T) e) >= 0)
break;
array[k] = e;
k = parent;
}
array[k] = x;
}
/**
* Inserts item x at position k, maintaining heap invariant by
* demoting x down the tree repeatedly until it is less than or
* equal to its children or is a leaf.
*
*
* @param k the position to fill
* @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<? super T> key = (Comparable<? super T>)x;
int half = n >>> 1; // loop while a non-leaf
while (k < half) {
int child = (k << 1) + 1; // assume left child is least
Object 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;
}
}
这些函数在PriorityQueue 源码解析 中已经详细分析过.
加入元素
/**
* 将元素e加入到优先队列中
* true 表示加入成功
* false 表示加入失败
*
* 以下情况下会抛出运行时异常:
* 1. e为null时会抛出NullPointerException异常
* 2. 如果元素e不能被比较 会抛出ClassCastException异常
* 3. 无法扩容是 抛出OutofMemoryException异常
*
* @param e the element to add
* @return {@code true} (as specified by {@link Queue#offer})
* @throws ClassCastException if the specified element cannot be compared
* with elements currently in the priority queue according to the
* priority queue's ordering
* @throws NullPointerException if the specified element is null
*/
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; // size加1
notEmpty.signal();
} finally {
lock.unlock();
}
return true;
}
/**
* 因为默认情况下该优先队列是无界
* 当无法扩容时 会抛出OutofMemoryException
* 所以直接调用offer方法即可
* 以下情况下会抛出运行时异常:
* 1. e为null时会抛出NullPointerException异常
* 2. 如果元素e不能被比较 会抛出ClassCastException异常
* 3. 无法扩容是 抛出OutofMemoryException异常
* @param e the element to add
* @throws ClassCastException if the specified element cannot be compared
* with elements currently in the priority queue according to the
* priority queue's ordering
* @throws NullPointerException if the specified element is null
*/
public void put(E e) {
offer(e); // never need to block
}
/**
*
* 因为无界 所以也不存在无法增加情况 所以直接调用offer方法即可
* 另外以下情况下会抛出运行时异常:
* 1. e为null时会抛出NullPointerException异常
* 2. 如果元素e不能被比较 会抛出ClassCastException异常
* 3. 无法扩容是 抛出OutofMemoryException异常
* @param e the element to add
* @param timeout This parameter is ignored as the method never blocks
* @param unit This parameter is ignored as the method never blocks
* @return {@code true} (as specified by
* {@link BlockingQueue#offer(Object,long,TimeUnit) BlockingQueue.offer})
* @throws ClassCastException if the specified element cannot be compared
* with elements currently in the priority queue according to the
* priority queue's ordering
* @throws NullPointerException if the specified element is null
*/
public boolean offer(E e, long timeout, TimeUnit unit) {
return offer(e); // never need to block
}
可以看到在
put方法中获得锁后最后调用siftUpComparable或者siftUpUsingComparator方法,这两个方法在单线程中已经有所讲解. 可以看到的是PriorityQueue是无界的,所以put和offer都是在不发生异常的情况下始终可以加入元素的.
获取元素
/**
* Mechanics for poll(). Call only while holding lock.
*
* 出队列
* 跟正常堆操作出堆一样
*
*/
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);
else
siftDownUsingComparator(0, x, array, n, cmp);
size = n;
return result;
}
}
/**
* 从优先队列中取优先级最高的元素 如果队列为空会一直等待
* 以下情况会出现异常:
* 1. 获得锁或者在因为队列空而休眠等待的过程中被其他线程中断
*
* @return
* @throws InterruptedException
*/
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;
}
最终还是到了
siftDownComparable或者siftDownUsingComparator方法.poll如果队列空的时候会立即返回null不会等待.就不多说了,
删除元素
/**
* Removes the ith element from queue.
* 删除下标为i的元素
*/
private void removeAt(int i) {
Object[] array = queue;
int n = size - 1;
if (n == i) // removed last element
array[i] = null;
else {
E moved = (E) array[n];
array[n] = null;
Comparator<? super E> cmp = comparator;
if (cmp == null)
siftDownComparable(i, moved, array, n);
else
siftDownUsingComparator(i, moved, array, n, cmp);
if (array[i] == moved) {
if (cmp == null)
siftUpComparable(i, moved, array);
else
siftUpUsingComparator(i, moved, array, cmp);
}
}
size = n;
}
/**
* Removes a single instance of the specified element from this queue,
* if it is present. More formally, removes an element {@code e} such
* that {@code o.equals(e)}, if this queue contains one or more such
* elements. Returns {@code true} if and only if this queue contained
* the specified element (or equivalently, if this queue changed as a
* result of the call).
*
* @param o element to be removed from this queue, if present
* @return {@code true} if this queue changed as a result of the call
*/
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();
}
}
都是获得锁后执行的就是PriorityQueue 源码解析 单线程的方法.
遍历元素
与
PriorityQueue遍历有所不同的是,PriorityQueue会事先copy一份当前队列中的数据给此iterator,即使在遍历过程中删除某个元素也不会影响此iterator中的元素.
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();
lastRet = cursor;
return (E)array[cursor++];
}
public void remove() {
if (lastRet < 0)
throw new IllegalStateException();
removeEQ(array[lastRet]);
lastRet = -1;
}
}
例子如下
package com.priorityblockingqueue;
import java.util.Comparator;
import java.util.Iterator;
public class Test01 {
public static void main(String[] args) {
PriorityBlockingQueue<Integer> pq = new PriorityBlockingQueue<>(10, new Comparator<Integer>() {
@Override
public int compare(Integer o1, Integer o2) {
return o1 - o2;
}
});
pq.add(1);
pq.add(3);
pq.add(12);
pq.add(4);
pq.add(6);
pq.add(17);
pq.add(13);
pq.add(8);
pq.add(5);
pq.add(10);
pq.add(11);
pq.add(19);
pq.add(23);
pq.add(14);
System.out.println(pq);
Iterator<Integer> iter = pq.iterator();
while (iter.hasNext()) {
int val = iter.next();
if (val == 23) iter.remove();
System.out.print(val + " ");
}
}
}
输出如下:
[1, 3, 12, 4, 6, 17, 13, 8, 5, 10, 11, 19, 23, 14]
1 3 12 4 6 17 13 8 5 10 11 19 23 14
参考
1.
Java 1.8
网友评论