解读 Java 并发队列 BlockingQueue

太过爱你忘了你带给我的痛 2022-01-20 08:27 278阅读 0赞

最近得空，想写篇文章好好说说 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 对插入操作、移除操作、获取元素操作提供了四种不同的方法用于不同的场景中使用：1、抛出异常；2、返回特殊值（null 或 true/false，取决于具体的操作）；3、阻塞等待此操作，直到这个操作成功；4、阻塞等待此操作，直到成功或者超时指定时间。总结如下：

<table> 
 <thead> 
 <tr> 
 <th>&nbsp;</th> 
 <th>Throws exception</th> 
 <th>Special value</th> 
 <th>Blocks</th> 
 <th>Times out</th> 
 </tr> 
 </thead> 
 <tbody> 
 <tr> 
 <td>Insert</td> 
 <td>add(e)</td> 
 <td>offer(e)</td> 
 <td>put(e)</td> 
 <td>offer(e, time, unit)</td> 
 </tr> 
 <tr> 
 <td>Remove</td> 
 <td>remove()</td> 
 <td>poll()</td> 
 <td>take()</td> 
 <td>poll(time, unit)</td> 
 </tr> 
 <tr> 
 <td>Examine</td> 
 <td>element()</td> 
 <td>peek()</td> 
 <td>not applicable</td> 
 <td>not applicable</td> 
 </tr> 
 </tbody> 
</table>

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（二）][AbstractQueuedSynchronizer]》 的关于 Condition 的文章的话，那么你一定能很容易看懂 ArrayBlockingQueue 的源码，它采用一个 ReentrantLock 和相应的两个 Condition 来实现。

ArrayBlockingQueue 共有以下几个属性：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 </td> 
 <td> <code>// 用于存放元素的数组</code> <code>final</code> <code>Object[] items;</code> <code>// 下一次读取操作的位置</code> <code>int</code> <code>takeIndex;</code> <code>// 下一次写入操作的位置</code> <code>int</code> <code>putIndex;</code> <code>// 队列中的元素数量</code> <code>int</code> <code>count;</code> &nbsp; <code>// 以下几个就是控制并发用的同步器</code> <code>final</code> <code>ReentrantLock lock;</code> <code>private</code> <code>final</code> <code>Condition notEmpty;</code> <code>private</code> <code>final</code> <code>Condition notFull;</code> </td> 
 </tr> 
 </tbody> 
</table>

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

![array-blocking-queue][]

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

对于 ArrayBlockingQueue，我们可以在构造的时候指定以下三个参数：

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

更具体的源码我就不进行分析了，因为它就是 AbstractQueuedSynchronizer 中 Condition 的使用，感兴趣的读者请看我写的《[一行一行源码分析清楚 AbstractQueuedSynchronizer（二）][AbstractQueuedSynchronizer 1]》，因为只要看懂了那篇文章，ArrayBlockingQueue 的代码就没有分析的必要了，当然，如果你完全不懂 Condition，那么基本上也就可以说看不懂 ArrayBlockingQueue 的源码了。

## BlockingQueue 实现之 LinkedBlockingQueue ##

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

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 </td> 
 <td> <code>// 传说中的无界队列</code> <code>public</code> <code>LinkedBlockingQueue() { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>this</code><code>(Integer.MAX_VALUE);</code> <code>}</code> <code>// 传说中的有界队列</code> <code>public</code> <code>LinkedBlockingQueue(</code><code>int</code> <code>capacity) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code> <code>(capacity &lt;= </code><code>0</code><code>) </code><code>throw</code> <code>new</code> <code>IllegalArgumentException();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>this</code><code>.capacity = capacity;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>last = head = </code><code>new</code> <code>Node&lt;E&gt;(</code><code>null</code><code>);</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

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

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

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

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

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

![linked-blocking-queue][]

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

先上构造方法：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 </td> 
 <td> <code>public</code> <code>LinkedBlockingQueue(</code><code>int</code> <code>capacity) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code> <code>(capacity &lt;= </code><code>0</code><code>) </code><code>throw</code> <code>new</code> <code>IllegalArgumentException();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>this</code><code>.capacity = capacity;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>last = head = </code><code>new</code> <code>Node&lt;E&gt;(</code><code>null</code><code>);</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

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

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

