Java中的阻塞队列
Published:
什么是阻塞队列?
阻塞队列(BlockingQueue)是一个支持两个附加操作的队列。这两个附加的操作都支持阻塞的插入和移除方法。
- 支持阻塞的插入方法:意思是当队列满时,队列会阻塞插入元素的线程,直到队列不满
- 支持阻塞的移除方法:意思是队列为空时,获取元素的线程会等待队列变为非空。
插入和移除操作的四种处理方式
方法/处理方式 | 抛出异常 | 返回特殊值 | 一直阻塞 | 超时退出 |
---|---|---|---|---|
插入方法 | add(e) | offer(e) | put(e) | offer(e,time,unit) |
移除方法 | remove() | poll() | take() | poll(time,unit) |
检查方法 | element() | peek() | 不可用 | 不可用 |
Java中的阻塞队列
- ArrayBlockingQueue:一个由数组结构组成的有界阻塞队列
- LinkedBlockingQueue:一个有由链表结构组成的无界阻塞队列
- PriorityBlockingQueue:一个支持优先级排序的无界阻塞队列
- DelayQueue:一个使用优先级队列实现的无界阻塞队列
- SynchronousQueue:一个不存储元素的阻塞队列
- LinkedTransferQueue:一个由链表结构组成的无界阻塞队列
- LinkedBlockingDeque:一个由链表结构组成的双向阻塞队列
ArrayBlockingQueue
是一个用数组实现的有界阻塞队列,此队列按照先进先出的原则对元素进行排序
默认情况下不保证线程公平的访问队列,所谓公平访问队列是指阻塞的线程,可以按照阻塞的先后顺序访问队列,即先阻塞的线程先访问队列。为了公平性往往会降低吞吐量
访问的公平性是使用可重入锁实现的
/**
* Creates an {@code ArrayBlockingQueue} with the given (fixed)
* capacity and the specified access policy.
*
* @param capacity the capacity of this queue
* @param fair if {@code true} then queue accesses for threads blocked
* on insertion or removal, are processed in FIFO order;
* if {@code false} the access order is unspecified.
* @throws IllegalArgumentException if {@code capacity < 1}
*/
public ArrayBlockingQueue(int capacity, boolean fair) {
if (capacity <= 0)
throw new IllegalArgumentException();
this.items = new Object[capacity];
lock = new ReentrantLock(fair);
notEmpty = lock.newCondition();
notFull = lock.newCondition();
}
LinkedBlockingQueue
是一个用链表实现的有界阻塞队列。此队列的默认和最大长度为Integer.MAX_VALUE。此队列按照先进先出的原则对元素进行排序。
/**
* Creates a {@code LinkedBlockingQueue} with a capacity of
* {@link Integer#MAX_VALUE}.
*/
public LinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}
PriorityBlockingQueue
是一个支持优先级的无界阻塞队列(无界只是没有指明界限,最大长度都为Integer.MAX_VALUE,具体可看size返回值,否则会出现问题),默认情况下元素采用自然顺序升序排列。也是自定义类实现compareTo()方法来指定元素排序规则,或者初始化队列指定构造参数比较器来对元素进行排序
DelayQueue
DelayQueue是一个支持延时获取元素的无界阻塞队列。队列使用PriorityQueue来实现。队列中的元素必须实现Delayed接口,在创建元素时可以指定多久才能从队列中获取当前元素。只有在延迟期满时才能从队列中提取元素
应用场景
- 缓存系统的设计:可以用DelayQueue保存缓存元素的有效期,使用一个线程循环查询DelayQueue,一旦能从DelayQueue中获取元素时,表示缓存有效期到了
- 定时任务调度:使用DelayQueue保存当天将会执行的任务和执行时间,一旦从DelayQueue中获取到任务就开始执行,比如TimerQueue就是使用DelayQueue实现的
SynchronousQueue
是一个不存储元素的阻塞队列。每一个put操作必须等待一个take操作,否则不能继续添加元素
它支持公平访问队列。默认情况下采用非公平策略访问队列。
/**
* Creates a {@code SynchronousQueue} with the specified fairness policy.
*
* @param fair if true, waiting threads contend in FIFO order for
* access; otherwise the order is unspecified.
*/
public SynchronousQueue(boolean fair) {
transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
}
SynchronousQueue可以看成是一个传球手,负责把生产者线程处理的数据直接传递给消费者线程。队列本身并不存储任何元素,非常适合传递性场景。吞吐量高于LinkedBlockingQueue和ArrayBlockingQueue
LinkedTransferQueue
是一个由链表结构组成的无界阻塞TransferQueue队列。相对于其他阻塞队列,多了tryTransfer和transfer方法
transfer方法
如果当前有消费者正在等待接收元素(消费者使用take()方法或带时间限制的poll()方法),transfer方法可以把生产者传入的元素立刻transfer给消费者。如果没有消费者在等待接收元素,transfer方法会将元素存放在队列的tail节点,并等到该元素被消费者消费了才返回。
tryTransfer方法
用来试探生产者传入的元素是否能直接传给消费者。如果没有消费者等待接收元素,则返回false。和transfer方法的区别是tryTransfer方法无论消费者是否接收,方法立即返回,而transfer方法是必须等待消费者消费了才返回
对于带有时间限制的tryTransfer(e,timeout,unit)方法,试图把生产者传入的元素直接传给消费者,但是如果没有消费者消费该元素则等待指定的时间再返回,如果超时还没消费元素,则返回false,如果在超时时间内消费了元素,则返回true
LinkedBlockingDeque
是一个由链表结构组成的双向阻塞队列。所谓双向队列指的是可以从队列的两端插入和移除元素,双向队列因为多了一个操作队列的入口,在多线程同时入队时,也就减少了一半的竞争。
双向队列可以运用在“工作窃取”模式中
阻塞队列的实现原理
使用通知模式实现 ,所谓通知模式,就是当生产者往满的队列里添加元素时会阻塞住生产者,当消费者消费了一个队列中的元素后,会通知生产者当前队列可用。
当往队列里插入一个元素时,如果队列不可用,那么阻塞生产者主要通过LockSupport.park(this)来实现,只有以下四种情况下的一种发生时,该方法才返回
- 与park对应的unpark执行或已经执行时。“已经执行”是指unpark先执行,然后再执行park的情况
- 线程中断时
- 等待完time参数指定的毫秒数时
- 异常现象发生时,这个异常现象没有任何原因
Linux下底层用的是系统方法pthread_cond_wait
实现