读PriorityBlockingQueue源码记录
简介
PriorityBlockingQueue是一个没有边界的优先级队列，它的排序规则和 java.util.PriorityQueue一样。PriorityBlockingQueue中允许插入null对象,所有插入PriorityBlockingQueue的对象必须实现java.lang.Comparable接口，队列优先级的排序规则就是按照我们对这个接口的实现来定义的。另外，从PriorityBlockingQueue获得一个迭代器Iterator，但这个迭代器并不保证按照优先级顺序进行迭代。
PriorityBlockingQueue底层是基于数组实现的堆结构，这里对堆结构不做多介绍，可以参考
堆，最大堆（大顶堆）及最小堆（小顶堆）的实现
常用数据结构与算法：二叉堆(binary heap)
探索PriorityBlockingQueue
1.主要属性
//默认容量
private static final int DEFAULT_INITIAL_CAPACITY = 11;
//最大容量
private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
//内部堆数组，存放实际数据，可以看成二叉树
//顶点 queue[i] 左子节点 queue[2*i+1],右子节点queue[2*(i+1)]，父节点queue[(i-1)/2]
private transient Object[] queue;
//当前队列中元素个数
private transient int size;
//比较器，为空时根据元素实现的Comparable接口的逻辑进行比较
private transient Comparator<? super E> comparator;
//全局锁
private final ReentrantLock lock;
//非空阻塞条件
private final Condition notEmpty;
//扩容数组分配资源时的自旋锁
private transient volatile int allocationSpinLock;
//?
private PriorityQueue q;
从属性可以看出，优先级队列PriorityBlockingQueue底层也是以数组的方式实现，与ArrayBlockingQueue一样，只有一把锁。
2.初始化
接着看构造方法：
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];
}
3.入队
入列方法有offer()，和put()。
先看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<? super E> cmp = comparator;
if (cmp == null)
//比较器为空，以元素实现的Comparable接口进行堆调整
siftUpComparable(n, e, array);
else
//比较器非空, 则按照比较器进行堆调整
siftUpUsingComparator(n, e, array, cmp);
//当前长度+1
size = n + 1;
//唤醒等待在notEmpty上的出队线程
notEmpty.signal();
} finally {
lock.unlock();
}
return true;
}
这里需要注意的队列的扩容方法tryGrow(Object[] array, int oldCap)，该方法会首先释放全局锁，通过简单的自旋锁实现扩容时不影响容器其他并发操作的方式。
private void tryGrow(Object[] array, int oldCap) {
//先释放锁，原因是PriorityBlockingQueue主要通过CAS设置allocationSpinLock来判断哪个线程获得了扩容权限，如果没抢到权限就会让出CPU使用权
//通过CAS来控制扩容安全性，可以保证扩容和入队/出队同时进行, 所以先释放全局锁
lock.unlock(); // must release and then re-acquire main lock
Object[] newArray = null;
if (allocationSpinLock == 0 && UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset,0, 1)) {
try {
// 计算新的数组大小，当前长度小于64时，长度扩大到原来的2倍+2，否则扩大到原来的1.5倍
int newCap = oldCap + ((oldCap < 64) ?(oldCap + 2) : (oldCap >> 1));
//溢出判断
if (newCap - MAX_ARRAY_SIZE > 0) {
int minCap = oldCap + 1;
if (minCap < 0 || minCap > MAX_ARRAY_SIZE)
throw new OutOfMemoryError();
newCap = MAX_ARRAY_SIZE;
}
//queue == array主要是判断当前的数组是否已经发生改变
if (newCap > oldCap && queue == array)
newArray = new Object[newCap];
} finally {
allocationSpinLock = 0;
}
}
if (newArray == null)
//让出 CPU 调度(因为其他线程扩容后必定有其他的操作)
Thread.yield();
lock.lock();
//再次判断数组是否发生变化
if (newArray != null && queue == array) {
//进行数据拷贝
queue = newArray;
System.arraycopy(array, 0, newArray, 0, oldCap);
}
}
接着看堆调整siftUpComparable和siftUpUsingComparator方法，这两个方法内部几乎一样，这里以siftUpComparable为例，如果有细看上面链接的关于二叉堆文章，下面的源码应该不难理解，主要思路是将当前元素与父节点[(n-1)/2]比较，直到k>=parent(或k<=parent，这里根据实现的Comparable接口的逻辑来定)。
private static <T> void siftUpComparable(int k, T x, Object[] array) {
Comparable<? super T> key = (Comparable<? super T>) x;
while (k > 0) {
// 相当于(k-1)/2, 即k结点的父结点索引parent
int parent = (k - 1) >>> 1;
Object e = array[parent];//parent的真实值
if (key.compareTo((T) e) >= 0)//当前元素与parent进行比较
break;
array[k] = e;//将k索引位置的数据设置成parent的值
k = parent; //将本次parent的索引作为下个循环比较的索引
}
array[k] = key;
}
再看offer()的重载方法offer(E e, long timeout, TimeUnit unit)和put()方法，其实调用的是offer()方法，由于队列是无界的，所以不会阻塞线程.
public boolean offer(E e, long timeout, TimeUnit unit) {
return offer(e); // never need to block
}
public void put(E e) {
offer(e); // never need to block
}
4.出队
出队同样有poll()和take()等方法，逻辑基本一致，所以这里以take()进行分析，从源码可以看到take()方法调用的dequeue()方法。
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;
}
dequeue()方法：
private E dequeue() {
//长度减1，如果队列为空，返回null
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;
}
}
接着看堆调整siftDownComparable和siftDownUsingComparator方法，这两个方法内部几乎一样，这里分析siftDownComparable源码。逻辑与上面链接上的删除操作基本一致。自己走一遍就能理解了。
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; //获取x的parent的父节点索引位置
while (k < half) {
int child = (k << 1) + 1; //获取k索引的左子节点索引
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索引设置为最小值的子节点
k = child; //该子节点的索引作为下个循环的父节点
}
array[k] = key;
}
}
查看16道真题和解析
拼多多集团-PDD成长空间 486人发布