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((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; }
图解说明
1.利用了满二叉树的理念,(k - 1) >>> 1可以获得存入节点的父节点下标,然后进行比较。若判断应该上升,则将父节点置于k存储的地方。
2.重复循环,知道二叉树root节点,或者已经找到了合适的位置
4)出队方法
public E poll() { final ReentrantLock lock = this.lock; lock.lock(); try { return dequeue(); } finally { lock.unlock(); } } 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() { 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; } }
5)队列修正方式
代码展示
private static <T> void siftDownComparable(int k, T x, Object[] array, int n) { if (n > 0) { Comparable<? super T> key = (Comparable<? super T>)x; //过滤掉最后一层的元素(以为满二叉树中,最后一层占据的元素就是n/2) 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; } }
图解说明
1.采用尾结点代替头节点,然后利用下沉的方式来修正优先队列里面的数据。
2.下沉限制在倒数第二层,以为倒数第二层就会与倒数第一层进行比较了,所以应该进行限制(下标位置超出倒数第二层的最大下标就应该停止了)
3.其次看的时候,可以把左右子树都当做一个小的满二叉树,不断逐层划分,这样条理更清晰。
PriorityBlockingQueue总结
【1】一个支持优先级排序的无界阻塞队列,优先级高的先出队,优先级低的后出队
【2】数据结构:数组+二叉堆(默认容量11,可指定初始容量,会自动扩容,最大容量是(Integer.MAX_VALUE - 8))
【3】锁:ReentrantLock,存取是同一把锁
【4】阻塞对象:NotEmpty,出队,队列为空时阻塞
【5】入队,不阻塞,永远返回成功,无界;根据比较器进行堆化(排序)自下而上,如果比较器为 null,则按照自然顺序排序,传入比较器对象就按照比较器的顺序排序。
【6】出队,优先级最高的元素在堆顶(弹出堆顶元素),弹出前比较两个子节点再进行堆化(自上而下)。
【7】应用场景:1.业务办理排队叫号,VIP客户插队;2.电商抢购活动,会员级别高的用户优先抢购到商品;
标签:int,lock,Object,private,详解,array,null,PriorityBlockingQueue From: https://www.cnblogs.com/chafry/p/16783035.html