并发队列之PriorityBlockingQueue

这一篇说一下PriorityBlockingQueue,引用书中的一句话:这就是带优先级的无界阻塞队列,每次出队都返回优先级最高或者最低的元素(这里规则可以自己制定),内部是使用平衡二叉树实现的,遍历不保证有序;

其实也比较容易,就是基于数组实现的一个平衡二叉树,不了解平衡二叉树的可以先了解一下,别想的太难,原理跟链表差不多,只不过链表中指向下一个节点的只有一个,而平衡二叉树中有两个,一个左,一个右,还有左边的节点的值小于当前节点的值,右边节点的值大于当前节点的值;看看平衡二叉树的增删改查即可;

一.认识PriorityBlockingQueue

底层是以数组实现的,我们看看几个重要的属性:

<code>//队列默认初始化容量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;//如果队列是空的,还有线程来队列取数据,就阻塞//这里只有一个条件变量,因为这个队列是无界的,向队列中插入数据的话就用CAS操作就行了private final Condition notEmpty;//一个自旋锁,CAS使得同时只有一个线程可以进行扩容,0表示没有进行扩容,1表示正在进行扩容private transient volatile int allocationSpinLock;/<code>

简单看看构造器:

<code>//默认数组大小是11public 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];}//这个构造器也可以传入一个集合public PriorityBlockingQueue(Collection extends E> c) { this.lock = new ReentrantLock(); this.notEmpty = lock.newCondition(); boolean heapify = true; // true if not known to be in heap order boolean screen = true; // true if must screen for nulls 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();}/<code>

有兴趣的可以看看下面这个图,说的更详细,个人觉得看重要的地方就行了;

二.offer方法

在队列中插入一个元素,由于是无界队列,所以一直返回true;

