PriorityBlockingQueue介绍

　　【1】PriorityBlockingQueue是一个无界的基于数组的优先级阻塞队列，数组的默认长度是11，也可以指定数组的长度，且可以无限的扩充，直到资源消耗尽为止，每次出队都返回优先级别最高的或者最低的元素。默认情况下元素采用自然顺序升序排序，当然我们也可以通过构造函数来指定Comparator来对元素进行排序。需要注意的是PriorityBlockingQueue不能保证同优先级元素的顺序。

　　【2】优先级队列PriorityQueue： 队列中每个元素都有一个优先级，出队的时候，优先级最高的先出。

PriorityBlockingQueue的源码分析

　　【1】属性值

//默认容量
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;

//用于分配的CAS自旋锁
private transient volatile int allocationSpinLock;

//只用于序列化的普通优先队列
private PriorityQueue<E> q;

 

　　【2】构造函数

//没有指定容量，则容量默认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];
}

 

　　【3】核心方法分析

　　　　1）核心扩容函数

//扩容函数
private void tryGrow(Object[] array, int oldCap) {
    lock.unlock(); //必须释放然后重新获得主锁，这一步的意义在于所有操作共享一把锁，在进行扩容时（因为写已满），释放锁（不能写，要等待扩容完才能写），提供给读操作
    Object[] newArray = null;
    // CAS 轻量级锁加锁，避免并发扩容
    if (allocationSpinLock == 0 && UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset, 0, 1)) {
        try {
            // 扩容步长，旧值小于64时，变为两倍+2。大于64时，变为1.5倍。
            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 {
            allocationSpinLock = 0;
        }
    }
    // 并发扩容时，线程让出 cpu 执行时间，给其他线程执行，自己稍后执行，原因：加锁不成功必然有其他线程也在扩容，在等待过程中让出资源给其他线程利用
    if (newArray == null) 
        Thread.yield();
    //重新加锁
    lock.lock();
    //变更内存指向，利用内存拷贝复制旧数据
    if (newArray != null && queue == array) {
        queue = newArray;
        System.arraycopy(array, 0, newArray, 0, oldCap);
    }
}

 

　　　　2）入队方法

public void put(E e) {
    offer(e); // never need to block
}
public boolean add(E e) {
    return offer(e);
}

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

 

　　　　4）队列填入方式

　　　　　　代码展示

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(