我们再看看 take 方法：

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

## BlockingQueue 实现之 SynchronousQueue ##

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

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

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

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

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

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

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 </td> 
 <td> <code>// 构造时，我们可以指定公平模式还是非公平模式，区别之后再说</code> <code>public</code> <code>SynchronousQueue(</code><code>boolean</code> <code>fair) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>transferer = fair ? </code><code>new</code> <code>TransferQueue() : </code><code>new</code> <code>TransferStack();</code> <code>}</code> <code>abstract</code> <code>static</code> <code>class</code> <code>Transferer { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 从方法名上大概就知道，这个方法用于转移元素，从生产者手上转到消费者手上</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 也可以被动地，消费者调用这个方法来从生产者手上取元素</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 第一个参数 e 如果不是 null，代表场景为：将元素从生产者转移给消费者</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 如果是 null，代表消费者等待生产者提供元素，然后返回值就是相应的生产者提供的元素</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 第二个参数代表是否设置超时，如果设置超时，超时时间是第三个参数的值</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 返回值如果是 null，代表超时，或者中断。具体是哪个，可以通过检测中断状态得到。</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>abstract</code> <code>Object transfer(Object e, </code><code>boolean</code> <code>timed, </code><code>long</code> <code>nanos);</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

Transferer 有两个内部实现类，是因为构造 SynchronousQueue 的时候，我们可以指定公平策略。公平模式意味着，所有的读写线程都遵守先来后到，FIFO 嘛，对应 TransferQueue。而非公平模式则对应 TransferStack。

![synchronous-queue][]

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

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

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 </td> 
 <td> <code>// 写入值</code> <code>public</code> <code>void</code> <code>put(E o) </code><code>throws</code> <code>InterruptedException { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code> <code>(o == </code><code>null</code><code>) </code><code>throw</code> <code>new</code> <code>NullPointerException();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code> <code>(transferer.transfer(o, </code><code>false</code><code>, </code><code>0</code><code>) == </code><code>null</code><code>) { </code><code>// 1</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>Thread.interrupted();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>throw</code> <code>new</code> <code>InterruptedException();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>}</code> <code>// 读取值并移除</code> <code>public</code> <code>E take() </code><code>throws</code> <code>InterruptedException { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>Object e = transferer.transfer(</code><code>null</code><code>, </code><code>false</code><code>, </code><code>0</code><code>); </code><code>// 2</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code> <code>(e != </code><code>null</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code> <code>(E)e;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>Thread.interrupted();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>throw</code> <code>new</code> <code>InterruptedException();</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

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

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

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