<code>public boolean offer(E e) { //如果传入的是null,就抛异常 if (e == null) throw new NullPointerException(); final ReentrantLock lock = this.lock; //获取锁 lock.lock(); int n, cap; Object[] array; //[1]当前数组中实际数据总数>=数组容量,就进行扩容 while ((n = size) >= (cap = (array = queue).length)) //扩容 tryGrow(array, cap); try { Comparator super E> cmp = comparator; //[2]默认比较器为空时 if (cmp == null) siftUpComparable(n, e, array); else //[3]默认比较器不为空就用我们传进去的默认比较器 siftUpUsingComparator(n, e, array, cmp); //数组实际数量加一 size = n + 1; //唤醒notEmpty条件队列中的线程 notEmpty.signal(); } finally { //释放锁 lock.unlock(); } return true;}/<code>

上面的代码中,我们就关注那三个地方就行了,首先是[1]中扩容:

<code>private void tryGrow(Object[] array, int oldCap) { //首先释放获取的锁,这里不释放也行,只是扩容有的时候很慢,需要花时间,此时入队和出队操作就不能进行了,极大地降低了并发性 lock.unlock(); // must release and then re-acquire main lock Object[] newArray = null; //自旋锁为0表示队列此时没有进行扩容,然后用CAS将自旋锁从0该为1 if (allocationSpinLock == 0 && UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset, 0, 1)) { try { //用这个算法确定扩容后的数组容量,可以看到如果当前数组容量小于64,新数组容量就是2n+2,大于64,新的容量就是3n/2 int newCap = oldCap + ((oldCap < 64) ? (oldCap + 2) : (oldCap >> 1)); //判断新的数组容量是不是超过了最大容量,如果超过了,就尝试在老的数组容量加一,如果还是大于最大容量,就抛异常了 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 { //扩容完毕就将自旋锁变为0 allocationSpinLock = 0; } } //第一个线程在上面的if中执行CAS成功之后,第二个线程就会到这里,然后执行yield方法让出CPU,尽量让第一个线程执行完毕; 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); }}/<code>

再看[2]中的默认的比较器:

<code>//这里k表示数组中实际数量,x表示要插入到数组中的数据,array表示存放数据的数组private static void siftUpComparable(int k, T x, Object[] array) { //由此可知,我们要放进数组中的数据类型,必须要是实现了Comparable接口的 Comparable super T> key = (Comparable super T>) x; //这里判断数组中有没有数据,第一次插入数据的时候,k=0,不满足这个循环条件,那就直接走最下面设置array[0] = key //满足这个条件的话,首先获取父节点的索引,然后取出值,再比较该值和需要插入值的大小,决定是跳出循环还是继续循环 //这里比较重要,这个循环就是不断的调整二叉树平衡的,下面我们画图看看 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;}/<code>


随便举个例子看看怎么把平衡二叉树中的元素放到数组中,节点中的数据类型就以Integer了,其实就是将每一层从做到右一次放到数组中存起来,很明显,在数组中不是从小到大的顺序的;

这里注意一点,平衡二叉树的存放顺序不是唯一的,有很多种情况,跟你的存放顺序有关!

所以我们看看siftUpComparable方法中的while循环是怎么进行的?假设第一次调用offer(3),也就是调用siftUpComparable(0,3,array),这里假设array有足够的大小,不考虑扩容,那么第一次会走到while循环后面执行array[0]=3,下图所示:

第二次调用offer(1),也就是调用siftUpComparable(1,1,array),k=1,parent=0,所以父节点此时应该是3,然后1<3,不满足if语句,设置array[1]=3,k=0,然后继续循环不满足条件,执行array[0]=1,下图所示:

第三次调用offer(7),也就是调用siftUpComparable(2,7,array),k=2,parent=0,父节点为索引0的位置也就是1,因为7>1满足if语句,所以break跳出循环,执行array[2]=7,下图所示:

第四次调用offer(2),也就是调用siftUpComparable(3,2,array),k=3,parent=(k-1)>>>1=1,所以父节点表示索引为1的位置,也就是3,因为2<3,不满足if语句,所以设置array[3]=3,k=1,再进行一次循环,parent=0,此时父节点的值是1,2<3,不满足if,设置array[1]=1,k=0;再继续循环不满足循环条件,跳出循环,设置array[0] = 2

还是很容易的,有兴趣的话再多试试添加几个节点啊!其实还有[3]中使用我们自定义的比较器进行比较,其实i和上面代码一样的,另外put方法就是调用的offer方法,这里就不多说了

三.poll方法

poll方法的作用是获取并删除队列内部二叉树的根节点,如果队列为空,就返回nul;

<code>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; //取到第0个元素,这个也就是要返回的根节点 E result = (E) array[0]; //获取队列实际数量的最后一个元素,并把该位置赋值为null 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; }}private static void siftDownComparable(int k, T x, Object[] array, int n) { if (n > 0) { Comparable super T> key = (Comparable super T>)x; //[1] int half = n >>> 1; // loop while a non-leaf //[2] while (k < half) { int child = (k << 1) + 1; // assume left child is least Object c = array[child]; int right = child + 1; //[3] if (right < n &&((Comparable super T>) c).compareTo((T) array[right]) > 0) c = array[child = right]; //[4] if (key.compareTo((T) c) <= 0) break; array[k] = c; k = child; } array[k] = key; }}/<code>

所以我们主要的是看看siftDownComparable方法中是怎么将一个去掉了根节点的平衡二叉树调整平衡的;比如现在有如下所示的平衡二叉树:

调用poll方法,先是把最后一个元素保存起来x=3,然后将最后一个位置设置为null,此时实际调用的是siftDownComparable(0,3,array,3),key=3,half=1,k=0,n=3,满足[2],于是child=1,c=1,right=2,不满足[3],不满足[4],设置array[0]=1,k=1;继续循环,不满足循环条件,跳出循环,直接设置array[1]=3,最后poll方法返回的时2,下图所示:

其实可以简单的说说,最开始将数组中最后一个值X保存起来在适当时机插入到二叉树中,什么时候是适当时机呢?首先去掉根节点之后,得到根节点左子节点和右子节点的值leftVal和rightVal,如果X比leftVal小,那就直接把X放入到根节点的位置,整个平衡二叉树就平衡了!如果X比leftVal大,那就将leftVal的值设置到根节点中,再以左子节点做递归,继续比较X和左子节点的左节点的大小!仔细看看也没啥。

四.take方法

这个方法作用是获取二叉树中的根节点,也就是数组的第一个节点,队列为空,就阻塞;

<code>public E take() throws InterruptedException { //获取锁,可中断 final ReentrantLock lock = this.lock; lock.lockInterruptibly(); E result; try { //如果二叉树为空了,那么dequeue方法就会返回null,然后这里就会阻塞 while ( (result = dequeue()) == null) notEmpty.await(); } finally { //释放锁 lock.unlock(); } return result;}//这个方法前面说过,就是删除根节点,然后调整平衡二叉树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; }}/<code>

五.一个简单的例子

前面看了这个多方法,那就说说怎么使用吧,看看PriorityBlockingQueue这个阻塞队列怎么使用;

<code>package com.example.demo.study;import java.util.Random;import java.util.concurrent.PriorityBlockingQueue;import lombok.Data;public class Study0208 { @Data static class MyTask implements Comparable<mytask>{ private int priority=0; private String taskName; @Override public int compareTo(MyTask o) { if (this.priority>o.getPriority()) { return 1; } return -1; } } public static void main(String[] args) { PriorityBlockingQueue<mytask> queue = new PriorityBlockingQueue<mytask>(); Random random = new Random(); //往队列中放是个任务,从TaskName是按照顺序放进去的,优先级是随机的 for (int i = 1; i < 11; i++) { MyTask task = new MyTask(); task.setPriority(random.nextInt(10)); task.setTaskName("taskName"+i); queue.offer(task); } //从队列中取出任务,这里是按照优先级去拿出来的,相当于是根据优先级做了一个排序 while(!queue.isEmpty()) { MyTask pollTask = queue.poll(); System.out.println(pollTask.toString()); } }}/<mytask>/<mytask>/<mytask>/<code>