PriorityBlockingQueue

offer

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
            // 按照cmp指定的顺序进行上浮调整
            siftUpUsingComparator(n, e, array, cmp);
        // 队列元素总数+1      
        size = n + 1;
        // notEmpty满足，唤醒
        notEmpty.signal();
    } finally {
        lock.unlock();
    }
    return true;
}

tryGrow

private void tryGrow(Object[] array, int oldCap) {
    // 首先释放独占锁
    // 后面的扩容操作是需要成本的，如果一直持有锁，那么势必会降低吞吐量，而这里通过cas的方式来避免将扩容
    // 纳入到锁定的过程，最大化吞吐量
    lock.unlock(); // must release and then re-acquire main lock
    Object[] newArray = null;
    // 首先扩容操作是排他的，看allocationSpinLock是否被锁定
    if (allocationSpinLock == 0 &&
        UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset,
                                 0, 1)) {
        // 如果能成功锁定的话                         
        try {        
            // 计算新的容量
            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;
        }
    }
    // 如果newArray为空，说明其他线程在进行扩容的操作或者当前扩容失败，那么出让CPU给其他人试试
    if (newArray == null) // back off if another thread is allocating
        Thread.yield();
    // 独占锁再次锁定，后面会对前面初始化的新的array做拷贝动作。
    lock.lock();
    if (newArray != null && queue == array) {
        queue = newArray;
        System.arraycopy(array, 0, newArray, 0, oldCap);
    }
}

siftUpComparable

1560848789088.png

private static <T> void siftUpComparable(int k, T x, Object[] array) {
        Comparable<? super T> key = (Comparable<? super T>) x;
        // 这里的意义是上浮
        // 当插入节点的值要小于它的父节点的话，需要跟父节点进行交换
        // 直到上浮到比父节点大为止
        while (k > 0) {
            // 获取parent的下标
            int parent = (k - 1) >>> 1;
            Object e = array[parent];
            if (key.compareTo((T) e) >= 0)
                break;
            array[k] = e;
            k = parent;
        }
        array[k] = key;
    }

poll

public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return dequeue();
    } finally {
        lock.unlock();
    }
}

dequeue

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

siftDownComparable

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
        // k < half代表是非叶子节点
        while (k < half) {
            // 拿到k位置的左节点
            int child = (k << 1) + 1; // assume left child is least
            Object c = array[child];
            // 拿到k位置的右节点
            int right = child + 1;
            if (right < n &&
                // 这里的意义就是找到k节点的左右子节点的较小的那个
                ((Comparable<? super T>) c).compareTo((T) array[right]) > 0)
                c = array[child = right];
            // 如果key要比左右子节点都笑，那么直接替换掉父节点就好，不然继续往下看左子树
            if (key.compareTo((T) c) <= 0)
                break;
            // 将子节点替换掉父节点
            array[k] = c;
            // 将k调整到子节点
            k = child;
        }
        // 将x元素调整到k位置
        array[k] = key;
    }
}

1560853692421.png

take

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    E result;
    try {
        // 如果队列为空，那么notEmpty不满足，进而需要等待offer唤醒
        while ( (result = dequeue()) == null)
            notEmpty.await();
    } finally {
        lock.unlock();
    }
    return result;
}