Java并发编程--并发队列原理之PriorityBlockingQueue

悠悠 2023-07-01 15:54 69阅读 0赞

### 文章目录 ###

*  PriorityBlockingQueue原理探究
 *   *   *  (1). 结构
         *  (2). PriorityBlockingQueue原理介绍
         *   *  1). offer操作
             *  2). poll操作
             *  3). take操作
         *  (3). 小结

# PriorityBlockingQueue原理探究 #

PriorityBlockingQueue是带优先级的无界阻塞队列,每次出队都返回优先级最高或者最低的元素,内部使用平衡二叉树堆实现,所以直接遍历队列元素不保证有序.在构造函数需传入comparator,用于插入元素时继续排序,若没有传入comparator,则插入的元素必须实现Comparatable接口.

### (1). 结构 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTk2NTY4_size_16_color_FFFFFF_t_70]

PriorityBlockingQueue内部有一个数组queue,用来存放队列元素,size存放元素个数.allocationSpinLock是一个自旋锁,其使用CAS操作保证只有一个线程可以对队列进行操作,状态为0或1.

没有notFull条件变量是因为这个队列是无界的,入队操作是非阻塞的.

### (2). PriorityBlockingQueue原理介绍 ###

#### 1). offer操作 ####

在队列中插入一个元素,因为是无界队列,所以一定会返回true.

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)
                // n是原队列第一个空位,e是入队元素,array是队列
                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) {
        
        // 释放锁
        // 使用CAS控制只能有一个线程成功扩容,释放锁让其他线程进行入队出队操作,降低并发性
        lock.unlock();
        Object[] newArray = null;
        // 这也是一个锁,只让一个线程进行扩容
        if (allocationSpinLock == 0 &&
            UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset, 0, 1)) {
        
            try {
        
                // 如果oldCap小于64,扩容为2倍+2,如果大于,扩容50%
                int newCap = oldCap + ((oldCap < 64) ? (oldCap + 2) : (oldCap >> 1));
                // 按照之前算法扩容后的容量如果溢出,则最小扩容量为原容量+1
                if (newCap - MAX_ARRAY_SIZE > 0) {
        
                    int minCap = oldCap + 1;
                    // 如果最小扩容量溢出或者小于0,那么扩容失败
                    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)
            Thread.yield();
        // 加锁,判断并拷贝数组
        lock.lock();
        if (newArray != null && queue == array) {
        
            queue = newArray;
            System.arraycopy(array, 0, newArray, 0, oldCap);
        }
    }
    // 建立堆
    private static <T> void siftUpComparable(int k, T x, Object[] array) {
        
        Comparable<? super T> key = (Comparable<? super T>) x;
        while (k > 0) {
        
            // 找到k的父节点,如果k小于父节点的值,将父节点置换为k
            // 直到k大于等于父节点的值,这样就构造了一个极小堆(所有父节点小于子节点)
            int parent = (k - 1) >>> 1;
            Object e = array[parent];
            if (key.compareTo((T) e) >= 0)
                break;
            array[k] = e;
            k = parent;
        }
        array[k] = key;
    }

#### 2). poll操作 ####

获取内部根节点的元素,如果队列为空,返回null

public E poll() {
        
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
        
            return dequeue();
        } finally {
        
            lock.unlock();
        }
    }
    // 获取内部根节点的元素,如果队列为空,返回null
    private E dequeue() {
        
        // 判断队列是否为空
        int n = size - 1;
        if (n < 0)
            return null;
        else {
        
            // 将第n个元素取出为x
            // 第0个元素取出为result
            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;
        }
    }
    // k为空闲位置,x为尾元素,array为堆,n为堆大小
    // 一直用小的孩子向上弥补父节点,直到最后一层,用最后一个节点补上
    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;// 无符号右移
            while (k < half) {
        
                // 子节点默认为左孩子
                int child = (k << 1) + 1;
                Object c = array[child];
                // 右孩子
                int right = child + 1;
                // 如果 (右孩子在堆内 && 左孩子大于右孩子) 那么右孩子代替左孩子作为孩子节点,并且c为孩子的值
                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;
        }
    }

#### 3). 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;
    }

### (3). 小结 ###

内部使用二叉树堆维护元素优先级,使用可扩容的数组作为元素存储的数据结构.出队时保证出队元素是根节点,并重置整个堆为极小堆.

内部使用了一个独占锁来控制并发.只使用了一个条件变量notEmpty而没有使用notFull是因为这个队列是无界的,不存在满队列情况.

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTk2NTY4_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200123154456782.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTk2NTY4,size_16,color_FFFFFF,t_70