JUC并发—9.并发安全集合二

技术分享 12个月前 (02-22) 0 999+

关注

大纲

1.并发安全的数组列表CopyOnWriteArrayList

2.并发安全的链表队列ConcurrentLinkedQueue

3.并发编程中的阻塞队列概述

4.JUC的各种阻塞队列介绍

5.LinkedBlockingQueue的具体实现原理

6.基于两个队列实现的集群同步机制

 

1.并发安全的数组列表CopyOnWriteArrayList

(1)CopyOnWriteArrayList的初始化

(2)基于锁 + 写时复制机制实现的增删改操作

(3)使用写时复制的原因是读操作不加锁 + 不使用Unsafe读取数组元素

(4)对数组进行迭代时采用了副本快照机制

(5)核心思想是通过弱一致性提升读并发

(6)写时复制的总结

 

(1)CopyOnWriteArrayList的初始化

并发安全的HashMap是ConcurrentHashMap

并发安全的ArrayList是CopyOnWriteArrayList

并发安全的LinkedList是ConcurrentLinkedQueue

 

从CopyOnWriteArrayList的构造方法可知，CopyOnWriteArrayList基于Object对象数组实现。

 

这个Object对象数组array会使用volatile修饰，保证了多线程下的可见性。只要有一个线程修改了数组array，其他线程可以马上读取到最新值。

//A thread-safe variant of java.util.ArrayList in which all mutative operations  //(add, set, and so on) are implemented by making a fresh copy of the underlying array. public class CopyOnWriteArrayList<E> implements List<E>, RandomAccess, Cloneable, java.io.Serializable {      ...     //The lock protecting all mutators     final transient ReentrantLock lock = new ReentrantLock();      //The array, accessed only via getArray/setArray.     private transient volatile Object[] array;          //Creates an empty list.     public CopyOnWriteArrayList() {         setArray(new Object[0]);     }          //Sets the array.     final void setArray(Object[] a) {         array = a;     }     ... }

(2)基于锁 + 写时复制机制实现的增删改操作

一.使用独占锁解决对数组的写写并发问题

每个CopyOnWriteArrayList都有一个Object数组 + 一个ReentrantLock锁。在对Object数组进行增删改时，都要先获取锁，保证只有一个线程增删改。从而确保多线程增删改CopyOnWriteArrayList的Object数组是并发安全的。注意：获取锁的动作需要在执行getArray()方法前执行。

 

但因为获取独占锁，所以导致CopyOnWriteArrayList的写并发并性能不太好。而ConcurrentHashMap由于通过CAS设置 + 分段加锁，所以写并发性能很高。

 

二.使用写时复制机制解决对数组的读写并发问题

CopyOnWrite就是写时复制。写数据时不直接在当前数组里写，而是先把当前数组的数据复制到新数组里。然后再在新数组里写数据，写完数据后再将新数组赋值给array变量。这样原数组由于没有了array变量的引用，很快就会被JVM回收掉。

 

其中会使用System.arraycopy()方法和Arrays.copyOf()方法来复制数据到新数组，从Arrays.copyOf(elements, len + 1)可知，新数组的大小比原数组大小多1。

 

所以CopyOnWriteArrayList不需要进行数组扩容，这与ArrayList不一样。ArrayList会先初始化一个固定大小的数组，然后数组大小达到阈值时会扩容。

 

三.总结

为了解决CopyOnWriteArrayList的数组写写并发问题，使用了锁。

为了解决CopyOnWriteArrayList的数组读写并发问题，使用了写时复制。

所以CopyOnWriteArrayList可以保证多线程对数组写写 + 读写的并发安全。

//A thread-safe variant of java.util.ArrayList in which all mutative operations  //(add, set, and so on) are implemented by making a fresh copy of the underlying array. public class CopyOnWriteArrayList<E> implements List<E>, RandomAccess, Cloneable, java.io.Serializable {      ...     //The lock protecting all mutators     final transient ReentrantLock lock = new ReentrantLock();      //The array, accessed only via getArray/setArray.     private transient volatile Object[] array;          //Creates an empty list.     public CopyOnWriteArrayList() {         setArray(new Object[0]);     }          //Sets the array.     final void setArray(Object[] a) {         array = a;     }          //Gets the array. Non-private so as to also be accessible from CopyOnWriteArraySet class.     final Object[] getArray() {         return array;     }          //增：Appends the specified element to the end of this list.     public boolean add(E e) {         final ReentrantLock lock = this.lock;         lock.lock();         try {             Object[] elements = getArray();             int len = elements.length;             Object[] newElements = Arrays.copyOf(elements, len + 1);             newElements[len] = e;             setArray(newElements);             return true;         } finally {             lock.unlock();         }     }          //删：Removes the element at the specified position in this list.     //Shifts any subsequent elements to the left (subtracts one from their indices).       //Returns the element that was removed from the list.     public E remove(int index) {         final ReentrantLock lock = this.lock;         lock.lock();         try {             Object[] elements = getArray();             int len = elements.length;             E oldValue = get(elements, index);             int numMoved = len - index - 1;             if (numMoved == 0) {                 setArray(Arrays.copyOf(elements, len - 1));             } else {                 //先创建新数组，新数组的大小为len-1，比原数组的大小少1                 Object[] newElements = new Object[len - 1];                 //把原数组里从0开始拷贝index个元素到新数组里，并且从新数组的0位置开始放置                 System.arraycopy(elements, 0, newElements, 0, index);                 //把原数组从index+1开始拷贝numMoved个元素到新数组里，并且从新数组的index位置开始放置；                 System.arraycopy(elements, index + 1, newElements, index, numMoved);                 setArray(newElements);             }             return oldValue;         } finally {             lock.unlock();         }     }          //改：Replaces the element at the specified position in this list with the specified element.     public E set(int index, E element) {         final ReentrantLock lock = this.lock;         lock.lock();         try {             Object[] elements = getArray();             E oldValue = get(elements, index);              if (oldValue != element) {                 int len = elements.length;                 Object[] newElements = Arrays.copyOf(elements, len);                 newElements[index] = element;                 setArray(newElements);             } else {                 //Not quite a no-op; ensures volatile write semantics                 setArray(elements);             }             return oldValue;         } finally {             lock.unlock();         }     }     ... }

