设为首页 加入收藏

TOP

PriorityBlockingQueue详解(一)
2023-07-23 13:45:44 】 浏览:37
Tags:PriorityBlockingQueue 详解

PriorityBlockingQueue介绍

  【1】PriorityBlockingQueue是一个无界的基于数组的优先级阻塞队列,数组的默认长度是11,也可以指定数组的长度,且可以无限的扩充,直到资源消耗尽为止,每次出队都返回优先级别最高的或者最低的元素。默认情况下元素采用自然顺序升序排序,当然我们也可以通过构造函数来指定Comparator来对元素进行排序。需要注意的是PriorityBlockingQueue不能保证同优先级元素的顺序。

  【2】优先级队列PriorityQueue: 队列中每个元素都有一个优先级,出队的时候,优先级最高的先出

PriorityBlockingQueue的源码分析

  【1】属性值

//默认容量
private static final int DEFAULT_INITIAL_CAPACITY = 11;

//最大容量设定
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;

//队列为空的时候的阻塞队列
private final Condition notEmpty;

//用于分配的CAS自旋锁
private transient volatile int allocationSpinLock;

//只用于序列化的普通优先队列
private PriorityQueue<E> q;

 

  【2】构造函数

//没有指定容量,则容量默认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];
}

 

  【3】核心方法分析

    1)核心扩容函数

//扩容函数
private void tryGrow(Object[] array, int oldCap) {
    lock.unlock(); //必须释放然后重新获得主锁,这一步的意义在于所有操作共享一把锁,在进行扩容时(因为写已满),释放锁(不能写,要等待扩容完才能写),提供给读操作
    Object[] newArray = null;
    // CAS 轻量级锁加锁,避免并发扩容
    if (allocationSpinLock == 0 && UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset, 0, 1)) {
        try {
            // 扩容步长,旧值小于64时,变为两倍+2。大于64时,变为1.5倍。
            int newCap = oldCap + ((oldCap < 64) ? (oldCap + 2) :  (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) 
        Thread.yield();
    //重新加锁
    lock.lock();
    //变更内存指向,利用内存拷贝复制旧数据
    if (newArray != null && queue == array) {
        queue = newArray;
        System.arraycopy(array, 0, newArray, 0, oldCap);
    }
}

 

    2)入队方法

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

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

 

    4)队列填入方式

      代码展示

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(
首页 上一页 1 2 下一页 尾页 1/2/2
】【打印繁体】【投稿】【收藏】 【推荐】【举报】【评论】 【关闭】 【返回顶部
上一篇@confirguration(proxyBeanMetho.. 下一篇Spring Boot 3.1 正式发布,王炸..

最新文章

热门文章

Hot 文章

Python

C 语言

C++基础

大数据基础

linux编程基础

C/C++面试题目