解读 java 并发队列 BlockingQueue

 

点击添加图片描述（最多60个字）编辑

 

 

今天呢！灯塔君跟你们讲：

解读 java 并发队列 BlockingQueue

最近得空，想写篇文章好好说说 java 线程池问题，我相信不少人都只知其一;不知其二的，包括我本身在仔仔细细看源码以前，也有许多的不解，甚至有些地方我一直都没有理解到位。

说到线程池实现，那么就不得不涉及到各类 BlockingQueue 的实现，那么我想就 BlockingQueue 的问题和你们分享分享我了解的一些知识。

本文没有像以前分析 AQS 那样一行一行源码分析了，不过仍是把其中最重要和最难理解的代码说了一遍，因此难免篇幅略长。本文涉及到比较多的 Doug Lea 对 BlockingQueue 的设计思想，但愿有心的读者真的能够有一些收获，我以为本身仍是写了一些干货的。

本文直接参考 Doug Lea 写的 Java doc 和注释，这也是咱们在学习 java 并发包时最好的材料了。但愿你们能有所思、有所悟，学习 Doug Lea 的代码风格，并将其优雅、严谨的做风应用到咱们写的每一行代码中。

目录：

BlockingQueue

开篇先介绍下 BlockingQueue 这个接口的规则，后面再看其实现。

首先，最基本的来讲， BlockingQueue 是一个先进先出的队列（Queue），为何说是阻塞（Blocking）的呢？是由于 BlockingQueue 支持当获取队列元素可是队列为空时，会阻塞等待队列中有元素再返回；也支持添加元素时，若是队列已满，那么等到队列能够放入新元素时再放入。

BlockingQueue 是一个接口，继承自 Queue，因此其实现类也能够做为 Queue 的实现来使用，而 Queue 又继承自 Collection 接口。

BlockingQueue 对插入操做、移除操做、获取元素操做提供了四种不一样的方法用于不一样的场景中使用：一、抛出异常；二、返回特殊值（null 或 true/false，取决于具体的操做）；三、阻塞等待此操做，直到这个操做成功；四、阻塞等待此操做，直到成功或者超时指定时间。总结以下：

 

点击添加图片描述（最多60个字）编辑

 

BlockingQueue 的各个实现都遵循了这些规则，固然咱们也不用死记这个表格，知道有这么回事，而后写代码的时候根据本身的须要去看方法的注释来选取合适的方法便可。

对于 BlockingQueue，咱们的关注点应该在 put(e) 和 take() 这两个方法，由于这两个方法是带阻塞的。

BlockingQueue 不接受 null 值的插入，相应的方法在碰到 null 的插入时会抛出 NullPointerException 异常。null 值在这里一般用于做为特殊值返回（表格中的第三列），表明 poll 失败。因此，若是容许插入 null 值的话，那获取的时候，就不能很好地用 null 来判断究竟是表明失败，仍是获取的值就是 null 值。

一个 BlockingQueue 多是有界的，若是在插入的时候，发现队列满了，那么 put 操做将会阻塞。一般，在这里咱们说的无界队列也不是说真正的无界，而是它的容量是 Integer.MAX_VALUE（21亿多）。

BlockingQueue 是设计用来实现生产者-消费者队列的，固然，你也能够将它当作普通的 Collection 来用，前面说了，它实现了 java.util.Collection 接口。例如，咱们能够用 remove(x) 来删除任意一个元素，可是，这类操做一般并不高效，因此尽可能只在少数的场合使用，好比一条消息已经入队，可是须要作取消操做的时候。

BlockingQueue 的实现都是线程安全的，可是批量的集合操做如 addAll, containsAll, retainAll 和 removeAll  不必定是原子操做。如 addAll(c) 有可能在添加了一些元素后中途抛出异常，此时 BlockingQueue 中已经添加了部分元素，这个是容许的，取决于具体的实现。

BlockingQueue 不支持 close 或 shutdown 等关闭操做，由于开发者可能但愿不会有新的元素添加进去，此特性取决于具体的实现，不作强制约束。

最后，BlockingQueue 在生产者-消费者的场景中，是支持多消费者和多生产者的，说的其实就是线程安全问题。

相信上面说的每一句都很清楚了，BlockingQueue 是一个比较简单的线程安全容器，下面我会分析其具体的在 JDK 中的实现，这里又到了 Doug Lea 表演时间了。

BlockingQueue 实现之 ArrayBlockingQueue

ArrayBlockingQueue 是 BlockingQueue 接口的有界队列实现类，底层采用数组来实现。

