解读 Java 并发队列 BlockingQueue
转自：https://javadoop.com/post/java-concurrent-queue
最近得空，想写篇文章好好说说 java 线程池问题，我相信很多人都一知半解的，包括我自己在仔仔细细看源码之前，也有许多的不解，甚至有些地方我一直都没有理解到位。
说到线程池实现，那么就不得不涉及到各种 BlockingQueue 的实现，那么我想就 BlockingQueue 的问题和大家分享分享我了解的一些知识。
本文没有像之前分析 AQS 那样一行一行源码分析了，不过还是把其中最重要和最难理解的代码说了一遍，所以不免篇幅略长。本文涉及到比较多的 Doug Lea 对 BlockingQueue 的设计思想，希望有心的读者真的可以有一些收获，我觉得自己还是写了一些干货的。
本文直接参考 Doug Lea 写的 Java doc 和注释，这也是我们在学习 java 并发包时最好的材料了。希望大家能有所思、有所悟，学习 Doug Lea 的代码风格，并将其优雅、严谨的作风应用到我们写的每一行代码中。
目录
- 阻塞队列概览
- Java中的阻塞队列
- BlockingQueue源码分析
- BlockingQueue 实现之 ArrayBlockingQueue
- BlockingQueue 实现之 LinkedBlockingQueue
- BlockingQueue 实现之 SynchronousQueue
- BlockingQueue 实现之 PriorityBlockingQueue
- 总结
阻塞队列概览
1. 什么是阻塞队列？
阻塞队列（BlockingQueue）是一个支持两个附加操作的队列。这两个附加的操作是：在队列为空时，获取元素的线程会等待队列变为非空。当队列满时，存储元素的线程会等待队列可用。阻塞队列常用于生产者和消费者的场景，生产者是往队列里添加元素的线程，消费者是从队列里拿元素的线程。阻塞队列就是生产者存放元素的容器，而消费者也只从容器里拿元素。
阻塞队列提供了四种处理方法:
| 方法\处理方式 | 抛出异常 | 返回特殊值 | 一直阻塞 | 超时退出 |
|---|---|---|---|---|
| 插入方法 | add(e) | offer(e) | put(e) | offer(e,time,unit) |
| 移除方法 | remove() | poll() | take() | poll(time,unit) |
| 检查方法 | element() | peek() | 不可用 | 不可用 |
- 抛出异常：是指当阻塞队列满时候，再往队列里插入元素，会抛出IllegalStateException(“Queue full”)异常。当队列为空时，从队列里获取元素时会抛出NoSuchElementException异常 。
- 返回特殊值：插入方法会返回是否成功，成功则返回true。移除方法，则是从队列里拿出一个元素，如果没有则返回null
- 一直阻塞：当阻塞队列满时，如果生产者线程往队列里put元素，队列会一直阻塞生产者线程，直到拿到数据，或者响应中断退出。当队列空时，消费者线程试图从队列里take元素，队列也会阻塞消费者线程，直到队列可用。
- 超时退出：当阻塞队列满时，队列会阻塞生产者线程一段时间，如果超过一定的时间，生产者线程就会退出。
2、Java里的阻塞队列
JDK7提供了7个阻塞队列。分别是
- ArrayBlockingQueue ：一个由数组结构组成的有界阻塞队列。
- LinkedBlockingQueue ：一个由链表结构组成的有界阻塞队列。
- PriorityBlockingQueue ：一个支持优先级排序的无界阻塞队列。
- DelayQueue：一个使用优先级队列实现的无界阻塞队列。
- SynchronousQueue：一个不存储元素的阻塞队列。
- LinkedTransferQueue：一个由链表结构组成的无界阻塞队列。
- LinkedBlockingDeque：一个由链表结构组成的双向阻塞队列。
ArrayBlockingQueue
ArrayBlockingQueue是一个用数组实现的有界阻塞队列。此队列按照先进先出（FIFO）的原则对元素进行排序。默认情况下不保证访问者公平的访问队列，所谓公平访问队列是指阻塞的所有生产者线程或消费者线程，当队列可用时，可以按照阻塞的先后顺序访问队列，即先阻塞的生产者线程，可以先往队列里插入元素，先阻塞的消费者线程，可以先从队列里获取元素。通常情况下为了保证公平性会降低吞吐量。
LinkedBlockingQueue
LinkedBlockingQueue是一个用链表实现的有界阻塞队列。此队列的默认和最大长度为Integer.MAX_VALUE。此队列按照先进先出的原则对元素进行排序。
PriorityBlockingQueue
PriorityBlockingQueue是一个支持优先级的无界队列。默认情况下元素采取自然顺序排列，也可以通过比较器comparator来指定元素的排序规则。元素按照升序排列。
DelayQueue
DelayQueue是一个支持延时获取元素的无界阻塞队列。队列使用PriorityQueue来实现。队列中的元素必须实现Delayed接口，在创建元素时可以指定多久才能从队列中获取当前元素。只有在延迟期满时才能从队列中提取元素。我们可以将DelayQueue运用在以下应用场景：
- 缓存系统的设计：可以用DelayQueue保存缓存元素的有效期，使用一个线程循环查询DelayQueue，一旦能从DelayQueue中获取元素时，表示缓存有效期到了。
- 定时任务调度。使用DelayQueue保存当天将会执行的任务和执行时间，一旦从DelayQueue中获取到任务就开始执行，从比如TimerQueue就是使用DelayQueue实现的。
3、阻塞队列源码分析：
BlockingQueue
首先，最基本的来说， BlockingQueue 是一个先进先出的队列（Queue），为什么说是阻塞（Blocking）的呢？是因为 BlockingQueue 支持当获取队列元素但是队列为空时，会阻塞等待队列中有元素再返回；也支持添加元素时，如果队列已满，那么等到队列可以放入新元素时再放入。
BlockingQueue 是一个接口，继承自 Queue，所以其实现类也可以作为 Queue 的实现来使用，而 Queue 又继承自 Collection 接口。
BlockingQueue 对插入操作、移除操作、获取元素操作提供了四种不同的方法用于不同的场景中使用：1、抛出异常；2、返回特殊值（null 或 true/false，取决于具体的操作）；3、阻塞等待此操作，直到这个操作成功；4、阻塞等待此操作，直到成功或者超时指定时间。总结如下：
| Throws exception | Special value | Blocks | Times out | |
|---|---|---|---|---|
| Insert | add(e) | offer(e) | put(e) | offer(e, time, unit) |
| Remove | remove() | poll() | take() | poll(time, unit) |
| Examine | element() | peek() | not applicable | not applicable |
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 共有以下几个属性：
// 用于存放元素的数组
final Object[] items;
// 下一次读取操作的位置
int takeIndex;
// 下一次写入操作的位置
int putIndex;
// 队列中的元素数量
int count;// 以下几个就是控制并发用的同步器
final ReentrantLock lock;
private final Condition notEmpty;
private final Condition notFull;
我们用个示意图来描述其同步机制：
转存失败一行一行源码分析清楚 AbstractQueuedSynchronizer（二）》，因为只要看懂了那篇文章，ArrayBlockingQueue 的代码就没有分析的必要了，当然，如果你完全不懂 Condition，那么基本上也就可以说看不懂 ArrayBlockingQueue 的源码了。
BlockingQueue 实现之 LinkedBlockingQueue
底层基于单向链表实现的阻塞队列，可以当做无界队列也可以当做有界队列来使用。看构造方法：
// 传说中的无界队列
public LinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}
// 传说中的有界队列
public LinkedBlockingQueue(int capacity) {
if (capacity <= 0) throw new IllegalArgumentException();
this.capacity = capacity;
last = head = new Node<E>(null);
}
我们看看这个类有哪些属性：
// 队列容量
private final int capacity;// 队列中的元素数量
private final AtomicInteger count = new AtomicInteger(0);// 队头
private transient Node<E> head;// 队尾
private transient Node<E> last;// take, poll, peek 等读操作的方法需要获取到这个锁
private final ReentrantLock takeLock = new ReentrantLock();// 如果读操作的时候队列是空的，那么等待 notEmpty 条件
private final Condition notEmpty = takeLock.newCondition();// put, offer 等写操作的方法需要获取到这个锁
private final ReentrantLock putLock = new ReentrantLock();// 如果写操作的时候队列是满的，那么等待 notFull 条件
private final Condition notFull = putLock.newCondition();
这里用了两个锁，两个 Condition，简单介绍如下：
takeLock 和 notEmpty 怎么搭配：如果要获取（take）一个元素，需要获取 takeLock 锁，但是获取了锁还不够，如果队列此时为空，还需要队列不为空（notEmpty）这个条件（Condition）。
putLock 需要和 notFull 搭配：如果要插入（put）一个元素，需要获取 putLock 锁，但是获取了锁还不够，如果队列此时已满，还需要队列不是满的（notFull）这个条件（Condition）。
首先，这里用一个示意图来看看 LinkedBlockingQueue 的并发读写控制，然后再开始分析源码：
转存失败http://cmsblogs.com/ 『chenssy』
DelayQueue是一个支持延时获取元素的无界阻塞队列。里面的元素全部都是“可延期”的元素，列头的元素是最先“到期”的元素，如果队列里面没有元素到期，是不能从列头获取元素的，哪怕有元素也不行。也就是说只有在延迟期到时才能够从队列中取元素。
DelayQueue主要用于两个方面：
– 缓存：清掉缓存中超时的缓存数据
– 任务超时处理
DelayQueue
DelayQueue实现的关键主要有如下几个：
- 可重入锁ReentrantLock
- 用于阻塞和通知的Condition对象
- 根据Delay时间排序的优先级队列：PriorityQueue
- 用于优化阻塞通知的线程元素leader
ReentrantLock、Condition这两个对象就不需要阐述了，他是实现整个BlockingQueue的核心。PriorityQueue是一个支持优先级线程排序的队列（参考【死磕Java并发】—–J.U.C之阻塞队列：PriorityBlockingQueue），leader后面阐述。这里我们先来了解Delay，他是实现延时操作的关键。
Delayed
Delayed接口是用来标记那些应该在给定延迟时间之后执行的对象，它定义了一个long getDelay(TimeUnit unit)方法，该方法返回与此对象相关的的剩余时间。同时实现该接口的对象必须定义一个compareTo 方法，该方法提供与此接口的 getDelay 方法一致的排序。
public interface Delayed extends Comparable<Delayed> {
long getDelay(TimeUnit unit);
}
如何使用该接口呢？上面说的非常清楚了，实现该接口的getDelay()方法，同时定义compareTo()方法即可。
内部结构
先看DelayQueue的定义：
public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
implements BlockingQueue<E> {
/** 可重入锁 */
private final transient ReentrantLock lock = new ReentrantLock();
/** 支持优先级的BlockingQueue */
private final PriorityQueue<E> q = new PriorityQueue<E>();
/** 用于优化阻塞 */
private Thread leader = null;
/** Condition */
private final Condition available = lock.newCondition(); /**
* 省略很多代码
*/
}
看了DelayQueue的内部结构就对上面几个关键点一目了然了，但是这里有一点需要注意，DelayQueue的元素都必须继承Delayed接口。同时也可以从这里初步理清楚DelayQueue内部实现的机制了：以支持优先级无界队列的PriorityQueue作为一个容器，容器里面的元素都应该实现Delayed接口，在每次往优先级队列中添加元素时以元素的过期时间作为排序条件，最先过期的元素放在优先级最高。
offer()
public boolean offer(E e) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
// 向 PriorityQueue中插入元素
q.offer(e);
// 如果当前元素的对首元素（优先级最高），leader设置为空，唤醒所有等待线程
if (q.peek() == e) {
leader = null;
available.signal();
}
// 无界队列，永远返回true
return true;
} finally {
lock.unlock();
}
}
offer(E e)就是往PriorityQueue中添加元素，具体可以参考（【死磕Java并发】—–J.U.C之阻塞队列：PriorityBlockingQueue）。整个过程还是比较简单，但是在判断当前元素是否为对首元素，如果是的话则设置leader=null，这是非常关键的一个步骤，后面阐述。
take()
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
for (;;) {
// 对首元素
E first = q.peek();
// 对首为空，阻塞，等待off()操作唤醒
if (first == null)
available.await();
else {
// 获取对首元素的超时时间
long delay = first.getDelay(NANOSECONDS);
// <=0 表示已过期，出对，return
if (delay <= 0)
return q.poll();
first = null; // don't retain ref while waiting
// leader != null 证明有其他线程在操作，阻塞
if (leader != null)
available.await();
else {
// 否则将leader 设置为当前线程，独占
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
// 超时阻塞
available.awaitNanos(delay);
} finally {
// 释放leader
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
// 唤醒阻塞线程
if (leader == null && q.peek() != null)
available.signal();
lock.unlock();
}
}
首先是获取对首元素，如果对首元素的延时时间 delay <= 0 ，则可以出对了，直接return即可。否则设置first = null，这里设置为null的主要目的是为了避免内存泄漏。如果 leader != null 则表示当前有线程占用，则阻塞，否则设置leader为当前线程，然后调用awaitNanos()方法超时等待。
first = null
这里为什么如果不设置first = null，则会引起内存泄漏呢？线程A到达，列首元素没有到期，设置leader = 线程A，这是线程B来了因为leader != null，则会阻塞，线程C一样。假如线程阻塞完毕了，获取列首元素成功，出列。这个时候列首元素应该会被回收掉，但是问题是它还被线程B、线程C持有着，所以不会回收，这里只有两个线程，如果有线程D、线程E…呢？这样会无限期的不能回收，就会造成内存泄漏。
这个入队、出对过程和其他的阻塞队列没有很大区别，无非是在出对的时候增加了一个到期时间的判断。同时通过leader来减少不必要阻塞。
ConcurrentLinkedQueue
原文出处http://cmsblogs.com/ 『chenssy』
要实现一个线程安全的队列有两种方式：阻塞和非阻塞。阻塞队列无非就是锁的应用，而非阻塞则是CAS算法的应用。下面我们就开始一个非阻塞算法的研究：CoucurrentLinkedQueue。
ConcurrentLinkedQueue是一个基于链接节点的无边界的线程安全队列，它采用FIFO原则对元素进行排序。采用“wait-free”算法（即CAS算法）来实现的。
CoucurrentLinkedQueue规定了如下几个不变性：
- 在入队的最后一个元素的next为null
- 队列中所有未删除的节点的item都不能为null且都能从head节点遍历到
- 对于要删除的节点，不是直接将其设置为null，而是先将其item域设置为null（迭代器会跳过item为null的节点）
- 允许head和tail更新滞后。这是什么意思呢？意思就说是head、tail不总是指向第一个元素和最后一个元素（后面阐述）。
head的不变性和可变性：
- 不变性
- 所有未删除的节点都可以通过head节点遍历到
- head不能为null
- head节点的next不能指向自身
- 可变性
- head的item可能为null，也可能不为null
2.允许tail滞后head，也就是说调用succc()方法，从head不可达tail
- head的item可能为null，也可能不为null
tail的不变性和可变性
- 不变性
- tail不能为null
- 可变性
- tail的item可能为null，也可能不为null
- tail节点的next域可以指向自身
3.允许tail滞后head，也就是说调用succc()方法，从head不可达tail
这些特性是否已经晕了？没关系，我们看下面的源码分析就可以理解这些特性了。
ConcurrentLinkedQueue源码分析
CoucurrentLinkedQueue的结构由head节点和tail节点组成，每个节点由节点元素item和指向下一个节点的next引用组成，而节点与节点之间的关系就是通过该next关联起来的，从而组成一张链表的队列。节点Node为ConcurrentLinkedQueue的内部类，定义如下：
private static class Node<E> {private static class Node<E> {
/** 节点元素域 */
volatile E item;
volatile Node<E> next; //初始化,获得item 和 next 的偏移量,为后期的CAS做准备 Node(E item) {
UNSAFE.putObject(this, itemOffset, item);
} boolean casItem(E cmp, E val) {
return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
} void lazySetNext(Node<E> val) {
UNSAFE.putOrderedObject(this, nextOffset, val);
} boolean casNext(Node<E> cmp, Node<E> val) {
return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
} // Unsafe mechanics private static final sun.misc.Unsafe UNSAFE;
/** 偏移量 */
private static final long itemOffset;
/** 下一个元素的偏移量 */ private static final long nextOffset; static {
try {
UNSAFE = sun.misc.Unsafe.getUnsafe();
Class<?> k = Node.class;
itemOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("item"));
nextOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("next"));
} catch (Exception e) {
throw new Error(e);
}
}
}
/** 节点元素域 */
volatile E item;
volatile Node<E> next; //初始化,获得item 和 next 的偏移量,为后期的CAS做准备 Node(E item) {
UNSAFE.putObject(this, itemOffset, item);
} boolean casItem(E cmp, E val) {
return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
} void lazySetNext(Node<E> val) {
UNSAFE.putOrderedObject(this, nextOffset, val);
} boolean casNext(Node<E> cmp, Node<E> val) {
return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
} // Unsafe mechanics private static final sun.misc.Unsafe UNSAFE;
/** 偏移量 */
private static final long itemOffset;
/** 下一个元素的偏移量 */ private static final long nextOffset; static {
try {
UNSAFE = sun.misc.Unsafe.getUnsafe();
Class<?> k = Node.class;
itemOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("item"));
nextOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("next"));
} catch (Exception e) {
throw new Error(e);
}
}
}
入列
入列，我们认为是一个非常简单的过程：tail节点的next执行新节点，然后更新tail为新节点即可。从单线程角度我们这么理解应该是没有问题的，但是多线程呢？如果一个线程正在进行插入动作，那么它必须先获取尾节点，然后设置尾节点的下一个节点为当前节点，但是如果已经有一个线程刚刚好完成了插入，那么尾节点是不是发生了变化？对于这种情况ConcurrentLinkedQueue怎么处理呢？我们先看源码：
offer(E e)：将指定元素插入都队列尾部：
public boolean offer(E e) {boolean offer(E e) {
//检查节点是否为null
checkNotNull(e);
// 创建新节点
final Node<E> newNode = new Node<E>(e); //死循环 直到成功为止
for (Node<E> t = tail, p = t;;) {
Node<E> q = p.next;
// q == null 表示 p已经是最后一个节点了，尝试加入到队列尾
// 如果插入失败，则表示其他线程已经修改了p的指向
if (q == null) { // --- 1
// casNext：t节点的next指向当前节点
// casTail：设置tail 尾节点
if (p.casNext(null, newNode)) { // --- 2
// node 加入节点后会导致tail距离最后一个节点相差大于一个，需要更新tail
if (p != t) // --- 3
casTail(t, newNode); // --- 4
return true;
}
}
// p == q 等于自身
else if (p == q) // --- 5
// p == q 代表着该节点已经被删除了
// 由于多线程的原因，我们offer()的时候也会poll()，如果offer()的时候正好该节点已经poll()了
// 那么在poll()方法中的updateHead()方法会将head指向当前的q，而把p.next指向自己，即：p.next == p
// 这样就会导致tail节点滞后head（tail位于head的前面），则需要重新设置p
p = (t != (t = tail)) ? t : head; // --- 6
// tail并没有指向尾节点
else
// tail已经不是最后一个节点，将p指向最后一个节点
p = (p != t && t != (t = tail)) ? t : q; // --- 7
}
}
//检查节点是否为null
checkNotNull(e);
// 创建新节点
final Node<E> newNode = new Node<E>(e); //死循环 直到成功为止
for (Node<E> t = tail, p = t;;) {
Node<E> q = p.next;
// q == null 表示 p已经是最后一个节点了，尝试加入到队列尾
// 如果插入失败，则表示其他线程已经修改了p的指向
if (q == null) { // --- 1
// casNext：t节点的next指向当前节点
// casTail：设置tail 尾节点
if (p.casNext(null, newNode)) { // --- 2
// node 加入节点后会导致tail距离最后一个节点相差大于一个，需要更新tail
if (p != t) // --- 3
casTail(t, newNode); // --- 4
return true;
}
}
// p == q 等于自身
else if (p == q) // --- 5
// p == q 代表着该节点已经被删除了
// 由于多线程的原因，我们offer()的时候也会poll()，如果offer()的时候正好该节点已经poll()了
// 那么在poll()方法中的updateHead()方法会将head指向当前的q，而把p.next指向自己，即：p.next == p
// 这样就会导致tail节点滞后head（tail位于head的前面），则需要重新设置p
p = (t != (t = tail)) ? t : head; // --- 6
// tail并没有指向尾节点
else
// tail已经不是最后一个节点，将p指向最后一个节点
p = (p != t && t != (t = tail)) ? t : q; // --- 7
}
}
光看源码还是有点儿迷糊的，插入节点一次分析就会明朗很多。
初始化
ConcurrentLinkedQueue初始化时head、tail存储的元素都为null，且head等于tail：
转存失败http://cmsblogs.com/ 『chenssy』
前面提到的各种BlockingQueue对读或者写都是锁上整个队列，在并发量大的时候，各种锁是比较耗资源和耗时间的，而前面的SynchronousQueue虽然不会锁住整个队列，但它是一个没有容量的“队列”，那么有没有这样一种队列，它即可以像其他的BlockingQueue一样有容量又可以像SynchronousQueue一样不会锁住整个队列呢？有！答案就是LinkedTransferQueue。
LinkedTransferQueue是基于链表的FIFO无界阻塞队列，它出现在JDK7中。Doug Lea 大神说LinkedTransferQueue是一个聪明的队列。它是ConcurrentLinkedQueue、SynchronousQueue (公平模式下)、无界的LinkedBlockingQueues等的超集。既然这么牛逼，那势必要弄清楚其中的原理了。
LinkedTransferQueue
看源码之前我们先稍微了解下它的原理，这样看源码就会有迹可循了。
LinkedTransferQueue采用一种预占模式。什么意思呢？有就直接拿走，没有就占着这个位置直到拿到或者超时或者中断。即消费者线程到队列中取元素时，如果发现队列为空，则会生成一个null节点，然后park住等待生产者。后面如果生产者线程入队时发现有一个null元素节点，这时生产者就不会入列了，直接将元素填充到该节点上，唤醒该节点的线程，被唤醒的消费者线程拿东西走人。是不是有点儿SynchronousQueue的味道？
结构
LinkedTransferQueue与其他的BlockingQueue一样，同样继承AbstractQueue类，但是它实现了TransferQueue，TransferQueue接口继承BlockingQueue，所以TransferQueue算是对BlockingQueue一种扩充，该接口提供了一整套的transfer接口：
public interface TransferQueue<E> extends BlockingQueue<E> { /**
* 若当前存在一个正在等待获取的消费者线程（使用take()或者poll()函数），使用该方法会即刻转移/传输对象元素e；
* 若不存在，则返回false，并且不进入队列。这是一个不阻塞的操作
*/
boolean tryTransfer(E e); /**
* 若当前存在一个正在等待获取的消费者线程，即立刻移交之；
* 否则，会插入当前元素e到队列尾部，并且等待进入阻塞状态，到有消费者线程取走该元素
*/
void transfer(E e) throws InterruptedException; /**
* 若当前存在一个正在等待获取的消费者线程，会立即传输给它;否则将插入元素e到队列尾部，并且等待被消费者线程获取消费掉；
* 若在指定的时间内元素e无法被消费者线程获取，则返回false，同时该元素被移除。
*/
boolean tryTransfer(E e, long timeout, TimeUnit unit)
throws InterruptedException; /**
* 判断是否存在消费者线程
*/
boolean hasWaitingConsumer(); /**
* 获取所有等待获取元素的消费线程数量
*/
int getWaitingConsumerCount();
}
相对于其他的BlockingQueue，LinkedTransferQueue就多了上面几个方法。这几个方法在LinkedTransferQueue中起到了核心作用。
LinkedTransferQueue定义的变量如下：
// 判断是否为多核
private static final boolean MP =
Runtime.getRuntime().availableProcessors() > 1; // 自旋次数
private static final int FRONT_SPINS = 1 << 7; // 前驱节点正在处理，当前节点需要自旋的次数
private static final int CHAINED_SPINS = FRONT_SPINS >>> 1; static final int SWEEP_THRESHOLD = 32; // 头节点
transient volatile Node head; // 尾节点
private transient volatile Node tail; // 删除节点失败的次数
private transient volatile int sweepVotes; /*
* 调用xfer()方法时需要传入,区分不同处理
* xfer()方法是LinkedTransferQueue的最核心的方法
*/
private static final int NOW = 0; // for untimed poll, tryTransfer
private static final int ASYNC = 1; // for offer, put, add
private static final int SYNC = 2; // for transfer, take
private static final int TIMED = 3; // for timed poll, tryTransfer
Node节点
Node节点由四个部分构成：
- isData：表示该节点是存放数据还是获取数据
- item：存放数据，isData为false时，该节点为null，为true时，匹配后，该节点会置为null
- next：指向下一个节点
- waiter：park住消费者线程，线程就放在这里
结构如下：
源码如下：
static final class Node {
// 表示该节点是存放数据还是获取数据
final boolean isData;
// 存放数据，isData为false时，该节点为null，为true时，匹配后，该节点会置为null
volatile Object item;
//指向下一个节点
volatile Node next; // park住消费者线程，线程就放在这里
volatile Thread waiter; // null until waiting /**
* CAS Next域
*/
final boolean casNext(Node cmp, Node val) {
return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
} /**
* CAS itme域
*/
final boolean casItem(Object cmp, Object val) {
return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
} /**
* 构造函数
*/
Node(Object item, boolean isData) {
UNSAFE.putObject(this, itemOffset, item); // relaxed write
this.isData = isData;
} /**
* 将next域指向自身，其实就是剔除节点
*/
final void forgetNext() {
UNSAFE.putObject(this, nextOffset, this);
} /**
* 匹配过或节点被取消的时候会调用
*/
final void forgetContents() {
UNSAFE.putObject(this, itemOffset, this);
UNSAFE.putObject(this, waiterOffset, null);
} /**
* 校验节点是否匹配过，如果匹配做取消了，item则会发生变化
*/
final boolean isMatched() {
Object x = item;
return (x == this) || ((x == null) == isData);
} /**
* 是否是一个未匹配的请求节点
* 如果是的话isData应为false，item == null，因位如果匹配了，item则会有值
*/
final boolean isUnmatchedRequest() {
return !isData && item == null;
} /**
* 如给定节点类型不能挂在当前节点后返回true
*/
final boolean cannotPrecede(boolean haveData) {
boolean d = isData;
Object x;
return d != haveData && (x = item) != this && (x != null) == d;
} /**
* 匹配一个数据节点
*/
final boolean tryMatchData() {
// assert isData;
Object x = item;
if (x != null && x != this && casItem(x, null)) {
LockSupport.unpark(waiter);
return true;
}
return false;
} private static final long serialVersionUID = -3375979862319811754L; // Unsafe mechanics
private static final sun.misc.Unsafe UNSAFE;
private static final long itemOffset;
private static final long nextOffset;
private static final long waiterOffset;
static {
try {
UNSAFE = sun.misc.Unsafe.getUnsafe();
Class<?> k = Node.class;
itemOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("item"));
nextOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("next"));
waiterOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("waiter"));
} catch (Exception e) {
throw new Error(e);
}
}
}
节点Node为LinkedTransferQueue的内部类，其内部结构和公平方式的SynchronousQueue差不多，里面也同样提供了一些很重要的方法。
put操作
LinkedTransferQueue提供了add、put、offer三类方法，用于将元素插入队列中，如下：
public void put(E e) {
xfer(e, true, ASYNC, 0);
} public boolean offer(E e, long timeout, TimeUnit unit) {
xfer(e, true, ASYNC, 0);
return true;
} public boolean offer(E e) {
xfer(e, true, ASYNC, 0);
return true;
} public boolean add(E e) {
xfer(e, true, ASYNC, 0);
return true;
}
由于LinkedTransferQueue是无界的，不会阻塞，所以在调用xfer方法是传入的是ASYNC，同时直接返回true.
take操作
LinkedTransferQueue提供了poll、take方法用于出列元素：
public E take() throws InterruptedException {
E e = xfer(null, false, SYNC, 0);
if (e != null)
return e;
Thread.interrupted();
throw new InterruptedException();
} public E poll() {
return xfer(null, false, NOW, 0);
} public E poll(long timeout, TimeUnit unit) throws InterruptedException {
E e = xfer(null, false, TIMED, unit.toNanos(timeout));
if (e != null || !Thread.interrupted())
return e;
throw new InterruptedException();
}
这里和put操作有点不一样，take()方法传入的是SYNC，阻塞。poll()传入的是NOW，poll(long timeout, TimeUnit unit)则是传入TIMED。
tranfer操作
实现TransferQueue接口，就要实现它的方法：
public boolean tryTransfer(E e, long timeout, TimeUnit unit)
throws InterruptedException {
if (xfer(e, true, TIMED, unit.toNanos(timeout)) == null)
return true;
if (!Thread.interrupted())
return false;
throw new InterruptedException();
}public void transfer(E e) throws InterruptedException {
if (xfer(e, true, SYNC, 0) != null) {
Thread.interrupted(); // failure possible only due to interrupt
throw new InterruptedException();
}
}public boolean tryTransfer(E e) {
return xfer(e, true, NOW, 0) == null;
}
xfer()
通过上面几个核心方法的源码我们清楚可以看到，最终都是调用xfer()方法，该方法接受四个参数，item或者null的E，put操作为true、take操作为false的havaData，how（有四个值NOW, ASYNC, SYNC, or TIMED，分别表示不同的操作），超时nanos。
private E xfer(E e, boolean haveData, int how, long nanos) { // havaData为true，但是e == null 抛出空指针
if (haveData && (e == null))
throw new NullPointerException();
Node s = null; // the node to append, if needed retry:
for (;;) { // 从首节点开始匹配
// p == null 队列为空
for (Node h = head, p = h; p != null;) { // 模型，request or data
boolean isData = p.isData;
// item域
Object item = p.item; // 找到一个没有匹配的节点
// item != p 也就是自身，则表示没有匹配过
// (item != null) == isData，表示模型符合
if (item != p && (item != null) == isData) { // 节点类型和待处理类型一致，这样肯定是不能匹配的
if (isData == haveData) // can't match
break;
// 匹配，将E加入到item域中
// 如果p 的item为data，那么e为null,如果p的item为null，那么e为data
if (p.casItem(item, e)) { // match
//
for (Node q = p; q != h;) {
Node n = q.next; // update by 2 unless singleton
if (head == h && casHead(h, n == null ? q : n)) {
h.forgetNext();
break;
} // advance and retry
if ((h = head) == null ||
(q = h.next) == null || !q.isMatched())
break; // unless slack < 2
} // 匹配后唤醒p的waiter线程;reservation则叫人收货，data则叫null收货
LockSupport.unpark(p.waiter);
return LinkedTransferQueue.<E>cast(item);
}
}
// 如果已经匹配了则向前推进
Node n = p.next;
// 如果p的next指向p本身，说明p节点已经有其他线程处理过了，只能从head重新开始
p = (p != n) ? n : (h = head); // Use head if p offlist
} // 如果没有找到匹配的节点，则进行处理
// NOW为untimed poll, tryTransfer，不需要入队
if (how != NOW) { // No matches available
// s == null，新建一个节点
if (s == null)
s = new Node(e, haveData);
// 入队，返回前驱节点
Node pred = tryAppend(s, haveData);
// 返回的前驱节点为null，那就是有race，被其他的抢了，那就continue 整个for
if (pred == null)
continue retry; // ASYNC不需要阻塞等待
if (how != ASYNC)
return awaitMatch(s, pred, e, (how == TIMED), nanos);
}
return e;
}
}
整个算法的核心就是寻找匹配节点找到了就返回，否则就入队（NOW直接返回）：
- matched。判断匹配条件（isData不一样，本身没有匹配），匹配后就casItem，然后unpark匹配节点的waiter线程，如果是reservation则叫人收货，data则叫null收货。
- unmatched。如果没有找到匹配节点，则根据传入的how来处理，NOW直接返回，其余三种先入对，入队后如果是ASYNC则返回，SYNC和TIMED则会阻塞等待匹配。
其实相当于SynchronousQueue来说，这个处理逻辑还是比较简单的。
如果没有找到匹配节点，且how != NOW会入队，入队则是调用tryAppend方法：
private Node tryAppend(Node s, boolean haveData) {
// 从尾节点tail开始
for (Node t = tail, p = t;;) {
Node n, u; // 队列为空则将节点S设置为head
if (p == null && (p = head) == null) {
if (casHead(null, s))
return s;
} // 如果为data
else if (p.cannotPrecede(haveData))
return null; // 不是最后一个节点
else if ((n = p.next) != null)
p = p != t && t != (u = tail) ? (t = u) : (p != n) ? n : null;
// CAS失败，一般来说失败的原因在于p.next != null，可能有其他增加了tail，向前推荐
else if (!p.casNext(null, s))
p = p.next; // re-read on CAS failure
else {
if (p != t) { // update if slack now >= 2
while ((tail != t || !casTail(t, s)) &&
(t = tail) != null &&
(s = t.next) != null && // advance and retry
(s = s.next) != null && s != t);
}
return p;
}
}
}
tryAppend方法是将S节点添加到tail上，然后返回其前驱节点。好吧，我承认这段代码我看的有点儿晕！！！
加入队列后，如果how还不是ASYNC则调用awaitMatch()方法阻塞等待：
private E awaitMatch(Node s, Node pred, E e, boolean timed, long nanos) {
// 超时控制
final long deadline = timed ? System.nanoTime() + nanos : 0L; // 当前线程
Thread w = Thread.currentThread(); // 自旋次数
int spins = -1; // initialized after first item and cancel checks // 随机数
ThreadLocalRandom randomYields = null; // bound if needed for (;;) {
Object item = s.item;
//匹配了，可能有其他线程匹配了线程
if (item != e) {
// 撤销该节点
s.forgetContents();
return LinkedTransferQueue.<E>cast(item);
} // 线程中断或者超时了。则调用将s节点item设置为e，等待取消
if ((w.isInterrupted() || (timed && nanos <= 0)) && s.casItem(e, s)) { // cancel
// 断开节点
unsplice(pred, s);
return e;
} // 自旋
if (spins < 0) {
// 计算自旋次数
if ((spins = spinsFor(pred, s.isData)) > 0)
randomYields = ThreadLocalRandom.current();
} // 自旋
else if (spins > 0) {
--spins;
// 生成的随机数 == 0 ，停止线程？不是很明白....
if (randomYields.nextInt(CHAINED_SPINS) == 0)
Thread.yield();
} // 将当前线程设置到节点的waiter域
// 一开始s.waiter == null 肯定是会成立的，
else if (s.waiter == null) {
s.waiter = w; // request unpark then recheck
} // 超时阻塞
else if (timed) {
nanos = deadline - System.nanoTime();
if (nanos > 0L)
LockSupport.parkNanos(this, nanos);
}
else {
// 不是超时阻塞
LockSupport.park(this);
}
}
}
整个awaitMatch过程和SynchronousQueue的awaitFulfill没有很大区别，不过在自旋过程会调用Thread.yield();这是干嘛？
在awaitMatch过程中，如果线程中断了，或者超时了则会调用unsplice()方法去除该节点：
final void unsplice(Node pred, Node s) {
s.forgetContents(); // forget unneeded fields if (pred != null && pred != s && pred.next == s) {
Node n = s.next;
if (n == null ||
(n != s && pred.casNext(s, n) && pred.isMatched())) { for (;;) { // check if at, or could be, head
Node h = head;
if (h == pred || h == s || h == null)
return; // at head or list empty
if (!h.isMatched())
break;
Node hn = h.next;
if (hn == null)
return; // now empty
if (hn != h && casHead(h, hn))
h.forgetNext(); // advance head
}
if (pred.next != pred && s.next != s) { // recheck if offlist
for (;;) { // sweep now if enough votes
int v = sweepVotes;
if (v < SWEEP_THRESHOLD) {
if (casSweepVotes(v, v + 1))
break;
}
else if (casSweepVotes(v, 0)) {
sweep();
break;
}
}
}
}
}
}
主体流程已经完成，这里总结下：
- 无论是入对、出对，还是交换，最终都会跑到xfer(E e, boolean haveData, int how, long nanos)方法中，只不过传入的how不同而已
- 如果队列不为空，则尝试在队列中寻找是否存在与该节点相匹配的节点，如果找到则将匹配节点的item设置e，然后唤醒匹配节点的waiter线程。如果是reservation则叫人收货，data则叫null收货
- 如果队列为空，或者没有找到匹配的节点且how ！= NOW，则调用tryAppend()方法将节点添加到队列的tail，然后返回其前驱节点
- 如果节点的how != NOW && how != ASYNC，则调用awaitMatch()方法阻塞等待，在阻塞等待过程中和SynchronousQuque的awaitFulfill()逻辑差不多，都是先自旋，然后判断是否需要自旋，如果中断或者超时了则将该节点从队列中移出
实例
这段摘自JAVA 1.7并发之LinkedTransferQueue原理理解。感觉看完上面的源码后，在结合这个例子会有更好的了解，掌握。
1：Head->Data Input->Data
Match: 根据他们的属性 发现 cannot match ，因为是同类的
处理节点: 所以把新的data放在原来的data后面，然后head往后移一位，Reservation同理
HEAD=DATA->DATA
2：Head->Data Input->Reservation （取数据）
Match: 成功match，就把Data的item变为reservation的值（null,有主了），并且返回数据。
处理节点： 没动，head还在原地
HEAD=DATA（用过）
3：Head->Reservation Input->Data（放数据）
Match: 成功match，就把Reservation的item变为Data的值（有主了），并且叫waiter来取
处理节点： 没动
HEAD=RESERVATION(用过)
总结
BlockingQueue
BlockingQueue接口实现Queue接口，它支持两个附加操作：获取元素时等待队列变为非空，以及存储元素时等待空间变得可用。相对于同一操作他提供了四种机制：抛出异常、返回特殊值、阻塞等待、超时：
BlockingQueue常用于生产者和消费者场景。
JDK 8 中提供了七个阻塞队列可供使用（上图的DelayedWorkQueue是ScheduledThreadPoolExecutor的内部类）：
- ArrayBlockingQueue ：一个由数组结构组成的有界阻塞队列。
- LinkedBlockingQueue ：一个由链表结构组成的无界阻塞队列。
- PriorityBlockingQueue ：一个支持优先级排序的无界阻塞队列。
- DelayQueue：一个使用优先级队列实现的无界阻塞队列。
- SynchronousQueue：一个不存储元素的阻塞队列。
- LinkedTransferQueue：一个由链表结构组成的无界阻塞队列。
- LinkedBlockingDeque：一个由链表结构组成的双向阻塞队列。
ArrayBlockingQueue
基于数组的阻塞队列，ArrayBlockingQueue内部维护这一个定长数组，阻塞队列的大小在初始化时就已经确定了，其后无法更改。
采用可重入锁ReentrantLock来保证线程安全性，但是生产者和消费者是共用同一个锁对象，这样势必会导致降低一定的吞吐量。当然ArrayBlockingQueue完全可以采用分离锁来实现生产者和消费者的并行操作，但是我认为这样做只会给代码带来额外的复杂性，对于性能而言应该不会有太大的提升，因为基于数组的ArrayBlockingQueue在数据的写入和读取操作已经非常轻巧了。
ArrayBlockingQueue支持公平性和非公平性，默认采用非公平模式，可以通过构造函数设置为公平访问策略（true）。
PriorityBlockingQueue
PriorityBlockingQueue是支持优先级的无界队列。默认情况下采用自然顺序排序，当然也可以通过自定义Comparator来指定元素的排序顺序。
PriorityBlockingQueue内部采用二叉堆的实现方式，整个处理过程并不是特别复杂。添加操作则是不断“上冒”，而删除操作则是不断“下掉”。
DelayQueue
DelayQueue是一个支持延时操作的无界阻塞队列。列头的元素是最先“到期”的元素，如果队列里面没有元素到期，是不能从列头获取元素的，哪怕有元素也不行。也就是说只有在延迟期满时才能够从队列中去元素。
它主要运用于如下场景：
- 缓存系统的设计：缓存是有一定的时效性的，可以用DelayQueue保存缓存的有效期，然后利用一个线程查询DelayQueue，如果取到元素就证明该缓存已经失效了。
- 定时任务的调度：DelayQueue保存当天将要执行的任务和执行时间，一旦取到元素（任务），就执行该任务。
DelayQueue采用支持优先级的PriorityQueue来实现，但是队列中的元素必须要实现Delayed接口，Delayed接口用来标记那些应该在给定延迟时间之后执行的对象，该接口提供了getDelay()方法返回元素节点的剩余时间。同时，元素也必须要实现compareTo()方法，compareTo()方法需要提供与getDelay()方法一致的排序。
SynchronousQueue
SynchronousQueue是一个神奇的队列，他是一个不存储元素的阻塞队列，也就是说他的每一个put操作都需要等待一个take操作，否则就不能继续添加元素了，有点儿像Exchanger，类似于生产者和消费者进行交换。
队列本身不存储任何元素，所以非常适用于传递性场景，两者直接进行对接。其吞吐量会高于ArrayBlockingQueue和LinkedBlockingQueue。
SynchronousQueue支持公平和非公平的访问策略，在默认情况下采用非公平性，也可以通过构造函数来设置为公平性。
SynchronousQueue的实现核心为Transferer接口，该接口有TransferQueue和TransferStack两个实现类，分别对应着公平策略和非公平策略。接口Transferer有一个tranfer()方法，该方法定义了转移数据，如果e != null，相当于将一个数据交给消费者，如果e == null，则相当于从一个生产者接收一个消费者交出的数据。
LinkedTransferQueue
LinkedTransferQueue是一个由链表组成的的无界阻塞队列，该队列是一个相当牛逼的队列：它是ConcurrentLinkedQueue、SynchronousQueue (公平模式下)、无界的LinkedBlockingQueues等的超集。
与其他BlockingQueue相比，他多实现了一个接口TransferQueue，该接口是对BlockingQueue的一种补充，多了tryTranfer()和transfer()两类方法：
- tranfer()：若当前存在一个正在等待获取的消费者线程，即立刻移交之。 否则，会插入当前元素e到队列尾部，并且等待进入阻塞状态，到有消费者线程取走该元素
-
tryTranfer()： 若当前存在一个正在等待获取的消费者线程（使用take()或者poll()函数），使用该方法会即刻转移/传输对象元素e；若不存在，则返回false，并且不进入队列。这是一个不阻塞的操作
LinkedBlockingDeque
LinkedBlockingDeque是一个有链表组成的双向阻塞队列，与前面的阻塞队列相比它支持从两端插入和移出元素。以first结尾的表示从对头操作，以last结尾的表示从对尾操作。
在初始化LinkedBlockingDeque时可以初始化队列的容量，用来防止其再扩容时过渡膨胀。另外双向阻塞队列可以运用在“工作窃取”模式中。
更多内容请关注微信公众号【Java技术江湖】
一位阿里 Java 工程师的技术小站。作者黄小斜，专注 Java 相关技术：SSM、SpringBoot、MySQL、分布式、中间件、集群、Linux、网络、多线程，偶尔讲点Docker、ELK，同时也分享技术干货和学习经验，致力于Java全栈开发！（关注公众号后回复”Java“即可领取 Java基础、进阶、项目和架构师等免费学习资料，更有数据库、分布式、微服务等热门技术学习视频，内容丰富，兼顾原理和实践，另外也将赠送作者原创的Java学习指南、Java程序员面试指南等干货资源）
