Java阻塞队列是一种可以实现阻塞插入阻塞移除的特殊队列,它被广泛应用于生产者消费者模式。如在线程池中,可以使用LinkedBlockingQueueSynchronousQueue等阻塞队列来生产与消费线程对象。

下面将简单介绍Java中常用的几个阻塞队列。

一、阻塞队列汇总

阻塞队列位于JUC包(即java.util.concurrent包),该包保存了Java并发的大多数类,如:锁、阻塞队列、线程池等。

JUC下的阻塞队列共有3个接口与7个实现类,他们的关系如下所示:

@startuml Java阻塞队列类图

title Java阻塞队列类图

class ArrayBlockingQueue
interface BlockingDeque
interface BlockingQueue
class DelayQueue
class LinkedBlockingDeque
class LinkedBlockingQueue
class LinkedTransferQueue
class PriorityBlockingQueue
class SynchronousQueue
interface TransferQueue

BlockingQueue <|.. ArrayBlockingQueue
BlockingQueue <|-- BlockingDeque
BlockingQueue <|.. DelayQueue
BlockingQueue <|.. LinkedBlockingQueue
BlockingQueue <|.. PriorityBlockingQueue
BlockingQueue <|.. SynchronousQueue
BlockingQueue <|-- TransferQueue
BlockingDeque <|.. LinkedBlockingDeque
TransferQueue <|.. LinkedTransferQueue

@enduml

上述类图中,BlockingQueue是阻塞队列的顶层接口,BlockingDequeTransferQueue则是继承自BlockingQueue的两个用于实现特殊功能的接口。剩下的7个阻塞队列实现类均实现了上述3个接口,它们如下所示:

阻塞队列类JDK版本有界性
ArrayBlockingQueue5有界(bounded)
LinkedBlockingQueue5有界(optionally-bounded)
PriorityBlockingQueue5无界(unbounded)
DelayQueue5无界(unbounded)
SynchronousQueue5-
LinkedBlockingDeque6有界(optionally-bounded)
LinkedTransferQueue7无界(unbounded)

除了阻塞队列外,JUC包下还有一个非阻塞队列:ConcurrentLinkedQueue,该类不在本节讨论之中。

二、阻塞队列概述

(一)阻塞队列的抽象定义:BlockingQueue

BlockingQueue定义了阻塞队列的抽象结构,所有的阻塞队列都需要实现它。BlockingQueue的核心方法如下所示:

方法抛出异常返回特殊值阻塞超时阻塞
插入add(e)offer(e)put(e)offer(e, time, unit)
移除remove()poll()take()poll(time, unit)
检查element()peek()--

在对队列执行以下三种操作时:

  1. 在队列满时插入元素:使用addofferput插入方法
  2. 在队列空时移除元素:使用removepolltake移除方法
  3. 在队列空时检查元素:使用elementpeek检查方法

一定会执行失败,此时根据使用方法的不同,可能会出现以下四种情况:

  1. 抛出异常
  2. 返回特殊值(如null或false)
  3. 阻塞
  4. 超时阻塞

其中,只有阻塞超时阻塞的4个方法是BlockingQueue接口独有的(这也是阻塞队列需要重点关注的方法),而抛出异常返回特殊值的6个方法则是继承自Queue接口。后续讲到的几种具体的阻塞队列实现,都需要实现上述方法。

(二)常规阻塞队列:ArrayBlockingQueue与LinkedBlockingQueue

ArrayBlockingQueue是使用数组实现的有界(bounded)阻塞队列,LinkedBlockingQueue是使用链表实现的有界(optionally-bounded)阻塞队列。它们都符合普通队列的FIFO(first-in-first-out,先入先出)原则:新增元素时在队尾插入,移除元素时从队首删除,队首的元素存在时间最长,队尾元素存在时间最短。