其并发控制采用可重入锁来控制，无论是插入操做仍是读取操做，都须要获取到锁才能进行操做。

若是读者看过我以前写的《一行一行源码分析清楚 AbstractQueuedSynchronizer（二）》 的关于 Condition 的文章的话，那么你必定能很容易看懂 ArrayBlockingQueue  的源码，它采用一个 ReentrantLock 和相应的两个 Condition 来实现。

ArrayBlockingQueue 共有如下几个属性：

点击添加图片描述（最多60个字）编辑

 

咱们用个示意图来描述其同步机制：

点击添加图片描述（最多60个字）编辑

 

ArrayBlockingQueue 实现并发同步的原理就是，读操做和写操做都须要获取到 AQS 独占锁才能进行操做。若是队列为空，这个时候读操做的线程进入到读线程队列排队，等待写线程写入新的元素，而后唤醒读线程队列的第一个等待线程。若是队列已满，这个时候写操做的线程进入到写线程队列排队，等待读线程将队列元素移除腾出空间，而后唤醒写线程队列的第一个等待线程。

对于 ArrayBlockingQueue，咱们能够在构造的时候指定如下三个参数：

1. 队列容量，其限制了队列中最多容许的元素个数；
2. 指定独占锁是公平锁仍是非公平锁。非公平锁的吞吐量比较高，公平锁能够保证每次都是等待最久的线程获取到锁；
3. 能够指定用一个集合来初始化，将此集合中的元素在构造方法期间就先添加到队列中。

更具体的源码我就不进行分析了，由于它就是 AbstractQueuedSynchronizer 中 Condition 的使用，感兴趣的读者请看我写的《一行一行源码分析清楚 AbstractQueuedSynchronizer（二）》，由于只要看懂了那篇文章，ArrayBlockingQueue 的代码就没有分析的必要了，固然，若是你彻底不懂 Condition，那么基本上也就能够说看不懂 ArrayBlockingQueue 的源码了。

BlockingQueue 实现之 LinkedBlockingQueue

底层基于单向链表实现的阻塞队列，能够当作无界队列也能够当作有界队列来使用。看构造方法：

 

点击添加图片描述（最多60个字）编辑

 

 

点击添加图片描述（最多60个字）编辑

咱们看看这个类有哪些属性：

点击添加图片描述（最多60个字）编辑

这里用了两个锁，两个 Condition，简单介绍以下：

takeLock 和 notEmpty 怎么搭配：若是要获取（take）一个元素，须要获取 takeLock 锁，可是获取了锁还不够，若是队列此时为空，还须要队列不为空（notEmpty）这个条件（Condition）。

putLock 须要和 notFull 搭配：若是要插入（put）一个元素，须要获取 putLock 锁，可是获取了锁还不够，若是队列此时已满，还须要队列不是满的（notFull）这个条件（Condition）。

首先，这里用一个示意图来看看 LinkedBlockingQueue 的并发读写控制，而后再开始分析源码：

点击添加图片描述（最多60个字）编辑

 

看懂这个示意图，源码也就简单了，读操做是排好队的，写操做也是排好队的，惟一的并发问题在于一个写操做和一个读操做同时进行，只要控制好这个就能够了。

先上构造方法：

点击添加图片描述（最多60个字）编辑

注意，这里会初始化一个空的头结点，那么第一个元素入队的时候，队列中就会有两个元素。读取元素时，也老是获取头节点后面的一个节点。count 的计数值不包括这个头节点。

咱们来看下 put 方法是怎么将元素插入到队尾的：

点击添加图片描述（最多60个字）编辑

咱们再看看 take 方法：

 

点击添加图片描述（最多60个字）编辑

源码分析就到这里结束了吧，毕竟仍是比较简单的源码，基本上只要读者认真点都看得懂。

BlockingQueue 实现之 SynchronousQueue

它是一个特殊的队列，它的名字其实就蕴含了它的特征 - - 同步的队列。为何说是同步的呢？这里说的并非多线程的并发问题，而是由于当一个线程往队列中写入一个元素时，写入操做不会当即返回，须要等待另外一个线程来将这个元素拿走；同理，当一个读线程作读操做的时候，一样须要一个相匹配的写线程的写操做。这里的 Synchronous 指的就是读线程和写线程须要同步，一个读线程匹配一个写线程。

咱们比较少使用到 SynchronousQueue 这个类，不过它在线程池的实现类 ThreadPoolExecutor 中获得了应用，感兴趣的读者能够在看完这个后去看看相应的使用。

