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

大纲

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 mutatorsfinal 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 mutatorsfinal 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，比原数组的大小少1Object[] 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 semanticssetArray(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)阻塞队列的应用

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

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

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