PriorityBlockingQueue源码解析

public class PriorityBlockingQueue<E> extends AbstractQueue<E>
    implements BlockingQueue<E>, java.io.Serializable {
    // 默认容量为11
    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更新这个值，谁更新成功了谁扩容，其它线程让出CPU
    private transient volatile int allocationSpinLock;
    // 不阻塞的优先级队列，非存储元素的地方，仅用于序列化/反序列化时
    private PriorityQueue<E> q;
}

// 默认容量为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.comparator = comparator;
    this.queue = new Object[Math.max(1, initialCapacity)];
}

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);
        // 插入元素完毕，元素个数加1            
        size = n + 1;
        // 唤醒notEmpty条件
        notEmpty.signal();
    } finally {
        // 解锁
        lock.unlock();
    }
    return true;
}

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];
        // 如果key大于父节点，堆化结束
        if (key.compareTo((T) e) >= 0)
            break;
        // 否则，交换二者的位置，继续下一轮比较
        array[k] = e;
        k = parent;
    }
    // 找到了应该放的位置，放入元素
    array[k] = key;
}

private void tryGrow(Object[] array, int oldCap) {
    // 先释放锁，因为是从offer()方法的锁内部过来的
    // 这里先释放锁，使用allocationSpinLock变量控制扩容的过程
    // 防止阻塞的线程过多
    lock.unlock(); // must release and then re-acquire main lock
    Object[] newArray = null;
    // CAS更新allocationSpinLock变量为1的线程获得扩容资格
    if (allocationSpinLock == 0 &&
        UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset,
                                 0, 1)) {
        try {
            // 旧容量小于64则翻倍，旧容量大于64则增加一半
            int newCap = oldCap + ((oldCap < 64) ?
                                   (oldCap + 2) : // grow faster if small
                                   (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) // 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);
    }
}

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

private E dequeue() {
    // 元素个数减1
    int n = size - 1;
    if (n < 0)
        // 数组元素不足，返回null
        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
        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;
        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];
            // key如果比左右子节点都小，则堆化结束
            if (key.compareTo((T) c) <= 0)
                break;
            // 否则，交换key与左右子节点中最小的节点的位置
            array[k] = c;
            k = child;
        }
        // 找到了放元素的位置，放置元素
        array[k] = key;
    }
}