虽然上面我说了队列，可是 SynchronousQueue 的队列实际上是虚的，其不提供任何空间（一个都没有）来存储元素。数据必须从某个写线程交给某个读线程，而不是写到某个队列中等待被消费。

你不能在 SynchronousQueue 中使用 peek 方法（在这里这个方法直接返回 null），peek 方法的语义是只读取不移除，显然，这个方法的语义是不符合 SynchronousQueue 的特征的。SynchronousQueue 也不能被迭代，由于根本就没有元素能够拿来迭代的。虽然 SynchronousQueue 间接地实现了 Collection 接口，可是若是你将其当作 Collection 来用的话，那么集合是空的。固然，这个类也是不容许传递 null 值的（并发包中的容器类好像都不支持插入 null 值，由于 null 值每每用做其余用途，好比用于方法的返回值表明操做失败）。

接下来，咱们来看看具体的源码实现吧，它的源码不是很简单的那种，咱们须要先搞清楚它的设计思想。

源码加注释大概有 1200 行，咱们先看大框架：

点击添加图片描述（最多60个字）编辑

 

Transferer 有两个内部实现类，是由于构造 SynchronousQueue 的时候，咱们能够指定公平策略。公平模式意味着，全部的读写线程都遵照先来后到，FIFO 嘛，对应 TransferQueue。而非公平模式则对应 TransferStack。

点击添加图片描述（最多60个字）编辑

咱们先采用公平模式分析源码，而后再说说公平模式和非公平模式的区别。

接下来，咱们看看 put 方法和 take 方法：

 

点击添加图片描述（最多60个字）编辑

咱们看到，写操做 put(E o) 和读操做 take() 都是调用 Transferer.transfer(…) 方法，区别在于第一个参数是否为 null 值。

咱们来看看 transfer 的设计思路，其基本算法以下：

1. 当调用这个方法时，若是队列是空的，或者队列中的节点和当前的线程操做类型一致（如当前操做是 put 操做，而队列中的元素也都是写线程）。这种状况下，将当前线程加入到等待队列便可。
2. 若是队列中有等待节点，并且与当前操做能够匹配（如队列中都是读操做线程，当前线程是写操做线程，反之亦然）。这种状况下，匹配等待队列的队头，出队，返回相应数据。

其实这里有个隐含的条件被知足了，队列若是不为空，确定都是同种类型的节点，要么都是读操做，要么都是写操做。这个就要看究竟是读线程积压了，仍是写线程积压了。

咱们能够假设出一个男女配对的场景：一个男的过来，若是一我的都没有，那么他须要等待；若是发现有一堆男的在等待，那么他须要排到队列后面；若是发现是一堆女的在排队，那么他直接牵走队头的那个女的。

既然这里说到了等待队列，咱们先看看其实现，也就是 QNode:

点击添加图片描述（最多60个字）编辑

相信说了这么多之后，咱们再来看 transfer 方法的代码就轻松多了。

点击添加图片描述（最多60个字）编辑

 

点击添加图片描述（最多60个字）编辑

 

Doug Lea 的巧妙之处在于，将各个代码凑在了一块儿，使得代码很是简洁，固然也同时增长了咱们的阅读负担，看代码的时候，仍是得仔细想一想各类可能的状况。

下面，再说说前面说的公平模式和非公平模式的区别。

相信你们内心面已经有了公平模式的工做流程的概念了，我就简单说说 TransferStack 的算法，就不分析源码了。

1. 当调用这个方法时，若是队列是空的，或者队列中的节点和当前的线程操做类型一致（如当前操做是 put 操做，而栈中的元素也都是写线程）。这种状况下，将当前线程加入到等待栈中，等待配对。而后返回相应的元素，或者若是被取消了的话，返回 null。
2. 若是栈中有等待节点，并且与当前操做能够匹配（如栈里面都是读操做线程，当前线程是写操做线程，反之亦然）。将当前节点压入栈顶，和栈中的节点进行匹配，而后将这两个节点出栈。配对和出栈的动做其实也不是必须的，由于下面的一条会执行一样的事情。
3. 若是栈顶是进行匹配而入栈的节点，帮助其进行匹配并出栈，而后再继续操做。

应该说，TransferStack 的源码要比 TransferQueue 的复杂一些，若是读者感兴趣，请自行进行源码阅读。

BlockingQueue 实现之 PriorityBlockingQueue

