Java集合-并发
@TOC
Java集合，包括List、set、map等，在并发环境下，遍历的过程中是不容许更新操作（增删改），如下反例：
1 | |
执行上述程序，会报出异常：
1 | |
上述移除操作，Idea工具给的建议语句为，将：
1 | |
改为：
1 | |
并发集合(ConcurrentCollection)
非阻塞式集合（Non-Blocking Collection）
这类集合也包括添加和移除数据的方法。如果方法不能立即被执行，则返回null或抛出异常，但是调用这个方法的线程不会被阻塞。
ConcurrentLinkedQeque
ConcurrentLinkedDeque
ConcurrentLinkedDeque 是双向链表结构的无界并发队列。从JDK 7开始加入到J.U.C的行列中。使用CAS实现并发安全，与 ConcurrentLinkedQueue 的区别是该阻塞队列同时支持FIFO和FILO两种操作方式，即可以从队列的头和尾同时操作(插入/删除)。适合“多生产，多消费”的场景。内存一致性遵循对 ConcurrentLinkedDeque 的插入操作先行发生于(happen-before)访问或移除操作。相较于 ConcurrentLinkedQueue，ConcurrentLinkedDeque 由于是双端队列，所以在操作和概念上会更加复杂。
注意：size方法不是一个准确的操作
方法
- add(E e)：在此deque的尾部插入指定的元素，返回值为Boolean。
- addFirst(E e)：在此deque前面插入指定的元素。
- addLast(E e)：在此deque的末尾插入指定的元素。
- clear()：从这个deque中删除所有的元素。
- contains(Object o)：返回 true如果这个deque包含至少一个元素 e ，返回值为Boolean。
- descendingIterator()：以相反的顺序返回此deque中的元素的迭代器，返回值为Iterator。
- element()：检索但不删除由此deque表示的队列的头部（换句话说，该deque的第一个元素）。
- getFirst()：检索，但不删除，这个deque的第一个元素。
- getLast()：检索，但不删除，这个deque的最后一个元素。
- isEmpty()：如果此集合不包含元素，则返回 true 。
- iterator()：以正确的顺序返回此deque中的元素的迭代器，返回值为Iterator 。
- offer(E e)：在此deque的尾部插入指定的元素，返回值为boolean。
- offerFirst(E e)：在此deque前面插入指定的元素，返回值为boolean。
- offerLast(E e)：在此deque的末尾插入指定的元素，返回值为boolean。
- peek()：检索但不删除由此deque表示的队列的头（换句话说，该deque的第一个元素），如果此deque为空，则返回 null 。
- peekFirst()：检索但不删除此deque的第一个元素，如果此deque为空，则返回 null 。
- peekLast()：检索但不删除此deque的最后一个元素，如果此deque为空，则返回 null 。
- poll()：检索并删除由此deque表示的队列的头部（换句话说，该deque的第一个元素），如果此deque为空，则返回 null 。
- pollFirst()：检索并删除此deque的第一个元素，如果此deque为空，则返回 null 。
- pollLast()：检索并删除此deque的最后一个元素，如果此deque为空，则返回 null 。
- pop()：从这个deque表示的堆栈中弹出一个元素。
- push(E e)：将元素推送到由此deque代表的堆栈（换句话说，在该deque的头部），如果可以立即执行，而不违反容量限制，则抛出 IllegalStateException如果当前没有可用空间）。
- remove()：检索并删除由此deque表示的队列的头（换句话说，该deque的第一个元素）。
- remove(Object o)：删除第一个元素 e ，使 o.equals(e) ，如果这样一个元素存在于这个deque，返回值为boolean。
- removeFirst()：检索并删除此deque的第一个元素。
- removeFirstOccurrence(Object o)：删除第一个元素 e ，使 o.equals(e) ，如果这样一个元素存在于这个deque，返回值为boolean。
- removeLast()：检索并删除此deque的最后一个元素。
- size()：返回此deque中的元素数。
实例
- 添加大量的数据到一个列表中；
- 从同一个列表中移除大量的数据。执行上述程序，输出：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43package com.leehao.collection;
import java.util.concurrent.ConcurrentLinkedDeque;
/**
* add and poll
*/
public class CollectionDemo01 {
public static void main(String[] args) throws InterruptedException {
ConcurrentLinkedDeque<String> list = new ConcurrentLinkedDeque<>();
//添加数据
//100个线程同时添加数据
Thread[] addThreads = new Thread[100];
for (int i = 0; i < 100; i++) {
addThreads[i] = new Thread(() -> {
for (int j = 0; j < 10000; j++) {
list.add(Thread.currentThread().getName() + ":Element " + j);
}
});
addThreads[i].start();
//有序执行
addThreads[i].join();
}
System.out.println("after add,the size is: " + list.size());
//移除数据
Thread[] pollThreads = new Thread[100];
for (int i = 0; i < 100; i++) {
pollThreads[i] = new Thread(() -> {
for (int j = 0; j < 5000; j++) {
//一次性移除头尾
list.pollFirst();
list.pollLast();
}
});
pollThreads[i].start();
//有序执行
pollThreads[i].join();
}
System.out.println("after poll,the size is: " + list.size());
}
}1
2after add,the size is: 1000000
after poll,the size is: 0join()方法：Thread类中的join方法的主要作用就是同步，它可以使得线程之间的并行执行变为串行执行
总结
- 基于链接节点的无界并发deque 。 并发插入，删除和访问操作可以跨多个线程安全执行。 一个 ConcurrentLinkedDeque是许多线程将共享对公共集合的访问的适当选择。 像大多数其他并发集合实现一样，此类不允许使用null元素， ConcurrentLinkedDeque是一个双向链表 。
- ConcurrentLinkedDeque使用了自旋+CAS的非阻塞算法来保证线程并发访问时的数据一致性。由于队列本身是一种双链表结构，所以虽然算法看起来很简单，但其实需要考虑各种并发的情况，实现复杂度较高，并且ConcurrentLinkedDeque不具备实时的数据一致性，实际运用中，如果需要一种线程安全的栈结构，可以使用ConcurrentLinkedDeque。
- 关于ConcurrentLinkedDeque还有以下需要注意的几点：
- ConcurrentLinkedDeque的迭代器是弱一致性的，这在并发容器中是比较普遍的现象，主要是指在一个线程在遍历队列结点而另一个线程尝试对某个队列结点进行修改的话不会抛出ConcurrentModificationException，这也就造成在遍历某个尚未被修改的结点时，在next方法返回时可以看到该结点的修改，但在遍历后再对该结点修改时就看不到这种变化。
- size方法需要遍历链表，所以在并发情况下，其结果不一定是准确的，只能供参考。
阻塞式集合（Blocking Collection）
阻塞式集合（Blocking Collection）：这类集合包括添加和移除数据的方法。当集合已满或为空时，被调用的添加或者移除方法就不能立即被执行，那么调用这个方法的线程将被阻塞，一直到该方法可以被成功执行。
LinkedBlockingDeque是双向链表实现的双向并发阻塞队列。该阻塞队列同时支持FIFO和FILO两种操作方式，即可以从队列的头和尾同时操作(插入/删除)；并且，该阻塞队列是支持线程安全。还有，LinkedBlockingDeque还是可选容量的(防止过度膨胀)，即可以指定队列的容量。如果不指定，默认容量大小等于Integer.MAX_VALUE（@Native public static final int MAX_VALUE = 0x7fffffff）。
LinkedBlockingDeque
方法
参加该类的官方文档。
实战
1 | |
执行上述程序，输出结果：
1 | |
Map
HsahMap
因为多线程环境下，使用Hashmap进行put操作可能会引起死循环，导致CPU利用率接近100%，所以在并发情况下不能使用HashMap。例如如下代码：
1 | |
HashTable
HashTable容器使用synchronized来保证线程安全，但在线程竞争激烈的情况下Hashtable的效率非常低下。因为当一个线程访问HashTable的同步方法时，其他线程访问HashTable的同步方法时，可能会进入阻塞或轮询状态。如线程1使用put进行添加元素，线程2不但不能使用put方法添加元素，并且也不能使用get方法来获取元素，所以竞争越激烈效率越低。
ConcurrentHashMap
分段锁
HashTable容器在竞争激烈的并发环境下表现出效率低下的原因，是因为所有访问HashTable的线程都必须竞争同一把锁，那假如容器里有多把锁，每一把锁用于锁容器其中一部分数据，那么当多线程访问容器里不同数据段的数据时，线程间就不会存在锁竞争，从而可以有效的提高并发访问效率，这就是ConcurrentHashMap所使用的锁分段技术，首先将数据分成一段一段的存储，然后给每一段数据配一把锁，当一个线程占用锁访问其中一个段数据的时候，其他段的数据也能被其他线程访问。有些方法需要跨段，比如size()和containsValue()，它们可能需要锁定整个表而而不仅仅是某个段，这需要按顺序锁定所有段，操作完毕后，又按顺序释放所有段的锁。这里“按顺序”是很重要的，否则极有可能出现死锁，在ConcurrentHashMap内部，段数组是final的，并且其成员变量实际上也是final的，但是，仅仅是将数组声明为final的并不保证数组成员也是final的，这需要实现上的保证。这可以确保不会出现死锁，因为获得锁的顺序是固定的。
ConcurrentHashMap是由Segment数组结构和HashEntry数组结构组成。Segment是一种可重入锁ReentrantLock，在ConcurrentHashMap里扮演锁的角色，HashEntry则用于存储键值对数据。一个ConcurrentHashMap里包含一个Segment数组，Segment的结构和HashMap类似，是一种数组和链表结构， 一个Segment里包含一个HashEntry数组，每个HashEntry是一个链表结构的元素， 每个Segment守护者一个HashEntry数组里的元素,当对HashEntry数组的数据进行修改时，必须首先获得它对应的Segment锁。
JDK1.8的实现已经抛弃了Segment分段锁机制，利用CAS+Synchronized来保证并发更新的安全。数据结构采用：数组+链表+红黑树。
源码
1 | |
tableSizeFor(int c)的原理：将c最高位以下通过|=运算全部变成1，最后返回的时候，返回n+1；
eg:当输入为25的时候，n等于24，转成二进制为1100，右移1位为0110，将1100与0110进行或(“|”)操作，得到1110。接下来右移两位得11，再进行或操作得1111，接下来操作n的值就不会变化了。最后返回的时候，返回n+1，也就是10000，十进制为32。按照这种逻辑得到2的n次幂的数。
那么为什么要先-1再+1呢？输入若是为0，那么不论怎么操作，n还是0，但是HashMap的容量只有大于0时才有意义。
table初始化
table初始化操作会延缓到第一次put行为。但是put是可以并发执行的，那么是如何实现table只初始化一次的？接着上源码：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode());
int binCount = 0;
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh; K fk; V fv;
if (tab == null || (n = tab.length) == 0)//判断table还未初始化
tab = initTable();//初始化table
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value)))
break; // no lock when adding to empty bin
}
...省略一部分源码
}
}
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
//如果一个线程发现sizeCtl<0，意味着另外的线程执行CAS操作成功，当前线程只需要让出cpu时间片，
//由于sizeCtl是volatile的，保证了顺序性和可见性
if ((sc = sizeCtl) < 0)//sc保存了sizeCtl的值
Thread.yield(); // lost initialization race; just spin
else if (U.compareAndSetInt(this, SIZECTL, sc, -1)) {//cas操作判断并置为-1
try {
if ((tab = table) == null || tab.length == 0) {
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;//DEFAULT_CAPACITY = 16，若没有参数则大小默认为16
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}put操作
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode());//哈希算法
int binCount = 0;
for (Node<K,V>[] tab = table;;) {//无限循环，确保插入成功
Node<K,V> f; int n, i, fh; K fk; V fv;
if (tab == null || (n = tab.length) == 0)//表为空或表长度为0
tab = initTable();//初始化表
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {//i = (n - 1) & hash为索引值，查找该元素，
//如果为null,说明第一次插入
if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value)))
break; // no lock when adding to empty bin
}
else if ((fh = f.hash) == MOVED)//MOVED=-1;当前正在扩容，一起进行扩容操作
tab = helpTransfer(tab, f);
else if (onlyIfAbsent && fh == hash && // check first node
((fk = f.key) == key || fk != null && key.equals(fk)) &&
(fv = f.val) != null)
return fv;
else {
V oldVal = null;
synchronized (f) {//其他情况加锁同步
if (tabAt(tab, i) == f) {
if (fh >= 0) {
binCount = 1;
for (Node<K,V> e = f;; ++binCount) {
K ek;
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
oldVal = e.val;
if (!onlyIfAbsent)
e.val = value;
break;
}
Node<K,V> pred = e;
if ((e = e.next) == null) {
pred.next = new Node<K,V>(hash, key, value);
break;
}
}
}
else if (f instanceof TreeBin) {
Node<K,V> p;
binCount = 2;
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
else if (f instanceof ReservationNode)
throw new IllegalStateException("Recursive update");
}
}
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);
return null;
}
//哈希算法
static final int spread(int h) {
return (h ^ (h >>> 16)) & HASH_BITS;
}
//保证拿到最新的数据
static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {
return (Node<K,V>)U.getObjectAcquire(tab, ((long)i << ASHIFT) + ABASE);
}
//CAS操作插入节点，比较数组下标为i的节点是否为c，若是，用v交换，否则不操作。
//如果CAS成功，表示插入成功，结束循环进行addCount(1L, binCount)看是否需要扩容
static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,
Node<K,V> c, Node<K,V> v) {
return U.compareAndSetObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
}table扩容
当table容量不足的时候，即table的元素数量达到容量阈值sizeCtl，需要对table进行扩容。 整个扩容分为两部分：- 构建一个nextTable，大小为table的两倍。
- 把table的数据复制到nextTable中。
这两个过程在单线程下实现很简单，但是ConcurrentHashMap是支持并发插入的，扩容操作自然也会有并发的出现，这种情况下，第二步可以支持节点的并发复制，这样性能自然提升不少，但实现的复杂度也上升了一个台阶。
继续上源码：
第一步，构建nextTable，毫无疑问，这个过程只能只有单个线程进行nextTable的初始化.
1 | |
节点从table移动到nextTable，大体思想是遍历、复制的过程。遍历过所有的节点以后就完成了复制工作，把table指向nextTable，并更新sizeCtl为新数组大小的0.75倍 ，扩容完成。
4. get操作
判断table是否为空，如果为空，直接返回null。
计算key的hash值，并获取指定table中指定位置的Node节点，通过遍历链表或则树结构找到对应的节点，返回value值。
1 | |
- 和HashTable的区别
ConcurrentHashMap 是一个并发散列映射表，它允许完全并发的读取，并且支持给定数量的并发更新。
而HashTable和同步包装器包装的 HashMap，使用一个全局的锁来同步不同线程间的并发访问，同一时间点，只能有一个线程持有锁，也就是说在同一时间点，只能有一个线程能访问容器，这虽然保证多线程间的安全并发访问，但同时也导致对容器的访问变成串行化的了。
总结
Hashtable的任何操作都会把整个表锁住，是阻塞的。好处是总能获取最实时的更新，比如说线程A调用putAll写入大量数据，期间线程B调用get，线程B就会被阻塞，直到线程A完成putAll，因此线程B肯定能获取到线程A写入的完整数据。坏处是所有调用都要排队，效率较低。
ConcurrentHashMap 是设计为非阻塞的。在更新时会局部锁住某部分数据，但不会把整个表都锁住。同步读取操作则是完全非阻塞的。好处是在保证合理的同步前提下，效率很高。坏处是严格来说读取操作不能保证反映最近的更新。例如线程A调用putAll写入大量数据，期间线程B调用get，则只能get到目前为止已经顺利插入的部分数据。
应该根据具体的应用场景选择合适的HashMap。
Java集合-并发
https://leehoward.cn/2020/05/11/Java集合-并发/