其实这里有个隐含的条件被满足了，队列如果不为空，肯定都是同种类型的节点，要么都是读操作，要么都是写操作。这个就要看到底是读线程积压了，还是写线程积压了。

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

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

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 </td> 
 <td> <code>static</code> <code>final</code> <code>class</code> <code>QNode { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>volatile</code> <code>QNode next;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>// 可以看出来，等待队列是单向链表</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>volatile</code> <code>Object item;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>// CAS'ed to or from null</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>volatile</code> <code>Thread waiter;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>// 将线程对象保存在这里，用于挂起和唤醒</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>final</code> <code>boolean</code> <code>isData;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>// 用于判断是写线程节点(isData == true)，还是读线程节点</code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>QNode(Object item, </code><code>boolean</code> <code>isData) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>this</code><code>.item = item;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>this</code><code>.isData = isData;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;</code><code>......</code> </td> 
 </tr> 
 </tbody> 
</table>

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

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 </td> 
 <td> <code>/**</code> <code>&nbsp;</code><code>* Puts or takes an item.</code> <code>&nbsp;</code><code>*/</code> <code>Object transfer(Object e, </code><code>boolean</code> <code>timed, </code><code>long</code> <code>nanos) { </code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>QNode s = </code><code>null</code><code>; </code><code>// constructed/reused as needed</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>boolean</code> <code>isData = (e != </code><code>null</code><code>);</code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>for</code> <code>(;;) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>QNode t = tail;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>QNode h = head;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code> <code>(t == </code><code>null</code> <code>|| h == </code><code>null</code><code>)&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>// saw uninitialized value</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>continue</code><code>;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>// spin</code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 队列空，或队列中节点类型和当前节点一致，</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 即我们说的第一种情况，将节点入队即可。读者要想着这块 if 里面方法其实就是入队</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code> <code>(h == t || t.isData == isData) { </code><code>// empty or same-mode</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>QNode tn = t.next;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// t != tail 说明刚刚有节点入队，continue 即可</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code> <code>(t != tail)&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>// inconsistent read</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>continue</code><code>;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 有其他节点入队，但是 tail 还是指向原来的，此时设置 tail 即可</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code> <code>(tn != </code><code>null</code><code>) {&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>// lagging tail</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 这个方法就是：如果 tail 此时为 t 的话，设置为 tn</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>advanceTail(t, tn);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>continue</code><code>;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code> <code>(timed &amp;&amp; nanos &lt;= </code><code>0</code><code>)&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>// can't wait</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code> <code>null</code><code>;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code> <code>(s == </code><code>null</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>s = </code><code>new</code> <code>QNode(e, isData);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 将当前节点，插入到 tail 的后面</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code> <code>(!t.casNext(</code><code>null</code><code>, s))&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>// failed to link in</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>continue</code><code>;</code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 将当前节点设置为新的 tail</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>advanceTail(t, s);&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>// swing tail and wait</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 看到这里，请读者先往下滑到这个方法，看完了以后再回来这里，思路也就不会断了</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>Object x = awaitFulfill(s, e, timed, nanos);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 到这里，说明之前入队的线程被唤醒了，准备往下执行</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code> <code>(x == s) {&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>// wait was cancelled</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>clean(t, s);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code> <code>null</code><code>;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code> <code>(!s.isOffList()) {&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>// not already unlinked</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>advanceHead(t, s);&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>// unlink if head</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code> <code>(x != </code><code>null</code><code>)&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>// and forget fields</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>s.item = s;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>s.waiter = </code><code>null</code><code>;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code> <code>(x != </code><code>null</code><code>) ? x : e;</code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 这里的 else 分支就是上面说的第二种情况，有相应的读或写相匹配的情况</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>} </code><code>else</code> <code>{&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>// complementary-mode</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>QNode m = h.next;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>// node to fulfill</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code> <code>(t != tail || m == </code><code>null</code> <code>|| h != head)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>continue</code><code>;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>// inconsistent read</code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>Object x = m.item;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code> <code>(isData == (x != </code><code>null</code><code>) ||&nbsp;&nbsp;&nbsp; </code><code>// m already fulfilled</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>x == m ||&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>// m cancelled</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>!m.casItem(x, e)) {&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>// lost CAS</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>advanceHead(h, m);&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>// dequeue and retry</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>continue</code><code>;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>advanceHead(h, m);&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>// successfully fulfilled</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>LockSupport.unpark(m.waiter);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code> <code>(x != </code><code>null</code><code>) ? x : e;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>}</code> &nbsp; <code>void</code> <code>advanceTail(QNode t, QNode nt) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code> <code>(tail == t)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>UNSAFE.compareAndSwapObject(</code><code>this</code><code>, tailOffset, t, nt);</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

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 方法在队列为空的时候会阻塞）。

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

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

此类实现了 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\]，其堆序性质为，每个节点的值都小于其左右子节点的值。二叉堆中最小的值就是根节点，但是删除根节点是比较麻烦的，因为需要调整树。

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

![priority-blocking-queue-1][]

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

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

扩容方法对并发的控制也非常的巧妙，释放了原来的独占锁 lock，这样的话，扩容操作和读操作可以同时进行，提高吞吐量。

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

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

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 </td> 
 <td> <code>// 这个方法就是将数据 x 插入到数组 array 的位置 k 处，然后再调整树</code> <code>private</code> <code>static</code> <code>&lt;T&gt; </code><code>void</code> <code>siftUpComparable(</code><code>int</code> <code>k, T x, Object[] array) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>Comparable&lt;? </code><code>super</code> <code>T&gt; key = (Comparable&lt;? </code><code>super</code> <code>T&gt;) x;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>while</code> <code>(k &gt; </code><code>0</code><code>) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 二叉堆中 a[k] 节点的父节点位置</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>int</code> <code>parent = (k - </code><code>1</code><code>) &gt;&gt;&gt; </code><code>1</code><code>;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>Object e = array[parent];</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code> <code>(key.compareTo((T) e) &gt;= </code><code>0</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>break</code><code>;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>array[k] = e;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>k = parent;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>array[k] = key;</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

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

![priority-blocking-queue-2][]

我们再看看 take 方法：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 </td> 
 <td> <code>public</code> <code>E take() </code><code>throws</code> <code>InterruptedException { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>final</code> <code>ReentrantLock lock = </code><code>this</code><code>.lock;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 独占锁</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>lock.lockInterruptibly();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>E result;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>try</code> <code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// dequeue 出队</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>while</code> <code>( (result = dequeue()) == </code><code>null</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>notEmpty.await();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>} </code><code>finally</code> <code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>lock.unlock();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code> <code>result;</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 </td> 
 <td> <code>private</code> <code>E dequeue() { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>int</code> <code>n = size - </code><code>1</code><code>;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code> <code>(n &lt; </code><code>0</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code> <code>null</code><code>;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>else</code> <code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>Object[] array = queue;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 队头，用于返回</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>E result = (E) array[</code><code>0</code><code>];</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 队尾元素先取出</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>E x = (E) array[n];</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 队尾置空</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>array[n] = </code><code>null</code><code>;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>Comparator&lt;? </code><code>super</code> <code>E&gt; cmp = comparator;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code> <code>(cmp == </code><code>null</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>siftDownComparable(</code><code>0</code><code>, x, array, n);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>else</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>siftDownUsingComparator(</code><code>0</code><code>, x, array, n, cmp);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>size = n;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code> <code>result;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

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

废话不多说，出队是非常简单的，因为队头就是最小的元素，对应的是数组的第一个元素。难点是队头出队后，需要调整树。

![priority-blocking-queue-3][]

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

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

![priority-blocking-queue-4][]

好了， PriorityBlockingQueue 我们也说完了。

## 总结 ##

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

ArrayBlockingQueue 底层是数组，有界队列，如果我们要使用生产者-消费者模式，这是非常好的选择。

LinkedBlockingQueue 底层是链表，可以当做无界和有界队列来使用，所以大家不要以为它就是无界队列。

SynchronousQueue 本身不带有空间来存储任何元素，使用上可以选择公平模式和非公平模式。

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

\----------------------------------------------------------

作者：Re

来源：ImportNew

链接：[http://www.importnew.com/28053.html][http_www.importnew.com_28053.html]

[AbstractQueuedSynchronizer]: https://javadoop.com/post/AbstractQueuedSynchronizer-2
[array-blocking-queue]: /images/20220120/91d5fcc6eedd4b7db3b9dff8c88f2cad.png
[AbstractQueuedSynchronizer 1]: https://javadoop.com/post/AbstractQueuedSynchronizer-2/
[linked-blocking-queue]: /images/20220120/525c45dfbf3e487dbf797470f4c55920.png
[synchronous-queue]: /images/20220120/1e1c3c842db2454096f24d6b6fc41ac5.png
[priority-blocking-queue-1]: /images/20220120/752b03d8de734e2e8eef6695baf757b7.png
[priority-blocking-queue-2]: /images/20220120/2eea23e8b74c4a13ada8a96f42f1b0bd.png
[priority-blocking-queue-3]: /images/20220120/652543879b4441afa17260754f13d9da.png
[priority-blocking-queue-4]: /images/20220120/89e089287e3c4c189ae45cb0eb316c54.png
[http_www.importnew.com_28053.html]: http://www.importnew.com/28053.html