为了实现阻塞插入或阻塞移除,ArrayBlockingQueueLinkedBlockingQueue均使用ReentrantLock锁来实现同步,当队列满时或队列空时,执行阻塞的插入或移除操作(如:puttake),会导致持有锁的线程将进入自旋等待状态,而其它线程因为无法获取到锁,也将进入等待状态,直到队列非满或非空时,等待状态才可以解除。

其中,ArrayBlockingQueue只有一个锁,可以通过构造函数的fair参数来指定该锁的公平性。而LinkedBlockingQueue则有两个锁:takeLockputLock,它们都是默认的非公平锁,takeLock可以由takepoll等方法持有,putLock可以由putoffe等方法持有。

LinkedBlockingQueue可以用来实现FixedThreadPool线程池(详见ExecutorsnewFixedThreadPool(int nThreads)方法)。

在大多数并发应用中,LinkedBlockingQueueArrayBlockingQueue有更高的吞吐量,但有更低的性能可预测性(Linked queues typically have higher throughput than array-based queues but less predictable performance in most concurrent applications.)

(三)优先阻塞队列:PriorityBlockingQueue与DelayQueue

PriorityBlockingQueueDelayQueue都是使用PriorityQueue实现的无界(unbounded)阻塞队列,它们都使用ReentrantLock锁来实现同步。

其中,PriorityBlockingQueue可以通过Comparator来自定义队列元素的优先级。DelayQueue通过Delayed泛型元素实现延时获取元素。DelayQueue可以用来设计缓存系统与定时任务调度。

(四)无容量阻塞队列:SynchronousQueue

SynchronousQueue是一个没有任何容量的阻塞队列(其capacity值为0),每一个插入操作都必须等待另一个线程对应的移除操作,反之亦然。SynchronousQueue使用ReentrantLock锁来实现同步,可以通过fair参数来指定该锁的公平性。

SynchronousQueue可以用来实现CachedThreadPool线程池(详见ExecutorsnewCachedThreadPool()方法)。

(五)双向阻塞队列:BlockingDeque与LinkedBlockingDeque

LinkedBlockingDeque是使用双向链表实现的有界(optionally-bounded)阻塞队列,它实现了BlockingDeque接口。LinkedBlockingDeque使用ReentrantLock锁来实现同步。

LinkedBlockingDeque可以用来实现工作窃取模式(即ForkJoin系列类)。

(六)Transfer阻塞队列:TransferQueue与LinkedTransferQueue

LinkedTransferQueue是使用链表实现的无界(unbounded)阻塞队列,它实现了TransferQueue接口。

可以将LinkedTransferQueue看作是LinkedBlockingQueueSynchronousQueue的结合。LinkedTransferQueuetransfer方法会优先将生产者传入的元素立刻传输给消费者,此过程类似于SynchronousQueue。如果没有对应的消费者生产者传入的元素,会将之放到队列中,此过程类似于LinkedTransferQueue

三、总结

JUC包下定义了10个阻塞队列相关的接口或类。简单将它们分下类如下所示:

  • BlockingQueue作为顶层接口定义了阻塞队列的阻塞队列
  • ArrayBlockingQueueLinkedBlockingQueue是两个最基础的阻塞队列实现
  • PriorityBlockingQueueDelayQueue是实现了优先级与延时获取的阻塞队列
  • SynchronousQueue是不存储元素并直接传递元素的阻塞队列
  • LinkedBlockingDeque是实现了BlockingDeque接口的双向阻塞队列
  • LinkedTransferQueue是实现了TransferQueue接口的阻塞队列,可以将其看作LinkedBlockingQueueSynchronousQueue的结合。

本文更偏向于阻塞队列基础概念的简介,很多内容只是一笔带过,没有详细介绍,由于理解有限,上述内容可能会有偏差,后续将按需修改或增加相关内容。

锁是阻塞队列的基础,上面提到的大部分阻塞队列都是使用ReentrantLock锁来实现同步。同时,阻塞队列又是线程池的基础,有关线程池的内容,将在后续介绍。

参考文档

  1. 《Java并发编程的艺术 方腾飞 魏鹏 程晓明 著》