并发队列之PriorityBlockingQueue

2020-02-10 16:05:36来源：博客园阅读 ()

并发队列之PriorityBlockingQueue

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

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

一.认识PriorityBlockingQueue

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

//队列默认初始化容量
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;

　　简单看看构造器：

//默认数组大小是11
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];
}
//这个构造器也可以传入一个集合
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();
}

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

二.offer方法

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

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;
}

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

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);
    }
}

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

//这里k表示数组中实际数量，x表示要插入到数组中的数据，array表示存放数据的数组
private static <T> 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;
}

　　随便举个例子看看怎么把平衡二叉树中的元素放到数组中，节点中的数据类型就以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；

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 <T> 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;
    }
}

　　所以我们主要的是看看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方法

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

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;
    }
}

五.一个简单的例子

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

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());
        }
        
    }

}