源码角度了解阻塞队列之PriorityBlockingQueue

源码角度了解阻塞队列之PriorityBlockingQueue

PriorityBlockingQueue是按照元素的优先级来进行出队的，同样我们看一下它的put()方法和take()方法，PriorityBlockingQueue用数组维护了一个元素二叉堆，它有一个锁和一个Condition notEmpty，notEmpty在队列为空的时候阻塞

它有一个成员变量比较器comparator，如果优先级队列使用元素的自然顺序，则为 null。

put()方法

PriorityBlockingQueue的put()方法之间调用了offer()方法，我们直接看一下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
                siftUpUsingComparator(n, e, array, cmp);
            size = n + 1;
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
        return true;
    }

将指定元素插入此优先级队列。由于队列是无界的，这个方法永远不会阻塞。

1. 加锁，可重入锁
2. 调用tryGrow()方法进行扩容，一般扩容百分之五十
3. 如果Comparator比较方法没有实现就调用自身的比较方法，调用siftUpComparable()方法入队，在位置 k 处插入项 x，将 x 沿着树往上移动直到它大于或等于其的父节点或者变成根
4. 如果实现了Comparator的比较方法，就调用比较方法进行比较调整元素位置
5. size加一，唤醒notEmpty的线程
6. 解锁

take()方法

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

1. 加锁，可中断锁
2. 调用dequeue()方法出队列，如果结果为空，说明是空队列，阻塞线程
3. 解锁
4. 返回结果

我们不妨看一下dequeue()方法的实现逻辑

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

1. 如果n的个数小于0，说明空队列，返回空
2. 否则，我们返回数组的第一个元素就可以，因为我们在入队的时候经过比较排列好了堆
3. 将第n个元素赋值给x，然后设置n的索引位置的值为空
4. 如果没有实现Comparator的比较方法就调用siftDownComparable()方法调整x的位置
5. 如果实现了Comparator的比较方法，就调用比较方法调整x的位置

总结

本篇文章主要介绍了PriorityBlockingQueue实现原理，它使用数组维护了一个二叉堆，将元素按照堆的结果来存储，入队的时候根据Comparator的比较方法来调整元素的位置，出队的时候直接出根元素，然后调整二叉堆