【集合】PriorityBlockingQueue 源码分析

àì夳堔傛蜴生んèń 2022-12-29 11:23 129阅读 0赞

# 前言 #

Github：[https://github.com/yihonglei/jdk-source-code-reading][https_github.com_yihonglei_jdk-source-code-reading]（java-concurrent）

# 一 概述 #

PriorityBlockingQueue 基于数组实现的顺序队列，有界队列，默认 数组大小 11。

# 二 PriorityBlockingQueue 实例 #

package com.jpeony.concurrent.queue;
    
    import java.util.concurrent.PriorityBlockingQueue;
    
    /**
     * @author yihonglei
     */
    public class PriorityBlockingQueueTest {
        public static void main(String[] args) throws InterruptedException {
            PriorityBlockingQueue<String> priorityQueue = new PriorityBlockingQueue<>();
    
            priorityQueue.offer("one");
    
            System.out.println(priorityQueue.take());
        }
    }

# 三 PriorityBlockingQueue 底层实现 #

主要分析 put()/offer() 入队操作，take()/poll() 出队操作。

## 重要属性 ##

public class PriorityBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {
    
        /** 默认数组容量 */
        private static final int DEFAULT_INITIAL_CAPACITY = 11;
    
        /** 数组最大容量 */
        private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
    
        /** 基于 Object 类型数组实现队列 */
        private transient Object[] queue;
    
        /** 队列实际元素大小 */
        private transient int size;
    
        /** 队列元素排序实现 */
        private transient Comparator<? super E> comparator;
    
        /** 加锁机制 */
        private final ReentrantLock lock;
    
        /** take() 队列为空时，阻塞线程条件 */
        private final Condition notEmpty;
    
        // ......
    }

## 构造器 ##

创建 PriorityBlockingQueue 队列，默认数组大小 11，可以指定大小，可以指定队列元素比较方法。

/**
     * 默认构造器
     */
    public PriorityBlockingQueue() {
        // 默认大小 11
        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();
        // take() 队列为空，阻塞线程，等待被唤醒条件
        this.notEmpty = lock.newCondition();
        // 比较机制
        this.comparator = comparator;
        // 初始化数组大小
        this.queue = new Object[initialCapacity];
    }

## PriorityBlockingQueue\#put()/offer() ##

**offer()**

非阻塞方式入队。

public boolean offer(E e) {
        if (e == null)
            throw new NullPointerException();
        final ReentrantLock lock = this.lock;
        lock.lock();
        // 声明元素个数 n，数组 array
        int n, cap;
        Object[] array;
        // 赋值 n = size，array = queue，cap = array.length，如果满了，进行扩容操作
        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;
            // 唤醒 take() 时因为队列为空被阻塞的线程，重新尝试获取元素
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
        return true;
    }

tryGrow() 扩容操作

/**
     * 扩容操作，通过 CAS 加轻量级锁
     */
    private void tryGrow(Object[] array, int oldCap) {
        lock.unlock(); // must release and then re-acquire main lock
        Object[] newArray = null;
        // CAS 轻量级锁加锁，避免并发扩容
        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;
            }
        }
        // 并发扩容时，线程让出 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);
        }
    }

siftUpComparable() 默认排序 和 siftUpUsingComparator() 自定义排序操作。

/**
     * 默认排序
     */
    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;
    }
    
    /**
     * 自定义排序
     */
    private static <T> void siftUpUsingComparator(int k, T x, Object[] array,
                                       Comparator<? super T> cmp) {
        // 自己定义放入对象的比较方法，可以根据自己的需求，根据某个规则比较正序或倒序排序
        while (k > 0) {
            int parent = (k - 1) >>> 1;
            Object e = array[parent];
            if (cmp.compare(x, (T) e) >= 0)
                break;
            array[k] = e;
            k = parent;
        }
        array[k] = x;
    }

**put()**

PriorityBlockingQueue 的 put() 不需要阻塞，跟 LinkedBlockingQueue 和 ArrayBlockingQueue 不一样。

public void put(E e) {
        offer(e); // never need to block
    }

## PriorityBlockingQueue\#take()/poll() ##

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

dequeue() 出队操作。

/**
     * 出队
     */
    private E dequeue() {
        // 判断队列是否为空
        int n = size - 1;
        if (n < 0)
            return null;
        else {
            // 获取数组下标为0的元素，队列出队
            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() 或 siftDownUsingComparator，判断数组是否需要移出重排序右移。

/**
     * 判断数组是否需要重排序右移，默认比较方法
     */
    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];
                if (key.compareTo((T) c) <= 0)
                    break;
                array[k] = c;
                k = child;
            }
            array[k] = key;
        }
    }
    
    /**
     * 判断数组是否需要重排序右移，自定义比较方法
     */
    private static <T> void siftDownUsingComparator(int k, T x, Object[] array,
                                                    int n,
                                                    Comparator<? super T> cmp) {
        if (n > 0) {
            int half = n >>> 1;
            // 数组剩余元素小于一半的时候，清除已经出队的内存空间，右移
            while (k < half) {
                int child = (k << 1) + 1;
                Object c = array[child];
                int right = child + 1;
                if (right < n && cmp.compare((T) c, (T) array[right]) > 0)
                    c = array[child = right];
                if (cmp.compare(x, (T) c) <= 0)
                    break;
                array[k] = c;
                k = child;
            }
            array[k] = x;
        }
    }

**poll()**

非阻塞式出队。

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

# 四 总结 #

1、PriorityBlockingQueue 顺序阻塞有界队列，默认数组大小 11；

2、入队元素支持 默认或自定义排序；

[https_github.com_yihonglei_jdk-source-code-reading]: https://github.com/yihonglei/jdk-source-code-reading