并发之BlockingQueue

2024-06-07 09:58

文章标签 并发 blockingqueue

本文主要是介绍并发之BlockingQueue，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

并发包中BlockingQueue的类关系图如下：

这里写图片描述

public interface BlockingQueue<E> extends Queue<E> {/*** 往队尾增加元素，如果队列已满则抛出IllegalStateException异常*/boolean add(E e);/*** 往队尾增加元素，如果队列满则返回false*/boolean offer(E e);/*** 往队尾添加元素，如果队列满则阻塞*/void put(E e) throws InterruptedException;/*** 往队尾增加元素，如果队列满等待timeout时间，如果仍无法添加，则返回false*/boolean offer(E e, long timeout, TimeUnit unit)throws InterruptedException;/*** 从队列中取出队头元素,如果队列为空则阻塞*/E take() throws InterruptedException;/*** 从队列中取出队头元素，如果为空则返回false*/E poll();/*** 从队列中取出队头元素,如果队列空等待timeout时间，如果仍无法获取，则返回false*/E poll(long timeout, TimeUnit unit)throws InterruptedException;/*** 从队列中取出队头元素,如果队列为空则抛出NoSuchElementException异常*/E remove();/*** 从队列中移除o对象（调用equals()比较），如果不存在或者队列为空，则返回false。如队列中存在多个o对象，则移除第一个*/boolean remove(Object o);/*** 取出队头元素但不删除，如果队列为空则抛出NoSuchElementException异常*/E element();/*** 取出队头元素但不删除，如果队列为空则返回null*/E peek();/*** 返回队列剩余空间*/int remainingCapacity();/*** 查询队列中是否存在o对象*/public boolean contains(Object o);/*** 删除队列中于c的公共元素*/int drainTo(Collection<? super E> c);/*** 删除队列中于c的公共元素，最多删除maxElements个*/int drainTo(Collection<? super E> c, int maxElements);
}

操作	抛出异常	返回boolean/null	阻塞
入队	add	offer	put
出队	remove	poll	take
查看队头	element	peek

ArrayBlockingQueue

使用ReentrantLock进行同步，用数组实现双向队列来保存元素，创建ArrayBlockingQueue需要指定队列的大小，不像ArrayList可以拓容，是一种有界队列。

ArrayBlockingQueue重要包含如下几个成员变量

/** 保存元素的数组 */final Object[] items;/** 队头元素位置 */int takeIndex;/** 队尾元素位置（待入队位置） */int putIndex;/** 队列中的元素个数 */int count;

LinkedBlockingQueue

使用有头结点链表实现队列，类似LinkedList，其最大节点数可达到Integer.MAX_VALUE。

LinkedBlockingDeque

使用链表实现的双向队列，最大节点数为Integer.MAX_VALUE。

PriorityBlockingQueue

优先阻塞队列使用数组存储队列元素，实则使用了大根堆来实现。大根堆的性质可以保证有线队列的堆顶元素一定为整个堆中最大的（但是不保证堆内有序）。
队列的初始大小为11，最大容量为Integer.MAX_VALUE - 8，每当原数组容量不够时，将会调用拓容方法，尤其注意，拓容前会释放队列的锁。

private void tryGrow(Object[] array, int oldCap) {lock.unlock(); // must release and then re-acquire main lockObject[] newArray = null;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 overflowint 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;}}if (newArray == null) // back off if another thread is allocatingThread.yield();lock.lock();//queue == array这个判断条件是为了判断此时队列是否已经被其他线程拓容if (newArray != null && queue == array) {queue = newArray;System.arraycopy(array, 0, newArray, 0, oldCap);}
}