(3)使用写时复制的原因是读操作不加锁 + 不使用Unsafe读取数组元素

CopyOnWriteArrayList的增删改采用写时复制的原因在于get操作不需加锁。get操作就是先获取array数组，然后再通过index定位返回对应位置的元素。

 

由于在写数据的时候，首先更新的是复制了原数组数据的新数组。所以同一时间大量的线程读取数组数据时，都会读到原数组的数据，因此读写之间不会出现并发冲突的问题。

 

而且在写数据的时候，在更新完新数组之后，才会更新volatile修饰的数组变量。所以读操作只需要直接对volatile修饰的数组变量进行读取，就能获取最新的数组值。

 

如果不使用写时复制机制，那么即便有写线程先更新了array引用的数组中的元素，后续的读线程也只是具有对使用volatile修饰的array引用的可见性，而不会具有对array引用的数组中的元素的可见性。所以此时只要array引用没有发生改变，读线程还是会读到旧的元素，除非使用Unsafe.getObjectVolatile()方法来获取array引用的数组的元素。

public class CopyOnWriteArrayList<E> implements List<E>, RandomAccess, Cloneable, java.io.Serializable {     ...     //The array, accessed only via getArray/setArray.     private transient volatile Object[] array;      //Gets the array.  Non-private so as to also be accessible from CopyOnWriteArraySet class.     final Object[] getArray() {         return array;     }      public E get(int index) {         //先通过getArray()方法获取array数组，然后再通过get()方法定位到数组某位置的元素         return get(getArray(), index);     }          private E get(Object[] a, int index) {         return (E) a[index];     }     ... }

(4)对数组进行迭代时采用了副本快照机制

CopyOnWriteArrayList的Iterator迭代器里有一个快照数组snapshot，该数组指向的就是创建迭代器时CopyOnWriteArrayList的当前数组array。

 

所以使用CopyOnWriteArrayList的迭代器进行迭代时，会遍历快照数组。此时如果有其他线程更新了数组array，也不会影响迭代的过程。

public class CopyOnWriteArrayListDemo {     static List<String> list = new CopyOnWriteArrayList<String>();     public static void main(String[] args) {         list.add("k");         System.out.println(list);          Iterator<String> iterator = list.iterator();         while (iterator.hasNext()) {             System.out.println(iterator.next());         }     } }  public class CopyOnWriteArrayList<E> implements List<E>, RandomAccess, Cloneable, java.io.Serializable {     ...     public Iterator<E> iterator() {         return new COWIterator<E>(getArray(), 0);     }     ...          static final class COWIterator<E> implements ListIterator<E> {         private final Object[] snapshot;         private int cursor;         private COWIterator(Object[] elements, int initialCursor) {             cursor = initialCursor;             snapshot = elements;         }         ...     }     ... }

(5)核心思想是通过最终一致性提升读并发

CopyOnWriteArrayList的核心思想是通过弱一致性来提升读写并发的能力。

 

CopyOnWriteArrayList基于写时复制机制存在的最大问题是最终一致性。

 

多个线程并发读写数组，写线程已将新数组修改好，但还没设置给array。此时其他读线程读到的(get或者迭代)都是数组array的数据，于是在同一时刻，读线程和写线程看到的数据是不一致的。这就是写时复制机制存在的问题：最终一致性或弱一致性。

 

(6)写时复制的总结

一.优点

读读不互斥，读写不互斥，写写互斥。同一时间只有一个线程可以写，写的同时允许其他线程来读。

 

二.缺点

空间换时间，写的时候内存里会出现一模一样的副本，对内存消耗大。通过数组副本可以保证大量的读不需要和写互斥。如果数组很大，可能要考虑内存占用会是数组大小的几倍。此外使用数组副本来统计数据，会存在统计数据不一致的问题。

 

三.使用场景

适用于读多写少的场景，这样大量的读操作不会被写操作影响，而且不要求统计数据具有实时性。

 

2.并发安全的链表队列ConcurrentLinkedQueue

(1)ConcurrentLinkedQueue的介绍

(2)ConcurrentLinkedQueue的构造方法

(3)ConcurrentLinkedQueue的offer()方法

(4)ConcurrentLinkedQueue的poll()方法

(5)ConcurrentLinkedQueue的peak()方法

(6)ConcurrentLinkedQueue的size()方法

 

(1)ConcurrentLinkedQueue的介绍

ConcurrentLinkedQueue是一种并发安全且非阻塞的链表队列(无界队列)。

 

ConcurrentLinkedQueue采用CAS机制来保证多线程操作队列时的并发安全。

 

链表队列会采用先进先出的规则来对结点进行排序。每次往链表队列添加元素时，都会添加到队列的尾部。每次需要获取元素时，都会直接返回队列头部的元素。

 

并发安全的HashMap是ConcurrentHashMap

并发安全的ArrayList是CopyOnWriteArrayList

并发安全的LinkedList是ConcurrentLinkedQueue

 

(2)ConcurrentLinkedQueue的构造方法

ConcurrentLinkedQueue是基于链表实现的，链表结点为其内部类Node。

 

ConcurrentLinkedQueue的构造方法会初始化链表的头结点和尾结点为同一个值为null的Node对象。

 

Node结点通过next指针指向下一个Node结点，从而组成一个单向链表。而ConcurrentLinkedQueue的head和tail两个指针指向了链表的头和尾结点。

public class ConcurrentLinkedQueueDemo {     public static void main(String[] args) {         ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<String>();         queue.offer("张三");//向队尾添加元素         queue.offer("李四");//向队尾添加元素         queue.offer("王五");//向队尾添加元素         System.out.println(queue.peek());//返回队头的元素不出队         System.out.println(queue.poll());//返回队头的元素而且出队         System.out.println(queue.peek());//返回队头的元素不出队     } }  //An unbounded thread-safe queue based on linked nodes. //This queue orders elements FIFO (first-in-first-out). //The head of the queue is that element that has been on the queue the longest time. //The tail of the queue is that element that has been on the queue the shortest time.  //New elements are inserted at the tail of the queue,  //and the queue retrieval operations obtain elements at the head of the queue. //A ConcurrentLinkedQueue is an appropriate choice when many threads will share access to a common collection.  //Like most other concurrent collection implementations, this class does not permit the use of null elements. public class ConcurrentLinkedQueue<E> extends AbstractQueue<E> implements Queue<E>, java.io.Serializable {     ...     private transient volatile Node<E> head;     private transient volatile Node<E> tail;          //构造方法，初始化链表队列的头结点和尾结点为同一个值为null的Node对象     //Creates a ConcurrentLinkedQueue that is initially empty.     public ConcurrentLinkedQueue() {         head = tail = new Node<E>(null);     }          private static class Node<E> {         volatile E item;         volatile Node<E> next;                 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);             }         }                 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);         }     }     ... }

(3)ConcurrentLinkedQueue的offer()方法

其中关键的代码就是"p.casNext(null, newNode))"，就是把p的next指针由原来的指向空设置为指向新的结点，并且通过CAS确保同一时间只有一个线程可以成功执行这个操作。

 

注意：更新tail指针并不是实时更新的，而是隔一个结点再更新。这样可以减少CAS指令的执行次数，从而降低CAS操作带来的性能影响。

插入第一个元素后，tail指针指向倒数第二个节点。

插入第二个元素后，tail指针指向最后一个节点。

插入第三个元素后，tail指针指向倒数第二个节点。

插入第四个元素后，tail指针指向最后一个节点。

//An unbounded thread-safe queue based on linked nodes. public class ConcurrentLinkedQueue<E> extends AbstractQueue<E> implements Queue<E>, java.io.Serializable {     ...     private transient volatile Node<E> head;     private transient volatile Node<E> tail;          private static final sun.misc.Unsafe UNSAFE;     private static final long headOffset;     private static final long tailOffset;     static {         try {             UNSAFE = sun.misc.Unsafe.getUnsafe();             Class<?> k = ConcurrentLinkedQueue.class;             headOffset = UNSAFE.objectFieldOffset(k.getDeclaredField("head"));             tailOffset = UNSAFE.objectFieldOffset(k.getDeclaredField("tail"));         } catch (Exception e) {             throw new Error(e);         }     }          //构造方法，初始化链表队列的头结点和尾结点为同一个值为null的Node对象     //Creates a ConcurrentLinkedQueue that is initially empty.     public ConcurrentLinkedQueue() {         head = tail = new Node<E>(null);     }          public boolean offer(E e) {         checkNotNull(e);         final Node<E> newNode = new Node<E>(e);         //插第一个元素时, tail和head都是初始化时的空节点, p也指向该空节点, q是该空节点的next元素;         //很明显q是null, p.casNext后, p的next设为第一个元素, 此时p和t相等, tail的next是第一个元素；         //由于p==t, 于是返回true, head和tail还是指向初始化时的空节点, tail指针指向的是倒数第二个节点;         //插第二个元素时, q成为第一个元素，不为null了, 而且p指向tail, tail的next是第一个元素, 所以p != q;         //由于此时p和t还是一样的, 所以会将q赋值给p, 也就是p指向第一个元素了, 再次进行新一轮循环;         //新一轮循环时, q指向第一个元素的next成为null, 所以会对第一个元素执行casNext操作;         //也就是将第二个元素设为第一个元素的next, 设完后由于p和t不相等了, 会执行casTail设第二个元素为tail;         //插入第三个元素时, 又会和插入第一个元素一样了, 这时tail指针指向的是倒数第二个节点;         //插入第四个元素时, 和插入第二个元素一样, 这是tail指针指向的是最后一个节点;         for (Node<E> t = tail, p = t;;) {             Node<E> q = p.next;//p是尾结点，q是尾结点的下一个结点             if (q == null) {                 //插入第一个元素时执行的代码                 if (p.casNext(null, newNode)) {//将新结点设置为尾结点的下一个结点                     if (p != t) {//隔一个结点再CAS更新tail指针                         casTail(t, newNode);                     }                     return true;                 }             } else if (p == q) {                 p = (t != (t = tail)) ? t : head;             } else {                 //插入第二个元素时执行的代码                 p = (p != t && t != (t = tail)) ? t : q;             }         }     }          private boolean casTail(Node<E> cmp, Node<E> val) {         return UNSAFE.compareAndSwapObject(this, tailOffset, cmp, val);     }     ... }

(4)ConcurrentLinkedQueue的poll()方法

poll()方法会将链表队列的头结点出队。

 

注意：更新head指针时也不是实时更新的，而是隔一个结点再更新。这样可以减少CAS指令的执行次数，从而降低CAS操作带来的性能影响。

//An unbounded thread-safe queue based on linked nodes. public class ConcurrentLinkedQueue<E> extends AbstractQueue<E> implements Queue<E>, java.io.Serializable {     ...     private transient volatile Node<E> head;     private transient volatile Node<E> tail;          private static final sun.misc.Unsafe UNSAFE;     private static final long headOffset;     private static final long tailOffset;     static {         try {             UNSAFE = sun.misc.Unsafe.getUnsafe();             Class<?> k = ConcurrentLinkedQueue.class;             headOffset = UNSAFE.objectFieldOffset(k.getDeclaredField("head"));             tailOffset = UNSAFE.objectFieldOffset(k.getDeclaredField("tail"));         } catch (Exception e) {             throw new Error(e);         }     }          //构造方法，初始化链表队列的头结点和尾结点为同一个值为null的Node对象     //Creates a ConcurrentLinkedQueue that is initially empty.     public ConcurrentLinkedQueue() {         head = tail = new Node<E>(null);     }          public E poll() {         restartFromHead:         for (;;) {             for (Node<E> h = head, p = h, q;;) {                 E item = p.item;                 if (item != null && p.casItem(item, null)) {                     if (p != h) {//隔一个结点才CAS更新head指针                         updateHead(h, ((q = p.next) != null) ? q : p);                     }                     return item;                 } else if ((q = p.next) == null) {                     updateHead(h, p);                     return null;                 } else if (p == q) {                     continue restartFromHead;                 } else {                     p = q;                 }             }         }     }        final void updateHead(Node<E> h, Node<E> p) {         if (h != p && casHead(h, p)) {             h.lazySetNext(h);         }     }        private boolean casHead(Node<E> cmp, Node<E> val) {         return UNSAFE.compareAndSwapObject(this, headOffset, cmp, val);     }     ... }

(5)ConcurrentLinkedQueue的peak()方法

peek()方法会获取链表的头结点，但是不会出队。

//An unbounded thread-safe queue based on linked nodes. public class ConcurrentLinkedQueue<E> extends AbstractQueue<E> implements Queue<E>, java.io.Serializable {     ...     private transient volatile Node<E> head;          public E peek() {         restartFromHead:         for (;;) {             for (Node<E> h = head, p = h, q;;) {                 E item = p.item;                 if (item != null || (q = p.next) == null) {                     updateHead(h, p);                     return item;                 } else if (p == q) {                     continue restartFromHead;                 } else {                     p = q;                 }             }         }     }          final void updateHead(Node<E> h, Node<E> p) {         if (h != p && casHead(h, p)) {             h.lazySetNext(h);         }     }          private boolean casHead(Node<E> cmp, Node<E> val) {         return UNSAFE.compareAndSwapObject(this, headOffset, cmp, val);     }     ... }

(6)ConcurrentLinkedQueue的size()方法

size()方法主要用来返回链表队列的大小，查看链表队列有多少个元素。size()方法不会加锁，会直接从头节点开始遍历链表队列中的每个结点。

//An unbounded thread-safe queue based on linked nodes. public class ConcurrentLinkedQueue<E> extends AbstractQueue<E> implements Queue<E>, java.io.Serializable {     ...     public int size() {         int count = 0;         for (Node<E> p = first(); p != null; p = succ(p))             if (p.item != null) {                 if (++count == Integer.MAX_VALUE) {                     break;                 }             }         }         return count;     }         //Returns the first live (non-deleted) node on list, or null if none.     //This is yet another variant of poll/peek; here returning the first node, not element.     //We could make peek() a wrapper around first(), but that would cost an extra volatile read of item,     //and the need to add a retry loop to deal with the possibility of losing a race to a concurrent poll().      Node<E> first() {         restartFromHead:         for (;;) {             for (Node<E> h = head, p = h, q;;) {                 boolean hasItem = (p.item != null);                 if (hasItem || (q = p.next) == null) {                     updateHead(h, p);                     return hasItem ? p : null;                 } else if (p == q) {                     continue restartFromHead;                 } else {                     p = q;                 }             }         }     }          //Returns the successor of p, or the head node if p.next has been linked to self,      //which will only be true if traversing with a stale pointer that is now off the list.     final Node<E> succ(Node<E> p) {         Node<E> next = p.next;         return (p == next) ? head : next;     }     ... }

如果在遍历的过程中，有线程执行入队或者是出队的操作，此时会怎样？

 

从队头开始遍历，遍历到一半时：如果有线程在队列尾部进行入队操作，此时的遍历能及时看到新添加的元素。因为入队操作就是设置队列尾部节点的next指针指向新添加的结点，而入队时设置next指针属于volatile写，因此遍历时是可以看到的。如果有线程从队列头部进行出队操作，此时的遍历则无法感知有元素出队了。

 

所以可以总结出这些并发安全的集合：ConcurrentHashMap、CopyOnWriteArrayList和ConcurrentLinkedQueue，为了优化多线程下的并发性能，会牺牲掉统计数据的一致性。为了保证多线程写的高并发性能，会大量采用CAS进行无锁化操作。同时会让很多读操作比如常见的size()操作，不使用锁。因此使用这些并发安全的集合时，要考虑并发下的统计数据的不一致问题。

 

3.并发编程中的阻塞队列概述

(1)什么是阻塞队列

(2)阻塞队列提供的方法

(3)阻塞队列的应用

 

(1)什么是阻塞队列

队列是一种只允许在一端进行移除操作、在另一端进行插入操作的线性表，队列中允许插入的一端称为队尾，允许移除的一端称为队头。

 

阻塞队列就是在队列的基础上增加了两个操作：

一.支持阻塞插入

在队列满时会阻塞继续往队列中添加数据的线程，直到队列中有元素被释放。

二.支持阻塞移除

在队列空时会阻塞从队列中获取元素的线程，直到队列中添加了新的元素。

 

阻塞队列其实实现了一个生产者/消费者模型：生产者往队列中添加数据，消费者从队列中获取数据。队列满了阻塞生产者，队列空了阻塞消费者。

 

阻塞队列中的元素可能会使用数组或者链表等来进行存储。一个队列中能容纳多少个元素取决于队列的容量大小，因此阻塞队列也分为有界队列和无界队列。

 

有界队列指有固定大小的队列，无界队列指没有固定大小的队列。实际上无界队列也是有大小限制的，只是大小限制为非常大，可认为无界。

 

注意：在无界队列中，由于理论上不存在队列满的情况，所以不存在阻塞。

 

阻塞队列在很多地方都会用到，比如线程池、ZooKeeper。一般使用阻塞队列来实现生产者/消费者模型。

 

(2)阻塞队列提供的方法

阻塞队列的操作有插入、移除、检查，在队列满或者空时会有不同的效果。

 

一.抛出异常

当队列满的时候通过add(e)方法添加元素，会抛出异常。

 

当队列空的时候调用remove(e)方法移除元素，也会抛出异常。

 

二.返回特殊值

调用offer(e)方法向队列入队元素时，会返回添加结果true或false。

 

调用poll()方法从队列出队元素时，会从队列取出一个元素或null。

 

三.一直阻塞

在队列满了的情况下，调用插入方法put(e)向队列中插入元素时，队列会阻塞插入元素的线程，直到队列不满或者响应中断才退出阻塞。

 

在队列空了的情况下，调用移除方法take()从队列移除元素时，队列会阻塞移除元素的线程，直到队列不为空时唤醒线程。

 

四.超时退出

超时退出其实就是在offer()和poll()方法中增加了阻塞的等待时间。

(3)阻塞队列的应用

阻塞队列可以理解为线程级别的消息队列。

消息中间件可以理解为进程级别的消息队列。

所以可以通过阻塞队列来缓存线程的请求，从而达到流量削峰的目的。

4.JUC的各种阻塞队列介绍

(1)基于数组的阻塞队列ArrayBlockingQueue

(2)基于链表的阻塞队列LinkedBlockingQueue

(3)优先级阻塞队列PriorityBlockingQueue

(4)延迟阻塞队列DelayQueue

(5)无存储结构的阻塞队列SynchronousQueue

(6)阻塞队列结合体LinkedTransferQueue

(7)双向阻塞队列LinkedBlockingDeque

(1)基于数组的阻塞队列ArrayBlockingQueue

ArrayBlockingQueue是一个基于数组实现的阻塞队列。其构造方法可以指定：数组的长度、公平还是非公平、数组的初始集合。

ArrayBlockingQueue会通过ReentrantLock来解决线程竞争的问题，以及采用Condition来解决线程的唤醒与阻塞的问题。

//A bounded BlockingQueue backed by an array.   //This queue orders elements FIFO (first-in-first-out).   //The head of the queue is that element that has been on the queue the longest time.   //The tail of the queue is that element that has been on the queue the shortest time.  //New elements are inserted at the tail of the queue,  //and the queue retrieval operations obtain elements at the head of the queue. public class ArrayBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable {      //The queued items     final Object[] items;      //items index for next take, poll, peek or remove     int takeIndex;      //items index for next put, offer, or add     int putIndex;      //Number of elements in the queue     int count;      //Main lock guarding all access     final ReentrantLock lock;      //Condition for waiting takes     private final Condition notEmpty;      //Condition for waiting puts     private final Condition notFull;          //Creates an ArrayBlockingQueue with the given (fixed) capacity and default access policy.     public ArrayBlockingQueue(int capacity) {         this(capacity, false);     }      //Creates an ArrayBlockingQueue with the given (fixed) capacity and the specified access policy.     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();     }          //Inserts the specified element at the tail of this queue,      //waiting for space to become available if the queue is full.     public void put(E e) throws InterruptedException {         checkNotNull(e);         final ReentrantLock lock = this.lock;         lock.lockInterruptibly();         try {             while (count == items.length) {                 notFull.await();             }             enqueue(e);         } finally {             lock.unlock();         }     }          //Inserts element at current put position, advances, and signals.     //Call only when holding lock.     private void enqueue(E x) {         final Object[] items = this.items;         items[putIndex] = x;         if (++putIndex == items.length) {             putIndex = 0;         }         count++;         notEmpty.signal();     }          public E take() throws InterruptedException {         final ReentrantLock lock = this.lock;         lock.lockInterruptibly();         try {             while (count == 0) {                 notEmpty.await();             }             return dequeue();         } finally {             lock.unlock();         }     }          //Returns the number of elements in this queue.     public int size() {         final ReentrantLock lock = this.lock;         lock.lock();         try {             return count;         } finally {             lock.unlock();         }     }     ... }

(2)基于链表的阻塞队列LinkedBlockingQueue

LinkedBlockingQueue是一个基于链表实现的阻塞队列，它可以不指定阻塞队列的长度，它的默认长度是Integer.MAX_VALUE。由于这个默认长度非常大，一般也称LinkedBlockingQueue为无界队列。

//An optionally-bounded BlockingQueue based on linked nodes. //This queue orders elements FIFO (first-in-first-out). //The head of the queue is that element that has been on the queue the longest time. //The tail of the queue is that element that has been on the queue the shortest time.  //New elements are inserted at the tail of the queue,  //and the queue retrieval operations obtain elements at the head of the queue. //Linked queues typically have higher throughput than array-based queues  //but less predictable performance in most concurrent applications. public class LinkedBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable {     ...     //The capacity bound, or Integer.MAX_VALUE if none     private final int capacity;     //Current number of elements     private final AtomicInteger count = new AtomicInteger();          //Head of linked list.     transient Node<E> head;     //Tail of linked list.     private transient Node<E> last;          //Lock held by take, poll, etc     private final ReentrantLock takeLock = new ReentrantLock();     //Lock held by put, offer, etc     private final ReentrantLock putLock = new ReentrantLock();          //Wait queue for waiting takes     private final Condition notEmpty = takeLock.newCondition();     //Wait queue for waiting puts     private final Condition notFull = putLock.newCondition();          //Creates a LinkedBlockingQueue with a capacity of Integer#MAX_VALUE.     public LinkedBlockingQueue() {         this(Integer.MAX_VALUE);     }          //Creates a LinkedBlockingQueue with the given (fixed) capacity.     public LinkedBlockingQueue(int capacity) {         if (capacity <= 0) throw new IllegalArgumentException();         this.capacity = capacity;         last = head = new Node<E>(null);     }              //Inserts the specified element at the tail of this queue,      //waiting if necessary for space to become available.     public void put(E e) throws InterruptedException {         if (e == null) throw new NullPointerException();         int c = -1;          Node<E> node = new Node<E>(e);         final ReentrantLock putLock = this.putLock;         final AtomicInteger count = this.count;         putLock.lockInterruptibly();         try {             while (count.get() == capacity) {                 notFull.await();             }             enqueue(node);             c = count.getAndIncrement();             if (c + 1 < capacity) {                 notFull.signal();             }         } finally {             putLock.unlock();         }         if (c == 0) {             signalNotEmpty();         }     }          //Links node at end of queue.     private void enqueue(Node<E> node) {         last = last.next = node;     }          //Signals a waiting take. Called only from put/offer (which do not otherwise ordinarily lock takeLock.)     private void signalNotEmpty() {         final ReentrantLock takeLock = this.takeLock;         takeLock.lock();         try {             notEmpty.signal();         } finally {             takeLock.unlock();         }     }          public E take() throws InterruptedException {         E x;         int c = -1;         final AtomicInteger count = this.count;         final ReentrantLock takeLock = this.takeLock;         takeLock.lockInterruptibly();         try {             while (count.get() == 0) {                 notEmpty.await();             }             x = dequeue();             c = count.getAndDecrement();             if (c > 1) {                 notEmpty.signal();             }         } finally {             takeLock.unlock();         }         if (c == capacity) {             signalNotFull();         }         return x;     }      private E dequeue() {         Node<E> h = head;         Node<E> first = h.next;         h.next = h; // help GC         head = first;         E x = first.item;         first.item = null;         return x;     }          public int size() {         return count.get();     }     ... }

(3)优先级阻塞队列PriorityBlockingQueue

PriorityBlockingQueue是一个支持自定义元素优先级的无界阻塞队列。默认情况下添加的元素采用自然顺序升序排列，当然可以通过实现元素的compareTo()方法自定义优先级规则。

PriorityBlockingQueue是基于数组实现的，这个数组会自动进行动态扩容。在应用方面，消息中间件可以基于优先级阻塞队列来实现。

//An unbounded BlockingQueue that uses the same ordering rules as class PriorityQueue and supplies blocking retrieval operations.   //While this queue is logically unbounded, attempted additions may fail due to resource exhaustion (causing OutOfMemoryError).  //This class does not permit null elements. public class PriorityBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable {      //Priority queue represented as a balanced binary heap: the two children of queue[n] are queue[2*n+1] and queue[2*(n+1)].     //The priority queue is ordered by comparator, or by the elements' natural ordering,      //if comparator is null: For each node n in the heap and each descendant d of n, n <= d.     //The element with the lowest value is in queue[0], assuming the queue is nonempty.     private transient Object[] queue;      //The number of elements in the priority queue.     private transient int size;      //The comparator, or null if priority queue uses elements' natural ordering.     private transient Comparator<? super E> comparator;      //Lock used for all public operations     private final ReentrantLock lock;      //Condition for blocking when empty     private final Condition notEmpty;      //Spinlock for allocation, acquired via CAS.     private transient volatile int allocationSpinLock;          //Creates a PriorityBlockingQueue with the default initial capacity (11) that      //orders its elements according to their Comparable natural ordering.     public PriorityBlockingQueue() {         this(DEFAULT_INITIAL_CAPACITY, null);     }      //Creates a PriorityBlockingQueue with the specified initial capacity that      //orders its elements according to their Comparable natural ordering.     public PriorityBlockingQueue(int initialCapacity) {         this(initialCapacity, null);     }      //Creates a PriorityBlockingQueue with the specified initial capacity that orders its elements according to the specified comparator.     public PriorityBlockingQueue(int initialCapacity, Comparator<? super E> comparator) {         if (initialCapacity < 1) {             throw new IllegalArgumentException();         }         this.lock = new ReentrantLock();         this.notEmpty = lock.newCondition();         this.comparator = comparator;         this.queue = new Object[initialCapacity];     }          //Inserts the specified element into this priority queue.     //As the queue is unbounded, this method will never block.     public void put(E e) {         offer(e); // never need to block     }          //Inserts the specified element into this priority queue.     //As the queue is unbounded, this method will never return false.     public boolean offer(E e) {         if (e == null) {             throw new NullPointerException();         }         final ReentrantLock lock = this.lock;         lock.lock();         int n, cap;         Object[] array;         while ((n = size) >= (cap = (array = queue).length)) {             tryGrow(array, cap);         }         try {             Comparator<? super E> cmp = comparator;             if (cmp == null) {                 siftUpComparable(n, e, array);             } else {                 siftUpUsingComparator(n, e, array, cmp);             }             size = n + 1;             notEmpty.signal();         } finally {             lock.unlock();         }         return true;     }          //Tries to grow array to accommodate at least one more element (but normally expand by about 50%),      //giving up (allowing retry) on contention (which we expect to be rare). Call only while holding lock.     private void tryGrow(Object[] array, int oldCap) {         lock.unlock(); // must release and then re-acquire main lock         Object[] newArray = null;         if (allocationSpinLock == 0 && UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset, 0, 1)) {         try {             int newCap = oldCap + ((oldCap < 64) ? (oldCap + 2) : (oldCap >> 1));                 if (newCap - MAX_ARRAY_SIZE > 0) {                     int minCap = oldCap + 1;                     if (minCap < 0 || minCap > MAX_ARRAY_SIZE) {                         throw new OutOfMemoryError();                     }                     newCap = MAX_ARRAY_SIZE;                 }                 if (newCap > oldCap && queue == array) {                     newArray = new Object[newCap];                 }             } finally {                 allocationSpinLock = 0;             }         }         if (newArray == null) {// back off if another thread is allocating             Thread.yield();         }         lock.lock();         if (newArray != null && queue == array) {             queue = newArray;             System.arraycopy(array, 0, newArray, 0, oldCap);         }     }          private static <T> void siftUpComparable(int k, T x, Object[] array) {         Comparable<? super T> key = (Comparable<? super T>) x;         while (k > 0) {             int parent = (k - 1) >>> 1;             Object e = array[parent];             if (key.compareTo((T) e) >= 0) {                 break;             }             array[k] = e;             k = parent;         }         array[k] = key;     }      private static <T> void siftUpUsingComparator(int k, T x, Object[] array, Comparator<? super T> cmp) {         while(k > 0) {             int parent = (k - 1) >>> 1;             Object e = array[parent];             if (cmp.compare(x, (T) e) >= 0) {                 break;             }             array[k] = e;             k = parent;         }         array[k] = x;     }          public E take() throws InterruptedException {         final ReentrantLock lock = this.lock;         lock.lockInterruptibly();         E result;         try {             while ( (result = dequeue()) == null) {                 notEmpty.await();             }         } finally {             lock.unlock();         }         return result;     }          public int size() {         final ReentrantLock lock = this.lock;         lock.lock();         try {             return size;         } finally {             lock.unlock();         }     }     ... }

(4)延迟阻塞队列DelayQueue

DelayQueue是一个支持延迟获取元素的无界阻塞队列，它是基于优先级队列PriorityQueue实现的。

往DelayQueue队列插入元素时，可以按照自定义的delay时间进行排序。也就是队列中的元素顺序是按照到期时间排序的，只有delay时间小于或等于0的元素才能够被取出。

DelayQueue的应用场景有：

一.订单超时支付需要自动取消订单

二.任务超时处理需要自动丢弃任务

三.消息中间件的实现

//An unbounded BlockingQueue of Delayed elements, in which an element can only be taken when its delay has expired. //The head of the queue is that Delayed element whose delay expired furthest in the past. //If no delay has expired there is no head and poll will return null.  //Expiration occurs when an element's getDelay(TimeUnit.NANOSECONDS) method returns a value less than or equal to zero. //Even though unexpired elements cannot be removed using take or poll, they are otherwise treated as normal elements.  //For example, the size method returns the count of both expired and unexpired elements. //This queue does not permit null elements. public class DelayQueue<E extends Delayed> extends AbstractQueue<E> implements BlockingQueue<E> {     private final transient ReentrantLock lock = new ReentrantLock();     private final PriorityQueue<E> q = new PriorityQueue<E>();          //Thread designated to wait for the element at the head of the queue.     //When a thread becomes the leader, it waits only for the next delay to elapse, but other threads await indefinitely.     //The leader thread must signal some other thread before returning from take() or poll(...),      //unless some other thread becomes leader in the interim.       //Whenever the head of the queue is replaced with an element with an earlier expiration time,      //the leader field is invalidated by being reset to null, and some waiting thread,      //but not necessarily the current leader, is signalled.      //So waiting threads must be prepared to acquire and lose leadership while waiting.     private Thread leader = null;      //Condition signalled when a newer element becomes available at the head of the queue or a new thread may need to become leader.     private final Condition available = lock.newCondition();      //Creates a new {@code DelayQueue} that is initially empty.     public DelayQueue() {                  }          //Inserts the specified element into this delay queue.      //As the queue is unbounded this method will never block.     public void put(E e) {         offer(e);     }          //Inserts the specified element into this delay queue.     public boolean offer(E e) {         final ReentrantLock lock = this.lock;         lock.lock();         try {             q.offer(e);             if (q.peek() == e) {                 leader = null;                 available.signal();             }             return true;         } finally {             lock.unlock();         }     }          //Retrieves and removes the head of this queue,      //waiting if necessary until an element with an expired delay is available on this queue.     public E take() throws InterruptedException {         final ReentrantLock lock = this.lock;         lock.lockInterruptibly();         try {             for (;;) {                 E first = q.peek();                 if (first == null) {                     available.await();                 } else {                     long delay = first.getDelay(NANOSECONDS);                     if (delay <= 0) {                         return q.poll();                     }                       first = null; // don't retain ref while waiting                     if (leader != null) {                         available.await();                     } else {                         Thread thisThread = Thread.currentThread();                         leader = thisThread;                         try {                             available.awaitNanos(delay);                         } finally {                             if (leader == thisThread) {                                 leader = null;                             }                         }                     }                 }             }         } finally {             if (leader == null && q.peek() != null) {                 available.signal();             }             lock.unlock();         }     }     ... }  public class PriorityQueue<E> extends AbstractQueue<E> implements java.io.Serializable {     //Priority queue represented as a balanced binary heap: the two children of queue[n] are queue[2*n+1] and queue[2*(n+1)].     //The priority queue is ordered by comparator, or by the elements' natural ordering, if comparator is null:      //For each node n in the heap and each descendant d of n, n <= d.       //The element with the lowest value is in queue[0], assuming the queue is nonempty.     transient Object[] queue;           //The number of elements in the priority queue.     private int size = 0;      //The comparator, or null if priority queue uses elements' natural ordering.     private final Comparator<? super E> comparator;          public E peek() {         return (size == 0) ? null : (E) queue[0];     }          //Inserts the specified element into this priority queue.     public boolean offer(E e) {         if (e == null) {             throw new NullPointerException();         }         modCount++;         int i = size;         if (i >= queue.length) {             grow(i + 1);         }         size = i + 1;         if (i == 0) {             queue[0] = e;         } else {             siftUp(i, e);         }         return true;     }          //Increases the capacity of the array.     private void grow(int minCapacity) {         int oldCapacity = queue.length;         // Double size if small; else grow by 50%         int newCapacity = oldCapacity + ((oldCapacity < 64) ? (oldCapacity + 2) : (oldCapacity >> 1));         // overflow-conscious code         if (newCapacity - MAX_ARRAY_SIZE > 0) {             newCapacity = hugeCapacity(minCapacity);         }         queue = Arrays.copyOf(queue, newCapacity);     }          private void siftUp(int k, E x) {         if (comparator != null) {             siftUpUsingComparator(k, x);         } else {             siftUpComparable(k, x);         }     }      @SuppressWarnings("unchecked")     private void siftUpComparable(int k, E x) {         Comparable<? super E> key = (Comparable<? super E>) x;         while (k > 0) {             int parent = (k - 1) >>> 1;             Object e = queue[parent];             if (key.compareTo((E) e) >= 0) {                 break;             }             queue[k] = e;             k = parent;         }         queue[k] = key;     }      @SuppressWarnings("unchecked")     private void siftUpUsingComparator(int k, E x) {         while (k > 0) {             int parent = (k - 1) >>> 1;             Object e = queue[parent];             if (comparator.compare(x, (E) e) >= 0) {                 break;             }             queue[k] = e;             k = parent;         }         queue[k] = x;     }     ... }

(5)无存储结构的阻塞队列SynchronousQueue

SynchronousQueue的内部没有容器来存储数据，因此当生产者往其添加一个元素而没有消费者去获取元素时，生产者会阻塞。当消费者往其获取一个元素而没有生产者去添加元素时，消费者也会阻塞。

SynchronousQueue的本质是借助了无容量存储的特点，来实现生产者线程和消费者线程的即时通信，所以它特别适合在两个线程之间及时传递数据。

线程池是基于阻塞队列来实现生产者/消费者模型的。当向线程池提交任务时，首先会把任务放入阻塞队列中，然后线程池中会有对应的工作线程专门处理阻塞队列中的任务。

Executors.newCachedThreadPool()就是基于SynchronousQueue来实现的，它会返回一个可以缓存的线程池。如果这个线程池大小超过处理当前任务所需的数量，会灵活回收空闲线程。当任务数量增加时，这个线程池会不断创建新的工作线程来处理这些任务。

public class Executors {     ...     //Creates a thread pool that creates new threads as needed,      //but will reuse previously constructed threads when they are available.     //These pools will typically improve the performance of programs that execute many short-lived asynchronous tasks.     //Calls to execute will reuse previously constructed threads if available.      //If no existing thread is available, a new thread will be created and added to the pool.      //Threads that have not been used for sixty seconds are terminated and removed from the cache.      //Thus, a pool that remains idle for long enough will not consume any resources.      //Note that pools with similar properties but different details (for example, timeout parameters)     //may be created using ThreadPoolExecutor constructors.     public static ExecutorService newCachedThreadPool() {         return new ThreadPoolExecutor(0, Integer.MAX_VALUE,             60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());     }     ... }

(6)阻塞队列结合体LinkedTransferQueue

LinkedTransferQueue是一个基于链表实现的无界阻塞TransferQueue。

阻塞队列的特性是根据队列的数据情况来阻塞生产者线程或消费者线程，TransferQueue的特性是生产者线程生产数据后必须等消费者消费才返回。

LinkedTransferQueue是TransferQueue和LinkedBlockingQueue的结合体，而SynchronousQueue内部其实也是基于TransferQueue来实现的，所以LinkedTransferQueue是带有阻塞队列功能的SynchronousQueue。

(7)双向阻塞队列LinkedBlockingDeque

LinkedBlockingDeque是一个基于链表实现的双向阻塞队列，双向队列的两端都可以插入和移除元素，可减少多线程并发下的一半竞争。