带排序的 BlockingQueue 实现，其并发控制采用的是 ReentrantLock，队列为无界队列（ArrayBlockingQueue 是有界队列，LinkedBlockingQueue 也能够经过在构造函数中传入 capacity 指定队列最大的容量，可是 PriorityBlockingQueue 只能指定初始的队列大小，后面插入元素的时候，若是空间不够的话会自动扩容）。

简单地说，它就是 PriorityQueue 的线程安全版本。不能够插入 null 值，同时，插入队列的对象必须是可比较大小的（comparable），不然报 ClassCastException 异常。它的插入操做 put 方法不会 block，由于它是无界队列（take 方法在队列为空的时候会阻塞）。

它的源码相对比较简单，本节将介绍其核心源码部分。

咱们来看看它有哪些属性：

点击添加图片描述（最多60个字）编辑

 

此类实现了 Collection 和 Iterator 接口中的全部接口方法，对其对象进行迭代并遍历时，不能保证有序性。若是你想要实现有序遍历，建议采用 Arrays.sort(queue.toArray()) 进行处理。PriorityBlockingQueue 提供了 drainTo 方法用于将部分或所有元素有序地填充（准确说是转移，会删除原队列中的元素）到另外一个集合中。还有一个须要说明的是，若是两个对象的优先级相同（compare 方法返回 0），此队列并不保证它们之间的顺序。

PriorityBlockingQueue 使用了基于数组的二叉堆来存放元素，全部的 public 方法采用同一个 lock 进行并发控制。

二叉堆：一颗彻底二叉树，它很是适合用数组进行存储，对于数组中的元素 a[i]，其左子节点为 a[2*i+1]，其右子节点为 a[2*i + 2]，其父节点为 a[(i-1)/2]，其堆序性质为，每一个节点的值都小于其左右子节点的值。二叉堆中最小的值就是根节点，可是删除根节点是比较麻烦的，由于须要调整树。

简单用个图解释一下二叉堆，我就不说太多专业的严谨的术语了，这种数据结构的优势是一目了然的，最小的元素必定是根元素，它是一棵满的树，除了最后一层，最后一层的节点从左到右紧密排列。

点击添加图片描述（最多60个字）编辑

下面开始 PriorityBlockingQueue 的源码分析，首先咱们来看看构造方法:

 

点击添加图片描述（最多60个字）编辑

 

接下来，咱们来看看其内部的自动扩容实现：

 

 

点击添加图片描述（最多60个字）编辑

 

扩容方法对并发的控制也很是的巧妙，释放了原来的独占锁 lock，这样的话，扩容操做和读操做能够同时进行，提升吞吐量。

下面，咱们来分析下写操做 put 方法和读操做 take 方法。

点击添加图片描述（最多60个字）编辑

对于二叉堆而言，插入一个节点是简单的，插入的节点若是比父节点小，交换它们，而后继续和父节点比较。

 

点击添加图片描述（最多60个字）编辑

 

咱们用图来示意一下，咱们接下来要将 11 插入到队列中，看看 siftUp 是怎么操做的。

 

点击添加图片描述（最多60个字）编辑

 

 

咱们再看看 take 方法：

 

点击添加图片描述（最多60个字）编辑

dequeue 方法返回队头，并调整二叉堆的树，调用这个方法必须先获取独占锁。

废话很少说，出队是很是简单的，由于队头就是最小的元素，对应的是数组的第一个元素。难点是队头出队后，须要调整树。

点击添加图片描述（最多60个字）编辑

 

点击添加图片描述（最多60个字）编辑

 

记住二叉堆是一棵彻底二叉树，那么根节点 10 拿掉后，最后面的元素 17 必须找到合适的地方放置。首先，17 和 10 不能直接交换，那么先将根节点 10 的左右子节点中较小的节点往上滑，即 12 往上滑，而后原来 12 留下了一个空节点，而后再把这个空节点的较小的子节点往上滑，即 13 往上滑，最后，留出了位子，17 补上便可。

我稍微调整下这个树，以便读者能更明白：

点击添加图片描述（最多60个字）编辑

好了， PriorityBlockingQueue 咱们也说完了。

总结

我知道本文过长，相信一字不漏看完的读者确定是少数。

ArrayBlockingQueue 底层是数组，有界队列，若是咱们要使用生产者-消费者模式，这是很是好的选择。

LinkedBlockingQueue 底层是链表，能够当作无界和有界队列来使用，因此你们不要觉得它就是无界队列。

SynchronousQueue 自己不带有空间来存储任何元素，使用上能够选择公平模式和非公平模式。

PriorityBlockingQueue 是无界队列，基于数组，数据结构为二叉堆，数组第一个也是树的根节点老是最小值。