Java并发编程实战（进阶篇

通过上一篇的学习我们就可以运用这些方法来构建线程安全的类，那么本篇我们会着重学习Java中的同步容器类、并发容器和框架、并发工具类、Executor 框架，看看这些并发类库或者框架是怎样实现的。

一、同步容器类

Java中同步容器类有2类：

Vector、Stack、Hashtable
Collections 类中提供的静态工厂方法创建的类（由 Collections.synchronizedXxx 创建）

这些类实现线程安全的方式是：把他们的状态封装起来，并对每个公共方法都进行同步，使得每次只有一个线程能访问容器的状态（这也导致了性能低的原因）。

1.同步容器类的问题

同步容器类都是线程安全的，但在某些情况下可能需要额外的客户端加锁来保护复合操作。

容器上常见的复合操作包括：

迭代：反复访问元素，直到遍历完全部元素；
跳转：根据指定顺序寻找当前元素的下一个（下 n 个）元素；
条件运算：例如若没有则添加等（例如 “若没有则添加”）；

下面代码在 Vector 中定义两个方法：getLast 和 delectLast，在 Vector 底层都会执行 “先检查后执行” 的操作：

public static Object getLast(Vector list) {int lastIndex = list.size() - 1;return list.get(lastIndex);}public static void delectLast(Vector list) {int lastIndex = list.size() - 1;list.remove(lastIndex);}

可以查看 Vector 的 get 和 remove 方法源码，其中会进去判断元素大小与坐标的关系，如果存在 index >= elementCount （当前下标大于等于容器元素个数）就会抛出 ArrayIndexOutOfBoundsException 异常：

