Java线程池队列PriorityBlockingQueue原理分析

/**
	* 默认数组长度
	*/
	private static final int DEFAULT_INITIAL_CAPACITY = 11;
    /**
	 * 最大达容量，分配时超出可能会出现 OutOfMemoryError 异常
     */
    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;
    /**
     * 空队列时进行线程阻塞的 Condition 对象
     */
    private final Condition notEmpty;

/**
	* 默认构造，使用长度为 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];
    }

    public void put(E e) {
        offer(e); // never need to block
    }
 	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.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];
            //比较是否大于该元素，不大于就没比较交换
            if (key.compareTo((T) e) >= 0)
                break;
            //以下都是元素位置交换
            array[k] = e;
            k = parent;
        }
        array[k] = key;
    }

/**
 * @Auther: Gentle
 * @Date: 2019/4/14 15:11
 * @Description: PriorityBlockingQueue 简单演示 demo 
 */
public class TestPriorityBlockingQueue {
    public static void main(String[] args) throws InterruptedException {
        PriorityBlockingQueue<Integer> concurrentLinkedQueue = new PriorityBlockingQueue<Integer>();
        concurrentLinkedQueue.offer(10);
        concurrentLinkedQueue.offer(20);
        concurrentLinkedQueue.offer(5);
        concurrentLinkedQueue.offer(1);
        concurrentLinkedQueue.offer(25);
        concurrentLinkedQueue.offer(30);
        //输出元素排列
        concurrentLinkedQueue.stream().forEach(e-> System.out.print(e+"  "));
        //取出元素
		Integer take = concurrentLinkedQueue.take();
        System.out.println();
        concurrentLinkedQueue.stream().forEach(e-> System.out.print(e+"  "));
    }
}

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

private static <T> void siftDownComparable(int k, T x, Object[] array,
                                               int n) {
    //判断队列长度
    if (n > 0) {
        Comparable<? super T> key = (Comparable<? super T>)x;
        //找到队列最后一个元素的父节点的索引。
        //如下图最大元素是30 父节点是 10，对于索引是 2
        int half = n >>> 1;           // loop while a non-leaf
        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 &&
                ((Comparable<? super T>) c).compareTo((T) array[right]) > 0)
                c = array[child = right];
            //比较父节点值是否大于子节点
            if (key.compareTo((T) c) <= 0)
                break;
            //下面都是元素替换
            array[k] = c;
            k = child;
        }
        array[k] = key;
    }
}