2 并发容器线程平安应答之道
引言
在后面，咱们学习了 hashmap
大家都晓得 HashMap 不是线程平安（put、删除、批改、递增、扩容都无锁）的
所以在解决并发的时候会呈现问题
接下来咱们看下 J.U.C 包外面提供的一个线程平安并且高效 Map（ConcurrentHashMap）看一下，他到底是如何实现线程并发平安的2.1 并发容器总体概述
指标：学习 ConcurrentHashMap 基本概念和意识它的数据结构
ConcurrentHashMap 概念：
ConcurrentHashMap 是 J.U.C 包外面提供的一个 线程平安 的 HashMap，在并发编程中应用的频率（Spring）比拟高。
数据结构如下
数组 + 链表 + 红黑树 + 锁（synchronized+cas）
总结：
1、数据结构和 hashmap 截然不同，惟一的区别就是 concurrenthashmap 在 put、删除、批改、递增、扩容和数据迁徙的时候都加锁了（syn or cas）
2、加锁只是锁住一个元素，区别于 HashTable(整个表，idea 能够查看源码来验证)
2.2 并发容器数据结构与继承
指标：
简略意识下 ConcurrentHashMap 继承关系
总结
ConcurrentHashMap：实现 Serializable 示意反对序列化
继承 AbstractMap（实现 map 接口），实现了一些基本操作
实现 ConcurrentMap 接口，封装了 map 的基本操作
2.3 并发容器源码深度分析
测试代码
见 put 局部2.3.1 并发容器成员变量
指标：意识下 ConcurrentHashMap 成员变量，先有个印象，不便后续源码剖析
private static final int MAXIMUM_CAPACITY = 1 << 30; //table 最大容量：2^30=1073741824
private static final int DEFAULT_CAPACITY = 16; // 默认容量，必须是 2 的幂数
static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8; //// 数组的倡议最大值
private static final int DEFAULT_CONCURRENCY_LEVEL = 16; // 并发级别，1.8 前的版本分段锁遗留下来的，为兼容以前的版本
static final int TREEIFY_THRESHOLD = 8;// 链表转红黑树阀值
static final int UNTREEIFY_THRESHOLD = 6;// 树转链表阀值
static final int MIN_TREEIFY_CAPACITY = 64;// 转化为红黑树的表的最小容量
private static final int MIN_TRANSFER_STRIDE = 16;// 每次进行转移的最小值
// 咦？threshold 呢？？？2.3.2 并发容器结构器
指标：
先意识下 ConcurrentHashMap 的 5 个结构器，看下在结构中（第一步）做了哪些事件
1、ConcurrentHashMap()型构造函数
public ConcurrentHashMap() {}
总结：该构造函数用于创立一个带有默认初始容量 (16)、负载因子 (0.75) 的空映射
2、ConcurrentHashMap(int)型构造函数
private static final int MAXIMUM_CAPACITY = 1 << 30
public ConcurrentHashMap(int initialCapacity) {if (initialCapacity < 0) // 初始容量小于 0，抛出异样
throw new IllegalArgumentException();
// 达到最大容量的一半以上后，间接取最大容量！int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
MAXIMUM_CAPACITY :
tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
// 初始化，sizeCtl 是什么鬼？？看上去是容量……
this.sizeCtl = cap;
}总结：该构造函数用于创立一个带有指定初始容量的 map
3、ConcurrentHashMap(Map<? extends K, ? extends V>)型构造函数
public ConcurrentHashMap(Map<? extends K, ? extends V> m) {
this.sizeCtl = DEFAULT_CAPACITY;
// 将汇合 m 的元素全副放入
putAll(m);
}总结：该构造函数用于结构一个与给定映射具备雷同映射关系的新映射。
4、ConcurrentHashMap(int, float)型构造函数
public ConcurrentHashMap(int initialCapacity, float loadFactor) {this(initialCapacity, loadFactor, 1);
}总结：该构造函数用于创立一个带有指定初始容量、加载因子 新的空映射。
5、ConcurrentHashMap(int, float, int)型构造函数
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0) // 合法性判断
throw new IllegalArgumentException();
if (initialCapacity < concurrencyLevel) // Use at least as many bins
initialCapacity = concurrencyLevel; // as estimated threads
long size = (long)(1.0 + (long)initialCapacity / loadFactor);
int cap = (size >= (long)MAXIMUM_CAPACITY) ?
MAXIMUM_CAPACITY : tableSizeFor((int)size);
this.sizeCtl = cap; // 如同是容量？没那么简略，待会往下看
}总结：该构造函数用于创立一个带有指定初始容量、加载因子和并发级别的新的空映射
扩大：和 HashMap 齐全一样？错！咱们来看一个实例
1）代码实例
package com.cmap;
import org.openjdk.jol.info.ClassLayout;
import java.util.HashMap;
import java.util.concurrent.ConcurrentHashMap;
public class CMapInit {public static void main(String[] args) {HashMap m = new HashMap(15,0.5f);
ConcurrentHashMap cm = new ConcurrentHashMap(15, 0.5f);
//debug here
System.out.println("before put");
m.put(1,1);
cm.put(1,1);
//and here
System.out.println("after put");
System.out.println(ClassLayout.parseInstance(cm).toPrintable());
}
}
2）调试，put 之前
3）持续，debug 到第二步试试，put 之后
- 容量并不是咱们之前认为的 16，而是 32
- 而 sizeCtl，咱们了解，应该类比于 hashMap 中的 threshold，它应该等于 32*0.5=16 才对
- 可是最终为 24
这是什么神操作？？？
4）原理分析
先说论断：办法调用的都是 tableSizeFor，只不过，Cmap 所计算的参数不一样，留神回顾下面的构造函数
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
if (initialCapacity < concurrencyLevel) // Use at least as many bins
initialCapacity = concurrencyLevel; // as estimated threads
//initial = 15, size = 31
long size = (long)(1.0 + (long)initialCapacity / loadFactor);
// 所以 tableSizeFor 做满 1 运算前，并不是 15 自身，而是 size，也就是 31
// 运算后，cap=32，不是 16
int cap = (size >= (long)MAXIMUM_CAPACITY) ?
MAXIMUM_CAPACITY : tableSizeFor((int)size);
this.sizeCtl = cap;
}那么它啥时候变成 24 的呢？
// 开始之初，table 为 null，在 put 时，会触发 table 的初始化，也就是以下办法
// 从 put 办法的入口能够追踪到，咱们猜测它必定在这里，初始化 table 的时候
private final Node<K,V>[] initTable() {Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
//sc = 原来的 sizeCtl 也就是 32
if ((sc = sizeCtl) < 0)
Thread.yield(); // lost initialization race; just spin
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {if ((tab = table) == null || tab.length == 0) {
//n = sc = 32 , 默认就是 default=16 了
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
// 创立 node 数组，长度为 n，也就是 32
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
// 创立完复制给 table，初始化实现，也就是咱们看到的 32 长度的数组
table = tab = nt;
// n >>> 2，相当于 n 除以 4 是 8，32-8=24
// 实际效果相当于，n* 3/4，也就是 n*0.75，你指定的 0.5 在初始化时对它没什么用！sc = n - (n >>> 2);
}
} finally {
// 在 finally 中将它赋给了 sizeCtl，也就是咱们最终看到的 24
sizeCtl = sc;
}
break;
}
}
return tab;
}
那么 sizeCtl 起不到 threshold 的作用，它是干嘛的呢？
其实它的作用远远比 hashmap 中的 thredhold 大的多，看看官网的说法：
/**
* Table initialization and resizing control. When negative, the
* table is being initialized or resized: -1 for initialization,
* else -(1 + the number of active resizing threads). Otherwise,
* when table is null, holds the initial table size to use upon
* creation, or 0 for default. After initialization, holds the
* next element count value upon which to resize the table.
*/
private transient volatile int sizeCtl;
翻译过去就是这样子：（官网就这么规定的，记住它！）
- 用来管制 table 的初始化和扩容操作
- 默认为 0，int 类型的，废话
- -1 代表 table 正在初始化
- -N 示意有 N - 1 个线程正在进行扩容操作
其余状况：
- 如果 table 未初始化，示意 table 须要初始化的大小。
- 如果 table 初始化实现，示意 table 的容量，默认是 table 大小的 0.75 倍
而批改它的办法也比拟多，initTable 只是其中的一个：
- initTable()
- addCount（）
- tryPresize（）
- transfer（）
- helpTransfer（）
2.3.3 put 办法
指标：1、ConcurrentHashMap 减少的逻辑是什么
2、ConcurrentHashMap 是如何保障线程平安的
根底回顾：对于 compareAndSwapInt（CAS）
肯定要了解 CAS 的原理，Cmap 的精华就在于 cas 和 sync 保障了线程平安，下文的源码剖析马上要用到它
（画图展现两个线程的 cas 交互操作）
(U.compareAndSwapInt(this, SIZECTL, sc, -1))解释：
- 此办法是 Java 的 native 办法，并不禁 Java 语言实现。
- 办法的作用是，读取传入对象 this 在内存中偏移量为 SIZECTL 地位的值与期望值 sc 作比拟。
- 相等就把 - 1 值赋值给 SIZECTL 地位的值。办法返回 true。
- 不相等，就勾销赋值，办法返回 false。
- 个别配合循环重试操作，被 for 或 while 所包裹
1）测试代码
package com.cmap;
import java.util.ArrayList;
import java.util.ConcurrentModificationException;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
public class CMapTest {
static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash
public static void main(String[] args) {ConcurrentHashMap<Integer, Integer> m = new ConcurrentHashMap<Integer, Integer>();
for (int i = 0; i < 64; i++) {if (i == 0) {m.put(i, i);// 失常新增（演示）} else if (i == 11) {
// 容量默认 16，临界值 =12，那么 i =11 正好是第 12 个值，引发扩容
m.put(i, i);// 扩容（演示）} else if (i == 10) {m.put(27, 27);
m.put(43, 43);
} else if (i == 9) {} else if(i==23){m.put(i,i); // 23, 第二次扩容
}else {m.put(i, i);// 失常新增
}
}
m.get(8);
System.out.println(m);
}
// 哈希抵触
static void testHashCode() {System.out.println((16 - 1) & spread(new Integer(27).hashCode()));
System.out.println((16 - 1) & spread(new Integer(43).hashCode()));
System.out.println((16 - 1) & spread(new Integer(11).hashCode()));
}
static final int spread(int h) {return (h ^ (h >>> 16)) & HASH_BITS;
}
}
2）减少过程
// 提醒：该办法岔路比拟多，要广度优先浏览，先看外围大路，再细分外面的子办法
final V putVal(K key, V value, boolean onlyIfAbsent) {if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode());//key 取 hash 扰动
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)
tab = initTable();// 表为空的话，初始化表，上面会具体介绍【预留 1】// 寻址，找到头结点 f
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//cas 在这里！！！// 插槽为空，cas 插入元素
// 比拟是否为 null，如果 null 才会设置并 break，否则到 else
if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value)))
break; // 插入胜利，break 终止即可，如果不胜利，会进入下一轮 for
}
//helpTransfer() 扩容。下大节具体讲，一个个来……【预留 2】else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
else {
V oldVal = null;
//synchronized 在这里！！！// 插槽不为空，阐明被别的线程 put 抢占了槽
// 那就加锁，锁的是以后插槽上的头节点 f（相似分段锁）synchronized (f) {if (tabAt(tab, i) == f) { // 这步的目标是再次确认，链表头元素没有被其余线程动过
if (fh >= 0) { // 失常节点的 hash 值
binCount = 1; // 统计节点个数
// 沿着以后插槽的 Node 链往后找
for (Node<K,V> e = f;; ++binCount) {
K ek;
// 如果找到雷同 key，阐明之前 put 过
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
oldVal = e.val;
if (!onlyIfAbsent) //abset 参数来决定要不要笼罩，默认是笼罩
e.val = value;
break;
}
Node<K,V> pred = e;
// 否则，新 key，新 Node 插入到最初
if ((e = e.next) == null) {pred.next = new Node<K,V>(hash, key, value);
break;
}
}
}
// 如果是红黑树，阐明曾经转化过，按树的规定放入 Node
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); // 计数，如果超了，调 transfer 扩容
return null;
}
//compareAndSetObject，比拟并插入，典型 CAS 操作
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);
}
3）初始化表办法
多线程下 initTable 的交互流程：
源码：
/**
* 留神点：先以单线程看业务流程，再类比多个线程操作下的并发是如何解决的？*/
private final Node<K,V>[] initTable() {Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) { // 自旋
// 第 1 个线程这个 if 不成立，会进入上面，设置为 -1
// 第 2 个线程来的时候 if 成立，留神了解多线程在跑。if ((sc = sizeCtl) < 0) // 留神回顾下面的值，小于 0 示意正在初始化，或扩容
Thread.yield();// 有线程在操作，将以后线程 yield 让出工夫片。唤醒后进入下一轮 while
//CAS 操作来设置 SIZECTL 为 -1，如果设置胜利，示意以后线程取得初始化的资格
// 传入对象 & 内存地址 & 期望值 & 将批改的值
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {if ((tab = table) == null || tab.length == 0) {
// 再次确认一下，table 是 null，还没初始化
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;// 默认容量 16
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n]; // 初始化 table
// 给 table 赋值，留神这个 table 是 volatile 的，会被其余线程及时看到！// 一旦其余线程看到不是 null，走 while 循环发现 table 不等于空就 return 了
table = tab = nt;
sc = n - (n >>> 2); // 计算下次扩容的阈值，容量的 0.75
}
} finally {sizeCtl = sc;}
break;
}
}
return tab;
}总结：
- 判断程序为先看 table=null 再看 sizeCtl = -1
- T1 来得早，循序渐进进行
- T2 – T4 在不同工夫点进入，口头不一样，有的是被 cas 挡住，有的被 table 非 null 挡住
2.3.4 扩容
指标：1、图解 + 断点剖析查看 ConcurrentHashMap 是如何扩容的
2、图解 + 断点剖析查看 ConcurrentHashMap 是如何迁徙数据的
测试代码
package com.cmap;
import java.util.ArrayList;
import java.util.ConcurrentModificationException;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
public class CMapTest {
static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash
public static void main(String[] args) {ConcurrentHashMap<Integer, Integer> m = new ConcurrentHashMap<Integer, Integer>();
for (int i = 0; i < 64; i++) {if (i == 0) {m.put(i, i);// 失常新增（演示）} else if (i == 11) {m.put(i, i);// 扩容 1
} else if (i == 10) {m.put(27, 27);
m.put(43, 43);
} else if (i == 9) {} else if(i==23){m.put(i,i); // 23, 第二次扩容（演示点，debug 打在这里再进去）}else {m.put(i, i);// 失常新增
}
}
System.out.println(m);
}
}
入口：
/*
在下面，putVal 办法的最初，进 addCount()，再跳到最初，发现：会走到 transfer() 办法，这是真正的扩容操作
同时，Cmap 还带有它的特色，也就是 多线程帮助扩容，helpTransfer
最初调的也是 transfer 办法
*/
final V putVal(K key, V value, boolean onlyIfAbsent) {
// ……
addCount(1L, binCount);
}
private final void addCount(long x, int check) {
// ...
// 扩容操作的外围在这里
transfer(tab, null);
}
/**
* Helps transfer if a resize is in progress. 如果正在扩容，下来帮忙
* tab = 旧数组，f= 头结点，如果正在扩容，它是一个 ForwardNode 类型
*/
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {Node<K,V>[] nextTab; int sc;
if (tab != null && (f instanceof ForwardingNode) &&
(nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {int rs = resizeStamp(tab.length);
while (nextTab == nextTable && table == tab &&
(sc = sizeCtl) < 0) {// 一堆条件判断，不去管它
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || transferIndex <= 0)
break;
// 其余线程进来，多了这一步：cas 将 sizeCtl + 1,（示意减少了一个线程帮忙其扩容）if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
// 找到了，外围在这里！这个外部藏着扩容的具体操作
transfer(tab, nextTab);
break;
}
}
return nextTab;
}
return table;
}外围源码【重点】
CMap 是如何多线程帮助迁徙数据的？？？
/**
* Moves and/or copies the nodes in each bin to new table. See
* above for explanation.
*/
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
// 将 length / 8 而后除以 CPU 外围数。如果失去的后果小于 16，那么就应用 16。// 如果桶较少的话，默认一个 CPU（一个线程）解决 16 个桶
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // 最小 16
if (nextTab == null) { // 新的 table 尚未初始化
try {
// 扩容 2 倍
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
// 赋值给新 table
nextTab = nt;
} catch (Throwable ex) {
sizeCtl = Integer.MAX_VALUE;
return;
}
// 更新成员变量
nextTable = nextTab;
// transferIndex 示意没迁徙的桶里最大索引的值，这个会被多个线程瓜分走越来越小。// 一开始这个值是旧 tab 的尾部：也就是 n
// 瓜分时，从大索引往后分，也就是程序是：15 14 13 12 ....0
transferIndex = n; // tag_0
}
// 新 tab 的 length
int nextn = nextTab.length;
// 创立一个 fwd 节点，用于标记。// 留神，它外面的 hash 属性是固定的 MOVED，还记得 putVal 里的 helpTransfer 前的判断吗？// 当别的线程 put 的时候，正好发现这个槽位中是 fwd 类型的节点，也调 helperTransfer 参加进来。ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
boolean advance = true; // 长期变量，示意要不要挪动槽
boolean finishing = false; // 长期变量，示意以后槽有没有迁徙完
for (int i = 0, bound = 0;;) { // 每次 for 遍历一个桶来迁徙，也就是旧 table 里的一个元素
Node<K,V> f; int fh;
while (advance) { // 这里的 while 是配合 tag_3 的 cas 做自旋，只有它可能会触发屡次循环，其余俩都是 1 次跳出
//while 比拟乱：能够打断点进来调试查看每次的通过
// 第一次 for 的时候进 tag_3 确定 bound 和 i，也就是给以后线程调配了 bound ~ i 之间的桶
// 当前每次 --i，只有不大于 bound，都进 tag_1，也就是啥都不干
// 最初一次，等于 bound 的时候，阐明调配给以后线程的桶被它 for 完了，退出
int nextIndex, nextBound;
if (--i >= bound || finishing) // tag_1
// 如果 i 比 bound 还大，或者以后 i 下的链表没挪动完，-- i 推动一格
advance = false;
else if ((nextIndex = transferIndex) <= 0) { // tag_2，留神！这个赋值操作第一次也要产生
// 如果 transferIndex <=0 阐明已迁徙实现，没有桶须要解决了，退出
i = -1;
advance = false;
}
else if (U.compareAndSwapInt // tag_3
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
// 第一次 for 的时候会走进这里，确定以后线程负责的桶的范畴，同时 cas 更新 transferIndex
// 也就是，多个线程第一次都会拜访到这里，通过 cas 来分一部分桶，cas 避免并发下反复调配
// 留神，来这里之前，通过了 tag_2 的赋值：// 所以这里在 cas 前 nextIndex = transferIndex = 16
// cas 后，transferIndex = nextBound = (nextIndex - stride) = 0
// 留神，这里不肯定是 0，只不过旧长度 16 被一个线程全拿走了，剩下了 0 个
// 也就是说，transfer 是本次调配后，还剩下的桶里最大的索引，别的线程还会持续分
bound = nextBound;// 最小下标 0（旧数组）i = nextIndex - 1;// 最大下标 15（旧数组）advance = false;
}
} // end while
// 判断 i 的范畴，不在可挪动插槽的索引范畴内，阐明全副迁徙完了！if (i < 0 || i >= n || i + n >= nextn) {
int sc;
// 如果实现了扩容
if (finishing) {
nextTable = null;// 开释
table = nextTab;// 更新 table
sizeCtl = (n << 1) - (n >>> 1); // 更新阈值
return;// 完结办法。}
// 如果没实现，尝试应用 cas 缩小 sizeCtl，也就是扩容的线程数，同时更新标记 finishing 为 true
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
return;
finishing = advance = true;//
i = n;
}
}
// 上面才是真正迁徙数据的操作！！！else if ((f = tabAt(tab, i)) == null)
// 获取老 tab i 下标地位的变量，如果是 null，就应用 fwd 占位。// cas 胜利，advance 为 true，下次 for 里 while 会做 -- i 挪动一个下标
advance = casTabAt(tab, i, null, fwd);
else if ((fh = f.hash) == MOVED)// 如果不是 null 且 hash 值是 MOVED。advance = true; // 阐明别的线程曾经解决过了，挪动一个下标
else {
// 到这里，阐明这个地位有理论值了，且不是占位符，那就须要咱们迁数据了。// 对这个节点上锁。避免别的线程 putVal 的时候向链表插入数据
synchronized (f) {
// 判断 i 下标处的桶节点是否和 f 雷同，确保没有被别的线程动过
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;// 定义 low, height 高位桶，低位桶
// 如果 f 的 hash 值大于 0 属于惯例 hash，开始拆分高下链表
// 参考动态变量：MOVED -1、TREEBIN -2、RESERVED -3、HASH_BITS > 0
if (fh >= 0) {
// 和老长度进行与运算，因为 Map 的长度都是 2 的次方（16 就是 10000 这类的数字）// 那么取与 n 只有 2 种后果，一种是 0，一种是 n
// 如果是后果是 0，拆分后，Doug Lea 将其放在低位链表，反之放在高位链表
// 这里和 HashMap 的算法一样！int runBit = fh & n; // 算算头结点是高位还是低位
Node<K,V> lastRun = f;
// 遍历这个桶，留神，这中央有个讨巧的操作！// 和 HashMap 不同这里不是一上来就挪动，而是先打标记
// 往下看 ↓（能够借助上面的图来同步阐明）//
for (Node<K,V> p = f.next; p != null; p = p.next) {
// 沿着链往下走，挨个取与
int b = p.hash & n;
// 如果和上次循环的值相等，那不动（当然第一次的话就是和头节点比拟）if (b != runBit) {
// 如果不相等的话，就切换值
runBit = b; // 0 遍。lastRun = p; // 这个 lastRun 保障前面的节点与本人的取于值雷同，防止前面没
}
}
// 思考一下，通过上一轮遍历完，产生了什么？// runBit 要么是 0 要么是 1，// lastRun 指向了最初一次切换的那个节点，它前面再没产生或切换
// 也就意味着，lastRun 前面所有的节点和它都具备雷同的 runBit 值
// 想想，能够做什么？？？// 对！在 lastRun 处间接切断！带着前面的尾巴，间接当做拆分后的高位，或者低位链表
// 这样就不须要和 hashMap 一样挨个断开指针，再挨个接一遍到新链，一锅端就行了
if (runBit == 0) {// 如果最初的 runBit 是 0，间接当低位链表
ln = lastRun;
hn = null;
}
else {
hn = lastRun; // 如果最初的 runBit 是 1，间接当高位链表
ln = null;
}
// 那么 lastRun 后面剩下的那些呢？// 再遍历一遍就是了，留神，是从头结点 f 遍历到 lastRun，前面的不须要操心了
for (Node<K,V> p = f; p != lastRun; p = p.next) {
int ph = p.hash; K pk = p.key; V pv = p.val;
if ((ph & n) == 0) // 如果与运算后果是 0，那么放低位链表，留神是头插
ln = new Node<K,V>(ph, pk, pv, ln); // 参数里的 ln 是 next，头插！else // 1 则放高位
hn = new Node<K,V>(ph, pk, pv, hn);
} // 为什么这里不怕多线程时的头插法出问题？（因为在 sync 里！）// 这里往下就相似 hashMap
// 设置低位链表放在新链表的 i
setTabAt(nextTab, i, ln);
// 设置高位链表，在原有长度上加 n
setTabAt(nextTab, i + n, hn);
// 将旧的链表设置成占位符，示意迁徙完了！setTabAt(tab, i, fwd);
// 持续向后推动
advance = true;
}
// 如果是红黑树同样的路子，设置高下位 node
else if (f instanceof TreeBin) {TreeBin<K,V> t = (TreeBin<K,V>)f;
TreeNode<K,V> lo = null, loTail = null;
TreeNode<K,V> hi = null, hiTail = null;
int lc = 0, hc = 0;
// 遍历
for (Node<K,V> e = t.first; e != null; e = e.next) {
int h = e.hash;
TreeNode<K,V> p = new TreeNode<K,V>
(h, e.key, e.val, null, null);
// 和链表雷同的判断，与运算 == 0 的放在低位
if ((h & n) == 0) {if ((p.prev = loTail) == null)
lo = p;
else
loTail.next = p;
loTail = p;
++lc;
} // 不是 0 的放在高位
else {if ((p.prev = hiTail) == null)
hi = p;
else
hiTail.next = p;
hiTail = p;
++hc;
}
}
// 如果树的节点数小于等于 6，那么转成链表，反之，创立一个新的树
ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
(hc != 0) ? new TreeBin<K,V>(lo) : t;
hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
(lc != 0) ? new TreeBin<K,V>(hi) : t;
// 低位树
setTabAt(nextTab, i, ln);
// 高位树
setTabAt(nextTab, i + n, hn);
// 旧的设置成占位符
setTabAt(tab, i, fwd);
// 持续向后推动
advance = true;
}
}
}
}
}
}总结
1、对于多线程协同
原来：128，扩容后 256
难道应用单线程去实现所有数据的迁徙工作？
既然应用多线程进行迁徙，如果保证数据不能乱？
将数组分段（桶），每个线程负责至多 16 个桶（stride），8 个线程就能够并行工作了
至于谁分哪些桶，从高索引到低索引，通过 cas 一起减 transferIndex 的值来实现，防止反复切分
切一段，低索引叫 bound，高索引叫 i，遍历迁徙就是了
2、对于数据迁徙（一个讨巧的小操作）
tips：
第一次，从 11 往后遍历，最初 runBit=0，lastRun 指向 31 节点
从 31 处切断，前面的一窝端间接当低位链表，不须要再挨个动他们
第二次，再遍历 11 – 30，依据状况头插到高位和低位新链表上
3、线程安全性
1、多个线程通过 cas 操作避免反复操作。
2、节点援用的中央应用 volatile 放弃了线程批改时对其余线程及时可见
3、迁徙的时候对插槽加 sync 锁，保障安全性
2.3.5 get 办法
指标：1、ConcurrentHashMap 查问是否加锁，如何保障线程平安
2、在查问的时候遇到扩容怎么办
ConcurrentHashMap 查问流程图如下
tips
多线程下，所谓 get 的不平安因素，就是最怕读到脏数据
get 的时候取到了数据，其实其余线程曾经把它改掉了，就是所谓的可见性问题。
get 办法源码如下
//get 操作无锁
// 因为 Node 的 val 和 next 是用 volatile 润饰的
// 多线程环境下线程 A 批改结点的 val 或者新增节点的时候是对线程 B 可见的
public V get(Object key) {Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
//key 取 hash
int h = spread(key.hashCode());
//1. 判断 table 是不是空的，2. 以后桶上是不是空的
// 如果为空，返回 null
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
// 找到对应 hash 槽的第一个 node，如果 key 相等，返回 value
if ((eh = e.hash) == h) {if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
else if (eh < 0)
//hash 值为负值示意正在扩容，这个时候查的是 ForwardingNode 的 find 办法来定位到 nextTable 新表中
return (p = e.find(h, key)) != null ? p.val : null;
while ((e = e.next) != null) { // 既不是首节点也不是 ForwardingNode，那就往下遍历
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
// 遍历完还没找到，返回 null
return null;
}
思考：
get 没有加锁，在进行查问的时候是如何保障读取不到脏数据呢？
猜测一下？
是在内部类 Node 类的 val 上加了 volatile？
2、是在成员变量数组 table 上加了 volatile？
论断：get 通过 Node 外部类 volatile 关键字来保障 可见性 、 有序性
总结
- 计算 hash 值，定位到该 table 索引地位，如果是首节点合乎就返回
- 如果遇到扩容的时候，会调用标记正在扩容节点 ForwardingNode 的 find 办法，查找该节点，匹配就返回
- 以上都不合乎的话，就往下遍历节点，匹配就返回，否则最初就返回 null
- get 不加锁，是因为 Node 的成员 val 和指针 next 是用 volatile 润饰的
- 在 1.8 中 ConcurrentHashMap 的 get 操作全程不须要加锁，这也是它比其余并发汇合比方 hashtable 平安效率高的起因之一
扩大：
remove 的操作与 put 一样。只是 put 是加到链表上，而 remove 是在链表上移除。
题外话
Cmap 里用到了大量的 CAS
CAS（Compare and Swap）, 比拟并替换，它是一个乐观锁
比拟的什么？替换的什么？
比拟当前工作内存的值和主内存的值，如雷同则批改，否则持续比拟；直到内存和工作内存中的值统一为止
解释
这是因为咱们执行第一个的时候，期望值（主存）和本来值是满足的，因而批改胜利，
第二次后，主内存的值曾经批改成了 B，不满足期望值，因而返回了 false，本次写入失败
cas 有什么毛病？如何解决
毛病一
毛病：最大毛病就是 ABA 问题
ABA: 如果一个值原来是 A，变成了 B，又变成了 A，那么应用 CAS 进行查看时会发现它的值没有发生变化，然而实际上却变动了
解决方案：1、应用版本号。在变量后面追加上版本号，每次变量更新的时候把版本号加一，那么 A-B-A 就会变成 1A-2B-3A
2、从 Java1.5 开始 JDK 的 atomic 包里提供了一个类 AtomicStampedReference 来解决 ABA 问题
这个类的 compareAndSet 办法作用是首先查看以后援用是否等于预期援用，并且以后标记是否等于预期标记，如果全副相等，则更新毛病二
不停自旋（循环）会给 CPU 带来更大的开销 本文由传智教育博学谷 – 狂野架构师教研团队公布，转载请注明出处！
如果本文对您有帮忙，欢送关注和点赞；如果您有任何倡议也可留言评论或私信，您的反对是我保持创作的能源