public synchronized E get(int index) {if (index >= elementCount)throw new ArrayIndexOutOfBoundsException(index);return elementData(index);}public synchronized E remove(int index) {modCount++;if (index >= elementCount)throw new ArrayIndexOutOfBoundsException(index);E oldValue = elementData(index);int numMoved = elementCount - index - 1;if (numMoved > 0)System.arraycopy(elementData, index+1, elementData, index,numMoved);elementData[--elementCount] = null; // Let gc do its workreturn oldValue;}

这些方法看似没有问题，但如果多个线程调用就会出现问题。

交替调用 getLast 和 delectLast 时将抛出 ArrayIndexOutOfBoundsException：

虽然在线程A抛出了异常，但这不是 getLast 调用者所希望看到的结果（应该在 9 移除之后，get 到 8 下标的元素）。

由于同步容器类要遵守同步策略，即支持客户端加锁，因此可能会创建一些新的操作，只要我们知道应该使用哪一个锁，那么这些新操作就与容器的其他操作一样都是原子操作。同步容器类通过其自身的内置锁来保护它的每个方法，所以我们可以在客户端加锁时，获取同步容器类的锁。

在使用客户端加锁的 Vector 上的复合操作：

public static Object getLast(Vector list) {synchronized (list) {int lastIndex = list.size() - 1;return list.get(lastIndex);}}public static void delectLast(Vector list) {synchronized (list) {int lastIndex = list.size() - 1;list.remove(lastIndex);}}

通过获取容器类的锁，我们可以使 getLast 和 delectLast 成为原子操作，并确保 Vector 的大小在调用 size 和 get 之间不会发生变化。

在调用 size 和相应的 get 之间，Vector 的长度可能会发生变化，这种风险在对 Vector 元素进行迭代时仍然会出现：

for (int i = 0; i < vector.size(); i++) {dosomething(vector.get(i));}

如果对 Vector 进行迭代时，另一个线程删除了一个元素，并且这两个操作交替执行，那么这种迭代方法将会抛出 ArrayIndexOutOfBoundsException 异常。

使用带有客户端加锁的迭代来解决：

synchronized (vector) {for (int i = 0; i < vector.size(); i++) {dosomething(vector.get(i));}}

这样会导致其他线程在迭代期间无法访问 Vector，因此降低了并发性。

2.迭代器与 ConcurrentModificationException

为了将问题阐述清楚，我们使用了 Vector，虽然 Vector 比较早，但现在出现的许多容器也没有消除复合操作的问题。

如果有其他线程并发地修改容器，那么即使是使用迭代器也无法避免在迭代期间对容器进行加锁。在设计同步容器类的迭代器时，并没有考虑到并发修改的问题，并且表现出 “即使失败” 的行为（当容器在迭代时发现被修改时，就会抛出 ConcurrentModificationException 异常）。

下面我们分析一下并发修改容器会怎样抛出 ConcurrentModificationException 异常（这里的修改仅仅包含向容器添加元素或者删除容器元素）

首先我们先使用迭代器遍历 Vector （无论在直接迭代还是 for-each 循环语句中，对容器类进行迭代的标准方式都是使用 Iterator）：

public static void visit(Vector<Integer> list) {Iterator iterator = list.iterator();while(iterator.hasNext()) {Integer next = (Integer) iterator.next();}}

另外一个线程调用删除容器元素：

public static void modif(Vector<Integer> list) {// 或者 list.add(0) 都会抛出异常list.remove(0);}

那么就可能会抛出 ConcurrentModificationException 异常，那么我们深入 iterator 源码看一看

首先它会返回一个 new Itr() （一个私有内部类实现了 Iterator 接口）：

public synchronized Iterator<E> iterator() {return new Itr();}

那么我们看一下这个私有内部类怎样实现的 Iterator 接口：

private class Itr implements Iterator<E> {int cursor;       int lastRet = -1; int expectedModCount = modCount;public boolean hasNext() {return cursor != elementCount;}public E next() {synchronized (Vector.this) {checkForComodification();int i = cursor;if (i >= elementCount)throw new NoSuchElementException();cursor = i + 1;return elementData(lastRet = i);}}final void checkForComodification() {if (modCount != expectedModCount)throw new ConcurrentModificationException();}
}

modCount 是容器维护的一个全局计数器，可以查看源码看到，在向容器添加或删除元素时，就会执行 modCount++ （记录全局的操作数）

创建迭代器 Itr 时会用 expectedModCount 来记录 modCount，在迭代器每次调用 next 方法时，先调用 checkForComodification() 方法来判断当前时刻的 modCount 与创建迭代器时记录的 expectedModCount 是否一致，如果不一致，那么就说明容器遭到修改，抛出 ConcurrentModificationException 异常。

解决方法有两种：

客户端加锁：在迭代过程持有容器的锁（会降低程序的可伸缩性、吞吐量和CPU利用率）
“克隆” 容器：在迭代线程克隆容器，并使用副本进行迭代，在克隆过程仍然需要对容器加锁（取决于容器的大小）

二、并发容器和框架

同步容器将所有对容器状态的访问都串行化，以实现他们的线程安全性。（这种方法的代价是严重降低并发性，当多个线程竞争容器的锁时，吞吐量将严重减低）

1.ConcurrentHashMap

ConcurrentHashMap 是线程安全且高效的HashMap。

1.1 为什么要使用 ConcurrentHashMap

在并发编程中使用 HashMap 可能导致程序死循环。而使用线程安全的 HashTable 效率又非常低下，基于以上两个原因，便有了 ConcurrentHashMap。

<1> 线程不安全的 HashMap

在多线程环境下，使用 HashMap 进行 put 操作会引起死循环，是因为多线程会导致 HashMap 的 Entry 链表形成环形数据结构（红黑树也可能会成环），导致 CUP 利用率接近 100%。

详细了解 HashMap 死循环问题：JAVA HASHMAP的死循环

<2> 效率低下的 HashTable

HashTable 容器使用 synchronized 保证线程安全，这是一种独占访问，导致在线程竞争激烈的情况下 HashTable 的效率非常低下。（当一个线程访问 HashTable 的同步方法，其他线程也访问 HashTable 的同步方法时，会进入阻塞或轮询状态）

<3> ConcurrentHashMap 的锁分段技术以及弱一致性可有效提升并发效率

ConcurrentHashMap 并不是将每个方法都在同一个锁上同步并使得每次只能有一个线程访问容器，而是使用一种粒度更细的加锁机制来实现更大程度的共享，容器里有多把锁，每一把锁用于锁容器其中一部分数据，那么当多线程访问容器里不同数据段的数据时，线程间就不会存在锁竞争，这就是锁分段技术；
ConcurrentHashMap 与其他并发容器增强了同步类：它们提供的迭代器不会抛出 ConcurrentModificationException，因此不需要再迭代过程中对容器加锁。ConcurrentHashMap 返回具有弱一致性的迭代器可以容忍并发修改，当创建迭代器时会遍历已有的元素，并可以在迭代器被构造后将修改操作反映给容器。
具体参考：ConcurrentHashMap遍历 — 弱一致性的迭代器（Iterator）实现原理

1.2 ConcurrentHashMap 的结构及操作源码解析

ConcurrentHashMap 的源码解析内容较多，我专门写了一篇进行分析：ConcurrentHashMap源码分析（JDK8版本）

2.CopyOnWriteArrayList

CopyOnWriteArrayList 用于替代同步 List，在某些情况下它提供了更好的并发性能，并且在迭代期间不需要对容器进行加锁或复制。

CopyOnWriteArrayList 底层实现不像 ConcurrentHashMap 那样复杂，相反却非常简单，CopyOnWriteArrayList 的线程安全是通过 ReentrantLock 显式锁对修改容器的方法进行加锁实现的，并且底层维护了一个Volatile修饰的数组array。

/** The lock protecting all mutators */
final transient ReentrantLock lock = new ReentrantLock();/** The array, accessed only via getArray/setArray. */
// 只能通过 getArray/setArray来操作array数组
private transient volatile Object[] array;

注意这里的 getArray、setArray方法也被final修饰，防治其他类重写该方法，从而只能得到唯一的数组变量。

/*** Gets the array.  Non-private so as to also be accessible* from CopyOnWriteArraySet class.*/
final Object[] getArray() {return array;
}/*** Sets the array.*/
final void setArray(Object[] a) {array = a;
}

我们接下来从源码角度来分析 CopyOnWriteArrayList 的特性：

2.1 为什么叫“写入时复制”容器

我们可以在一些 add、move 之类的修改容器方法中看到“写入时复制”的操作。

分析add(E e)方法：

/**
* Appends the specified element to the end of this list.** @param e element to be appended to this list* @return {@code true} (as specified by {@link Collection#add})*/
public boolean add(E e) {final ReentrantLock lock = this.lock;// 对该方法独占加锁lock.lock();try {// 1.获取底层array数组,可以在每个修改容器的方法中看到这段代码Object[] elements = getArray();int len = elements.length;// 2.在添加元素时,它会将原array的元素复制到新的数组并且大小+1Object[] newElements = Arrays.copyOf(elements, len + 1);// 存储新的元素newElements[len] = e;// 3.将新的数组设为底层的arraysetArray(newElements);return true;} finally {// 释放锁lock.unlock();}
}

正是因为1、2、3步代码才会实现“写入时复制”这种特性。

2.2 “写入时复制”这样的特性有什么用

写入时复制”容器的迭代器保留了一个指向底层基础数组（array）的引用，这个数组当前位于迭代器的起始位置，由于他不会被修改，因此在对其进行同步时只需要确保数组内容的可见性（所以使用了Volatile修饰array）。因此，多个线程可以同时对这个容器进行迭代，而不会彼此干扰或者与修改容器的线程相互干扰。

因为每当修改容器时都会复制底层数组，这需要一定的开销，所以仅当迭代操作远远多于修改操作时，才应该使用“写入时复制”容器。

CopyOnWriteArrayList 迭代器源码分析：

static final class COWIterator<E> implements ListIterator<E> {/** Snapshot of the array */// 这个就是指向底层基础数组的引用private final Object[] snapshot;/** Index of element to be returned by subsequent call to next.  */// 游标用来记录遍历数组的索引private int cursor;private COWIterator(Object[] elements, int initialCursor) {cursor = initialCursor;snapshot = elements;}public boolean hasNext() {return cursor < snapshot.length;}public boolean hasPrevious() {return cursor > 0;}@SuppressWarnings("unchecked")public E next() {if (! hasNext())throw new NoSuchElementException();return (E) snapshot[cursor++];}@SuppressWarnings("unchecked")public E previous() {if (! hasPrevious())throw new NoSuchElementException();return (E) snapshot[--cursor];}public int nextIndex() {return cursor;}public int previousIndex() {return cursor-1;}
}

可以发现在 CopyOnWriteArrayList 的迭代器中没有对elementCount的判断，因为array是不可变的，所以使用elementCount也是毫无意义的。

3.Queue - ConcurrentLinkedQueue

在并发编程中我们有时候需要使用线程安全的队列。如果我们要实现一个线程安全的队列有两种实现方式一种是使用阻塞算法，另一种是使用非阻塞算法。使用阻塞算法的队列可以用一个锁（入队和出队用同一把锁）或两个锁（入队和出队用不同的锁）等方式来实现，而非阻塞的实现方式则可以使用循环CAS的方式来实现，下面我们一起来研究下Doug Lea是如何使用非阻塞的方式来实现线程安全队列 ConcurrentLinkedQueue 的。

在JUC包下，并发容器的非阻塞方式实现一般都是由 CAS与Volatile修饰的变量来控制同步的。

在 ConcurrentLinkedQueue 中最重要的就是head和tail两个引用。

/*** A node from which the first live (non-deleted) node (if any)* can be reached in O(1) time.* Invariants:* - all live nodes are reachable from head via succ()* - head != null* - (tmp = head).next != tmp || tmp != head* Non-invariants:* - head.item may or may not be null.* - it is permitted for tail to lag behind head, that is, for tail*   to not be reachable from head!*/
private transient volatile Node<E> head;/*** A node from which the last node on list (that is, the unique* node with node.next == null) can be reached in O(1) time.* Invariants:* - the last node is always reachable from tail via succ()* - tail != null* Non-invariants:* - tail.item may or may not be null.* - it is permitted for tail to lag behind head, that is, for tail*   to not be reachable from head!* - tail.next may or may not be self-pointing to tail.*/
private transient volatile Node<E> tail;

那么我们通过源码来分析 ConcurrentLinkedQueue 是如何非阻塞地实现线程安全的，由于别的方法比较简单，这里只分析 offer(E e)和poll() 方法。

3.1 入队：offer(E e) 源码分析

ConcurrentLinkedQueue 的入队操作较别的 Queue 方法不太相似，一般的思路大概为：CAS将尾结点的next设为newNode->CAS将newNode设为新的尾结点，这样代码实现最多5行。但在 ConcurrentLinkedQueue 中却有不同的思路：
先找到尾结点->CAS将尾结点的next设为newNode->CAS将newNode设为新的尾结点（某些情况下），这里先记住在 ConcurrentLinkedQueue 中 tail 指向的不一定是尾结点，可以发现 ConcurrentLinkedQueue 的入队多了很多步骤，那么我们下面通过源码分析之后来说说这么做的好处。

/*** Inserts the specified element at the tail of this queue.* As the queue is unbounded, this method will never return {@code false}.** @return {@code true} (as specified by {@link Queue#offer})* @throws NullPointerException if the specified element is null*/
public boolean offer(E e) {// 检查echeckNotNull(e);final Node<E> newNode = new Node<E>(e);/*** 循环CAS直到入队成功*  t：记录tail结点*  p：记录尾结点*     初始情况 p和t都指向tail*/for (Node<E> t = tail, p = t;;) {// 获取p(尾结点)下一个结点 qNode<E> q = p.next;// 用来判断p是否为尾结点(判断尾结点依据是该结点的next结点是否为null)if (q == null) {// p is last node(p是尾结点)// 设置p节点的下一个节点为新节点,设置成功则casNext返回true；// 否则返回false，说明有其他线程更新过尾节点if (p.casNext(null, newNode)) {// Successful CAS is the linearization point// for e to become an element of this queue,// and for newNode to become "live".// p != t：说明当前tail指向的已经不是尾结点if (p != t) // hop two nodes at a time// 将新结点设为尾结点,失败表示其他线程更新了tail结点casTail(t, newNode);  // Failure is OK.return true;}// Lost CAS race to another thread; re-read next}/*** 在多线程操作时,出队操作会将旧的head变为自引用(便于GC回收)* 如果tail恰好指向了旧head指向的结点,那么就会出现这样的情况* 下面会图解分析*/else if (p == q)// We have fallen off list.  If tail is unchanged, it// will also be off-list, in which case we need to// jump to head, from which all live nodes are always// reachable.  Else the new tail is a better bet.// 出现这种情况：1.如果tail改变,p 就指向新的tail// 2.p 指向新的head结点,新的head结点才是能够使用的结点(live nodes)p = (t != (t = tail)) ? t : head;// CAS失败的线程,进入该分支 else// Check for tail updates after two hops./*** 此处为一个短路判断*  如果p==t,那么p就向前移动1位指向q(此时hops=0+1=1,p在tail前面)*    如果p!=t,说明hops=1即p与tail之间存在1个间距*       再次判断 t != (t = tail) tail是否改变*        如果改变,那么p指向新的tail*       如果没有改变,p向前移动1位指向q*          此时tail在p的前面,因为在第一个判断中*          第二个判断执行之后会将tail向前移动2位*          这里也是设计精髓*/p = (p != t && t != (t = tail)) ? t : q;}
}

入队操作的流程图：

那么通过分析入队操作我们便提出了这样的疑问：

tail结点不一定是尾结点的设计意图

在入队方法开头我也提到了 ConcurrentLinkedQueue 的设计思路与其他队列集合不太相似，那么不同点就在于tail结点不一定是尾结点。

在JDK1.7的实现中，doug lea使用hops变量来控制并减少tail节点的更新频率，并不是每次节点入队后都将 tail节点更新成尾节点，而是当tail节点和尾节点的距离大于等于常量HOPS的值（默认等于1）时才更新tail节点，tail和尾节点的距离越长使用CAS更新tail节点的次数就会越少，但是距离越长带来的负面效果就是每次入队时定位尾节点的时间就越长，因为循环体需要多循环一次来定位出尾节点，但是这样仍然能提高入队的效率，因为从本质上来看它通过增加对volatile变量的读操作来减少了对volatile变量的写操作，而对volatile变量的写操作开销要远远大于读操作，所以入队效率会有所提升。

在JDK 1.8的实现中，tail的更新时机是通过p和t是否相等来判断的，其实现结果和JDK 1.7相同，即当tail节点和尾节点的距离大于等于1时，更新tail。

3.2 出队：poll() 源码分析

和入队方法思路一样，并不是每次出队都会更新head结点，当head结点有元素直接元素并置为null，而不会更新head结点。只有head结点内为null时，出队操作才会更新head结点。

public E poll() {// 循环跳出的标记位restartFromHead:for (;;) {// h：记录head结点// p：记录头结点for (Node<E> h = head, p = h, q;;) {// 记录头结点的元素E item = p.item;// 头结点存在元素，那么CAS将头结点元素置为nullif (item != null && p.casItem(item, null)) {// Successful CAS is the linearization point// for item to be removed from this queue.// 如果CAS成功,那么判断当前头结点与head结点的距离// 如果hops>=1,那么就进入判断if (p != h) // hop two nodes at a time// 更新 head结点为头结点updateHead(h, ((q = p.next) != null) ? q : p);// 返回头结点元素    return item;}// 如果头结点为空或者被其他线程修改// 获取p的下一个结点，如果为空表示队列为空else if ((q = p.next) == null) {// 更新头结点updateHead(h, p);// 队列为空,返回nullreturn null;}// 出现旧head结点自引用,那么调到外层循环,重新开始else if (p == q)continue restartFromHead;// 分支2的判断q=p.next,说明p的下一个结点不为空,p向前移1位指向q    elsep = q;}}
}

出队操作的流程图：

可以发现图中有虚线连接的部分，那么这就是接下来要将的在入队和出队方法中都有的一个判断p==q这个情况。在注释中我提到只是因为poll时旧的头结点自引用导致的，那么究竟是怎么导致的，我们还是通过流程图来看。

在updateHead(h, ((q = p.next) != null) ? q : p); 方法中会调用如下代码块：

final void updateHead(Node<E> h, Node<E> p) {if (h != p && casHead(h, p))// 将h的next引用指向h// 即就是自己的next引用指向自己(自引用)h.lazySetNext(h);}

那么这一步就会出现下面的情况：

如果在这种情况下在调用入队方法，就会进入 p=q判断，让p指向新的head，找到尾结点，入队流程图如下：

使用这种延迟更新的方式来提高并发队列容器的效率，这样的做法非常妙。

4.BlockingQueue

4.1 什么是阻塞队列

阻塞队列（BlockingQueue）是一个支持两个附加操作的队列。这两个附加的操作的支持阻塞的插入和移除方法。

支持阻塞的插入方法：意思是在队列满时，队列会阻塞插入元素的线程，直到队列不满。
支持阻塞的移除方法：意思是在队列为空时，获取元素的线程会等待队列变为非空。

阻塞队列常用于生产者和消费者的场景，生产者是向队列里添加元素的线程，消费者是从队列里取元素的线程。阻塞队列就是生产者用于存放元素、消费者用来获取元素的容器。

在阻塞队列不可用时，两个附加操作还提供了4种处理方式。

抛出异常：是指当阻塞队列满时候，再往队列里插入元素，会抛出 IllegalStateException(“Queue full”) 异常。当队列为空时，从队列里获取元素时会抛出 NoSuchElementException 异常；
返回特殊值：插入方法会返回是否成功，成功则返回 true。移除方法，则是从队列里拿出一个元素，如果没有则返回 null；
一直阻塞：当阻塞队列满时，如果生产者线程往队列里 put 元素，队列会一直阻塞生产者线程，直到拿到数据，或者响应中断退出。当队列空时，消费者线程试图从队列里 take 元素，队列也会阻塞消费者线程，直到队列可用；
超时退出：当阻塞队列满时，队列会阻塞生产者线程一段时间，如果超过一定的时间，生产者线程就会退出。

4.2 Java里的阻塞队列

JDK1.8提供了 7 个阻塞队列

ArrayBlockingQueue：一个由数组结构组成的有界阻塞队列；

ArrayBlockingQueue 是一个用数组实现的有界阻塞队列。此队列按照先进先出（FIFO）的原则对元素进行排序。
LinkedBlockingQueue ：一个由链表结构组成的有界阻塞队列；

LinkedBlockingQueue 是一个用链表实现的有界阻塞队列。此队列的默认和最大长度为 Integer.MAX_VALUE。此队列按照先进先出的原则对元素进行排序。
PriorityBlockingQueue ：一个支持优先级排序的无界阻塞队列；

PriorityBlockingQueue是一个支持优先级的无界阻塞队列。默认情况下元素采取自然顺序升序排列。继承Comparable类实现compareTo()方法来指定元素排序规则，或者初始化PriorityBlockingQueue时，指定构造参数Comparator来对元素进行排序。需要注意的是不能保证同优先级元素的顺序。
DelayQueue：一个使用优先级队列实现的无界阻塞队列；

DelayQueue是一个支持延时获取元素的无界阻塞队列。队列使用PriorityQueue来实现。队列中的元素必须实现Delayed接口，在创建元素时可以指定多久才能从队列中获取当前元素。只有在延迟期满时才能从队列中提取元素。
运用场景：
- 缓存系统的设计：可以用DelayQueue保存缓存元素的有效期，使用一个线程循环查询DelayQueue，一旦能从DelayQueue中获取元素时，表示缓存有效期到了；
- 定时任务调度：使用DelayQueue保存当天将会执行的任务和执行时间，一旦从DelayQueue中获取到任务就开始执行，比如TimerQueue就是使用DelayQueue实现的；
DelayQueue中的对象必须实现Delayed接口，并重写getDelay和compareTo方法。
SynchronousQueue：一个不存储元素的阻塞队列；

SynchronousQueue是一个不存储元素的阻塞队列。每一个put操作必须等待一个take操作，否则不能继续添加元素。

SynchronousQueue可以看成是一个传球手，负责把生产者线程处理的数据直接传递给消费者线程。队列本身并不存储任何元素，非常适合传递性场景。SynchronousQueue的吞吐量高于LinkedBlockingQueue和ArrayBlockingQueue。
LinkedTransferQueue：一个由链表结构组成的无界阻塞队列；

LinkedTransferQueue是一个由链表结构组成的无界阻塞TransferQueue队列。相对于其他阻塞队列，LinkedTransferQueue多了tryTransfer和transfer方法。

(1) transfer 方法
如果当前有消费者正在等待接收元素（消费者使用take()方法或带时间限制的poll()方法时），transfer方法可以把生产者传入的元素立刻transfer（传输）给消费者。如果没有消费者在等待接收元素，transfer方法会将元素存放在队列的tail节点，并等到该元素被消费者消费了才返回。

(2) tryTransfer 方法
tryTransfer方法是用来试探生产者传入的元素是否能直接传给消费者。如果没有消费者等待接收元素，则返回false。和transfer方法的区别是tryTransfer方法无论消费者是否接收，方法立即返回，而transfer方法是必须等到消费者消费了才返回。
LinkedBlockingDeque：一个由链表结构组成的双向阻塞队列；

LinkedBlockingDeque是一个由链表结构组成的双向阻塞队列。所谓双向队列指的是可以从队列的两端插入和移出元素。双向队列因为多了一个操作队列的入口，在多线程同时入队时，也就减少了一半的竞争。相比其他的阻塞队列，LinkedBlockingDeque多了addFirst、addLast、offerFirst、offerLast、peekFirst和peekLast等方法。

4.3 阻塞队列实现原理

如果队列为空，消费者会一直等待，当生产者添加元素时，消费者是如何知道当前队列有元素的呢？

使用通知模式实现。当生产者往满的队列里添加元素时会阻塞住生产者，当消费者消费了一个队列的元素后，会通知生产者当前队列可用。

通过查看JDK1.8源码 ArrayBlockingQueue 使用了 Condition 来实现，分析如何实现的。

在 ArrayBlockingQueue 的成员变量中定义了生产者和消费者的 Condition （用来控制生产者和消费者线程）

/** Main lock guarding all access */
final ReentrantLock lock;/** Condition for waiting takes */
private final Condition notEmpty;/** Condition for waiting puts */
private final Condition notFull;

这些成员变量会在构造方法中初始化

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();
}

分析生产者线程调用put(E e)方法队列满的情况被阻塞

public void put(E e) throws InterruptedException {checkNotNull(e);final ReentrantLock lock = this.lock;// 中断加锁 可以响应中断lock.lockInterruptibly();try {// 使用while循环,保证生产线程唤醒之后会再次判断// 队列已满while (count == items.length)// 调用await方法让当前线程等待notFull.await();// 队列未满,就添加元素e            enqueue(e);} finally {// 最后释放锁lock.unlock();}
}

进入到enqueue(e)方法

private void enqueue(E x) {// 在put方法中加锁,调用该方法不需要同步// assert lock.getHoldCount() == 1;// assert items[putIndex] == null;final Object[] items = this.items;items[putIndex] = x;// 如果添加完元素之后,队列满了,那么下一次添加索引置为0if (++putIndex == items.length)putIndex = 0;// 队列元素+1    count++;// 如果队列有元素就会唤醒消费者线程notEmpty.signal();
}

分析消费者线程调用take()方法队列为空的情况被阻塞

public E take() throws InterruptedException {final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {// 循环判断队列元素是否为0while (count == 0)// 如果队列为空,那么消费者线程等待notEmpty.await();// 队列不为空返回出队元素    return dequeue();} finally {lock.unlock();}
}

进入到dequeue()方法

private E dequeue() {// assert lock.getHoldCount() == 1;// assert items[takeIndex] != null;final Object[] items = this.items;@SuppressWarnings("unchecked")E x = (E) items[takeIndex];// 取走元素的索引位置为nullitems[takeIndex] = null;if (++takeIndex == items.length)takeIndex = 0;// 队列数量-1    count--;// 更新迭代器中的元素保证并发时迭代器元素的正确性if (itrs != null)itrs.elementDequeued();// 唤醒生产者线程    notFull.signal();return x;
}

阻塞队列的迭代器分析
由于ArrayBlockingQueue迭代器较为复杂，这里就另写了一篇博客，感兴趣的可以移步参考：阻塞队列（ArrayBlockingQueue）迭代器源码分析

Condition 原理分析
从 Condition 源码来分析await和signal方法是怎样实现的，具体分析参考：
JUC Condition 源码分析