JUC的PriorityBlockingQueue如何使用

public class PriorityBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, Serializable {

    /** 队列默认初始容量 */
    private static final int DEFAULT_INITIAL_CAPACITY = 11;

    /**
     * 队列容量上限
     *
     * Some VMs reserve some header words in an array.
     * Attempts to allocate larger arrays may result in OutOfMemoryError: Requested array size exceeds VM limit
     */
    private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;

    /**
     * 存储队列元素的数组，按照最小堆组织
     *
     * Priority queue represented as a balanced binary heap:
     * the two children of queue[n] are queue[2*n+1] and queue[2*(n+1)].
     * The priority queue is ordered by comparator, or by the elements' natural ordering,
     * if comparator is null: For each node n in the heap and each descendant d of n, n <= d.
     * The element with the lowest value is in queue[0], assuming the queue is nonempty.
     */
    private transient Object[] queue;

    /** 队列中元素个数 */
    private transient int size;

    /** 队列元素比较器，如果为 null 则使用元素自带的比较器 */
    private transient Comparator<? super E> comparator;

    /** 保证队列操作线程安全的可重入独占锁 */
    private final ReentrantLock lock;

    /** 记录因为队列为空而阻塞的线程 */
    private final Condition notEmpty;

    /**
     * 扩容标记位，保证同一时间只有一个线程在扩容队列，状态为 0 或 1：
     * - 0： 表示当前没有在执行扩容操作
     * - 1： 表示当前正在执行扩容操作
     */
    private transient volatile int allocationSpinLock;

    /** 辅助支持序列化和反序列化 */
    private PriorityQueue<E> q;

    // ... 省略方法实现

}

public boolean offer(E e) {
    // 待添加元素不能为 null
    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)) {
        this.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.signal();
    } finally {
        // 释放锁
        lock.unlock();
    }
    return true;
}

private void tryGrow(Object[] array, int oldCap) {
    // 扩容之前，先释放锁，避免扩容期间阻塞其它线程的出队列、入队列操作
    lock.unlock(); // must release and then re-acquire main lock
    Object[] newArray = null;
    if (allocationSpinLock == 0 &&
            // 基于 CAS 操作将扩容标记位由 0 改为 1
            UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset, 0, 1)) {
        try {
            // 如果当前队列长度小于 64，则扩容为 2(n + 1)，否则扩容为 (1 + 1/2)n
            int newCap = oldCap + ((oldCap < 64) ? (oldCap + 2) : (oldCap >> 1)); // grow faster if small
            // 避免队列容量超过允许上限
            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;
        }
    }

    // 当前线程扩容失败，则让渡其它线程获取锁
    if (newArray == null) {
        Thread.yield();
    }

    // 加锁
    lock.lock();

    // 替换底层存储为扩容后的数组，并复制元素
    if (newArray != null && queue == array) {
        queue = newArray;
        System.arraycopy(array, 0, newArray, 0, oldCap);
    }
}

public E poll() {
    final ReentrantLock lock = this.lock;
    // 加锁
    lock.lock();
    try {
        // 出队列，获取最小堆堆顶元素值，并移除堆顶结点，调整最小堆
        return this.dequeue();
    } finally {
        // 释放锁
        lock.unlock();
    }
}

private E dequeue() {
    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);
        }
        // 队列结点计数减 1
        size = n;
        return result;
    }
}

public boolean remove(Object o) {
    final ReentrantLock lock = this.lock;
    // 加锁
    lock.lock();
    try {
        // 获取待删除元素的数组下标
        int i = this.indexOf(o);
        if (i == -1) {
            // 不存在
            return false;
        }
        // 移除元素
        this.removeAt(i);
        return true;
    } finally {
        // 释放锁
        lock.unlock();
    }
}

private void removeAt(int i) {
    Object[] array = queue;
    int n = size - 1;
    // 当前移除的是最后一个元素
    if (n == i) { // removed last element
        array[i] = null;
    }
    // 当前移除的是中间元素
    else {
        // 将数组最后一个位置置为 null，并调整堆的结构以满足最小堆定义
        E moved = (E) array[n];
        array[n] = null;
        Comparator<? super E> cmp = comparator;
        // 自上而下调整堆结构以满足最小堆定义
        if (cmp == null) {
            siftDownComparable(i, moved, array, n);
        } else {
            siftDownUsingComparator(i, moved, array, n, cmp);
        }
        // 自下而上调整堆结构以满足最小堆定义
        if (array[i] == moved) {
            if (cmp == null) {
                siftUpComparable(i, moved, array);
            } else {
                siftUpUsingComparator(i, moved, array, cmp);
            }
        }
    }
    // 队列结点计数减 1
    size = n;
}