大纲
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)阻塞队列的应用
阻塞队列可以理解为线程级别的消息队列。
消息中间件可以理解为进程级别的消息队列。
所以可以通过阻塞队列来缓存线程的请求，从而达到流量削峰的目的。
![[大模型笔记]扣子-知识库搭建，并用Java-SDK调用的笔记](https://i-blog.csdnimg.cn/direct/b6346ac446594598ba36c0080f6909d8.png)
