最初在看ConcurrentHashMap时，主要了解了:
1.对槽位的entry通过cas添加或者置null
2.对槽位上的列表或者红黑树通过synchronized(f) 对entry加锁，实现粒度非常细的锁定减少并发冲突。
当时对于多线程扩容过程感觉有点复杂，没有进一步研究。最近听同事讲了redis的渐进式rehash过程，感觉挺有意思，就想着趁着假期回头把ConcurrentHashMap的多线程扩容过程学习了一下，此篇作为记录和分享。
过程中参考了（引用）这篇文章，其重点讲述了trashfer本身的细节逻辑，对整个过程中的代码注解较少，本文主要补齐其他部分。
ConcurrentHashMap#transfer() 扩容逐行分析
1.关键控制变量
// 0 表示table尚未初始化
// >0 表示的为下一个扩容点 即容量*0.75
// <0 表示rs + 2开始 表示的参与扩容的线程数量 全部线程退出扩容并且所有槽位的占位都清除后才判定扩容结束
private transient volatile int sizeCtl;
2.插入元素 putVal
这个函数是扩容+写入过程操作的关键位置 其核心是在while(true)过程中 根据sizeCtl 和待插入槽位进行判断，
1.可能直接即可插入。
2.当前线程可能加入扩容。
3.可能在正在扩容的nextTable进行插入
特别注意tab = this.helpTransfer(tab, f); 结合调用过程全面的分析。
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key != null && value != null) {
int hash = spread(key.hashCode());
int binCount = 0;
ConcurrentHashMap.Node[] tab = this.table;
// 扩容每次是2的倍数 那么槽位上的item 要么在原来槽位上 要么在 i + N 上 N 表示当前len
// 如果当前未开始扩容 一个线程 插入完数据后，计数时发现需要扩容 该线程开始进行扩容
// 此时 其他线程执行插入时，检测到要插入的槽位已经被标记为扩容状态 会尝试加入扩容线程 或者在临时的nextTable上插入
// 此时 其他线程执行插入时，检测到要插入的槽位未标记为扩容状态 加锁并在当前tab上直接插入 如果扩容线程刚好处理到这个槽位 只能lock等
// 线程插入期间可能不加入扩容 在后面的增加计数时 还可能判断是否加入扩容
// cas 一般 在while() 中结合cas结果和其他状态的值 退出或者自旋
// 在分析过程中注意结合break return 和各种条件
while(true) {
int n;
while(tab == null || (n = tab.length) == 0) {
tab = this.initTable();
}
ConcurrentHashMap.Node f;
int i;
//当前槽位是null 直接cas写入 失败的话继续while判断
if ((f = tabAt(tab, i = n - 1 & hash)) == null) {
if (casTabAt(tab, i, (ConcurrentHashMap.Node)null, new ConcurrentHashMap.Node(hash, key, value))) {
break;
}
} else {
int fh;
// hash 值结果为-1的话说明当前待插入位置已经被其他扩容线程占位 不可用
// 这种情况进入helpTransfer 根据情况加入扩容 或者 直接拿到新的tab 继续尝试操作
if ((fh = f.hash) == -1) {
// 参考对helpTransfer的解析 一定要清楚返回值的tab可能已经改了
// 再次循环时 f 的值也重新计算了
tab = this.helpTransfer(tab, f);
} else {
// 没有处于扩容状态 或者扩容线程还未扫描到该位置 加锁更新列表或者红黑树
synchronized(f) {
// 这里类似double check 在加锁后 确保tab的i位置仍然为f
if (tabAt(tab, i) == f) {
// may java 11 new todo
if (fh < 0) {
if (f instanceof ConcurrentHashMap.TreeBin) {
binCount = 2;
ConcurrentHashMap.TreeNode p;
if ((p = ((ConcurrentHashMap.TreeBin)f).putTreeVal(hash, key, value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent) {
p.val = value;
}
}
} else if (f instanceof ConcurrentHashMap.ReservationNode) {
throw new IllegalStateException("Recursive update");
}
} else {
label124: {
binCount = 1;
ConcurrentHashMap.Node e;
Object ek;
for(e = f; e.hash != hash || (ek = e.key) != key && (ek == null || !key.equals(ek)); ++binCount) {
ConcurrentHashMap.Node<K, V> pred = e;
if ((e = e.next) == null) {
pred.next = new ConcurrentHashMap.Node(hash, key, value);
break label124;
}
}
oldVal = e.val;
if (!onlyIfAbsent) {
e.val = value;
}
}
}
}
}
if (binCount != 0) {
if (binCount >= 8) {
this.treeifyBin(tab, i);
}
if (oldVal != null) {
return oldVal;
}
break;
}
}
}
}
// 计算是否需要扩容的代码
this.addCount(1L, binCount);
return null;
} else {
throw new NullPointerException();
}
}
3.addCount
插入数据后，都会修改元素数量，并判断是否开启扩容
private final void addCount(long x, int check) {
int n;
ConcurrentHashMap.Node[] tab;
if (check >= 0) {
for(; s >= (long)(sc = this.sizeCtl) && (tab = this.table) != null && (n = tab.length) < 1073741824; s = this.sumCount()) {
int rs = resizeStamp(n);
if (sc < 0) {
//在扩容中，尝试加入扩容过程
ConcurrentHashMap.Node[] nt;
if (sc >>> 16 != rs || sc == rs + 1 || sc == rs + '\uffff' || (nt = this.nextTable) == null || this.transferIndex <= 0) {
加入失败 不需要新的线程帮忙扩容了
break;
}
if (U.compareAndSetInt(this, SIZECTL, sc, sc + 1)) {
//线程数+1 后帮忙扩容
this.transfer(tab, nt);
}
} else if (U.compareAndSetInt(this, SIZECTL, sc, (rs << 16) + 2)) {
//当前线程首次开启扩容，将sizeCtl的值设置为(rs << 16) + 2
this.transfer(tab, (ConcurrentHashMap.Node[])null);
}
}
}
}
- helpTransfer
线程调用helpTransfer 尝试帮忙扩容 这里会增加一系列判断比如扩容已经结束 直接返回当前table 或者执行扩容 当前线程无法加入扩容或者本线程扩容完成（整体未完成）直接返回正在扩容过程中的nextTab 执行插入操作
final ConcurrentHashMap.Node<K, V>[] helpTransfer(ConcurrentHashMap.Node<K, V>[] tab, ConcurrentHashMap.Node<K, V> f) {
//
ConcurrentHashMap.Node[] nextTab;
if (tab != null && f instanceof ConcurrentHashMap.ForwardingNode && (nextTab = ((ConcurrentHashMap.ForwardingNode)f).nextTable) != null) {
int rs = resizeStamp(tab.length);
int sc;
//尝试增加同时扩容线程数之前 判断 1. 当前的table,nextTable是否还在扩容装填
while(nextTab == this.nextTable && this.table == tab && (sc = this.sizeCtl) < 0 && sc >>> 16 == rs && sc != rs + 1 && sc != rs + '\uffff' && this.transferIndex > 0) {
// sizeCtl是否在扩容状态
// 是否有更多区间给未加入线程使用
// rs sizeCtl 初始值和这个有关联
if (U.compareAndSetInt(this, SIZECTL, sc, sc + 1)) {
//如果当前扩容线程数量cas加1成功则开始协助扩容
this.transfer(tab, nextTab);
break;
}
}
// 当while退出后 至少当前线程完成或者没有可分配空间
//如果此时退回到putVal的while循环中 尝试直接在nextTab上赋值或者操作
return nextTab;
} else {
// tab == null 说明扩容结束 table == nexttable
//f.nexttable == null 同样效果
return this.table;
}
}
5.transfer
这部分来自文章开头的链接 过程比较复杂 请转至原文阅读
/**
* Moves and/or copies the nodes in each bin to new table. See
* above for explanation.
*
* transferIndex 表示转移时的下标，初始为扩容前的 length。
*
* 我们假设长度是 32
*/
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
// 将 length / 8 然后除以 CPU核心数。如果得到的结果小于 16，那么就使用 16。
// 这里的目的是让每个 CPU 处理的桶一样多，避免出现转移任务不均匀的现象，如果桶较少的话，默认一个 CPU（一个线程）处理 16 个桶
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range 细分范围 stridea：TODO
// 新的 table 尚未初始化
if (nextTab == null) { // initiating
try {
// 扩容 2 倍
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
// 更新
nextTab = nt;
} catch (Throwable ex) { // try to cope with OOME
// 扩容失败， sizeCtl 使用 int 最大值。
sizeCtl = Integer.MAX_VALUE;
return;// 结束
}
// 更新成员变量
nextTable = nextTab;
// 更新转移下标，就是 老的 tab 的 length
transferIndex = n;
}
// 新 tab 的 length
int nextn = nextTab.length;
// 创建一个 fwd 节点，用于占位。当别的线程发现这个槽位中是 fwd 类型的节点，则跳过这个节点。
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
// 首次推进为 true，如果等于 true，说明需要再次推进一个下标（i--），反之，如果是 false，那么就不能推进下标，需要将当前的下标处理完毕才能继续推进
boolean advance = true;
// 完成状态，如果是 true，就结束此方法。
boolean finishing = false; // to ensure sweep before committing nextTab
// 死循环,i 表示下标，bound 表示当前线程可以处理的当前桶区间最小下标
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
// 如果当前线程可以向后推进；这个循环就是控制 i 递减。同时，每个线程都会进入这里取得自己需要转移的桶的区间
while (advance) {
int nextIndex, nextBound;
// 对 i 减一，判断是否大于等于 bound （正常情况下，如果大于 bound 不成立，说明该线程上次领取的任务已经完成了。那么，需要在下面继续领取任务）
// 如果对 i 减一大于等于 bound（还需要继续做任务），或者完成了，修改推进状态为 false，不能推进了。任务成功后修改推进状态为 true。
// 通常，第一次进入循环，i-- 这个判断会无法通过，从而走下面的 nextIndex 赋值操作（获取最新的转移下标）。其余情况都是：如果可以推进，将 i 减一，然后修改成不可推进。如果 i 对应的桶处理成功了，改成可以推进。
if (--i >= bound || finishing)
advance = false;// 这里设置 false，是为了防止在没有成功处理一个桶的情况下却进行了推进
// 这里的目的是：1. 当一个线程进入时，会选取最新的转移下标。2. 当一个线程处理完自己的区间时，如果还有剩余区间的没有别的线程处理。再次获取区间。
else if ((nextIndex = transferIndex) <= 0) {
// 如果小于等于0，说明没有区间了 ，i 改成 -1，推进状态变成 false，不再推进，表示，扩容结束了，当前线程可以退出了
// 这个 -1 会在下面的 if 块里判断，从而进入完成状态判断
i = -1;
advance = false;// 这里设置 false，是为了防止在没有成功处理一个桶的情况下却进行了推进
}// CAS 修改 transferIndex，即 length - 区间值，留下剩余的区间值供后面的线程使用
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
bound = nextBound;// 这个值就是当前线程可以处理的最小当前区间最小下标
i = nextIndex - 1; // 初次对i 赋值，这个就是当前线程可以处理的当前区间的最大下标
advance = false; // 这里设置 false，是为了防止在没有成功处理一个桶的情况下却进行了推进，这样对导致漏掉某个桶。下面的 if (tabAt(tab, i) == f) 判断会出现这样的情况。
}
}// 如果 i 小于0 （不在 tab 下标内，按照上面的判断，领取最后一段区间的线程扩容结束）
// 如果 i >= tab.length(不知道为什么这么判断)
// 如果 i + tab.length >= nextTable.length （不知道为什么这么判断）
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
if (finishing) { // 如果完成了扩容
nextTable = null;// 删除成员变量
table = nextTab;// 更新 table
sizeCtl = (n << 1) - (n >>> 1); // 更新阈值
return;// 结束方法。
}// 如果没完成
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {// 尝试将 sc -1. 表示这个线程结束帮助扩容了，将 sc 的低 16 位减一。
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)// 如果 sc - 2 不等于标识符左移 16 位。如果他们相等了，说明没有线程在帮助他们扩容了。也就是说，扩容结束了。
return;// 不相等，说明没结束，当前线程结束方法。
finishing = advance = true;// 如果相等，扩容结束了，更新 finising 变量
i = n; // 再次循环检查一下整张表
}
}
else if ((f = tabAt(tab, i)) == null) // 获取老 tab i 下标位置的变量，如果是 null，就使用 fwd 占位。
advance = casTabAt(tab, i, null, fwd);// 如果成功写入 fwd 占位，再次推进一个下标
else if ((fh = f.hash) == MOVED)// 如果不是 null 且 hash 值是 MOVED。
advance = true; // already processed // 说明别的线程已经处理过了，再次推进一个下标
else {// 到这里，说明这个位置有实际值了，且不是占位符。对这个节点上锁。为什么上锁，防止 putVal 的时候向链表插入数据
synchronized (f) {
// 判断 i 下标处的桶节点是否和 f 相同
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;// low, height 高位桶，低位桶
// 如果 f 的 hash 值大于 0 。TreeBin 的 hash 是 -2
if (fh >= 0) {
// 对老长度进行与运算（第一个操作数的的第n位于第二个操作数的第n位如果都是1，那么结果的第n为也为1，否则为0）
// 由于 Map 的长度都是 2 的次方（000001000 这类的数字），那么取于 length 只有 2 种结果，一种是 0，一种是1
// 如果是结果是0 ，Doug Lea 将其放在低位，反之放在高位，目的是将链表重新 hash，放到对应的位置上，让新的取于算法能够击中他。
int runBit = fh & n;
Node<K,V> lastRun = f; // 尾节点，且和头节点的 hash 值取于不相等
// 遍历这个桶
for (Node<K,V> p = f.next; p != null; p = p.next) {
// 取于桶中每个节点的 hash 值
int b = p.hash & n;
// 如果节点的 hash 值和首节点的 hash 值取于结果不同
if (b != runBit) {
runBit = b; // 更新 runBit，用于下面判断 lastRun 该赋值给 ln 还是 hn。
lastRun = p; // 这个 lastRun 保证后面的节点与自己的取于值相同，避免后面没有必要的循环
}
}
if (runBit == 0) {// 如果最后更新的 runBit 是 0 ，设置低位节点
ln = lastRun;
hn = null;
}
else {
hn = lastRun; // 如果最后更新的 runBit 是 1， 设置高位节点
ln = null;
}// 再次循环，生成两个链表，lastRun 作为停止条件，这样就是避免无谓的循环（lastRun 后面都是相同的取于结果）
for (Node<K,V> p = f; p != lastRun; p = p.next) {
int ph = p.hash; K pk = p.key; V pv = p.val;
// 如果与运算结果是 0，那么就还在低位
if ((ph & n) == 0) // 如果是0 ，那么创建低位节点
ln = new Node<K,V>(ph, pk, pv, ln);
else // 1 则创建高位
hn = new Node<K,V>(ph, pk, pv, hn);
}
// 其实这里类似 hashMap
// 设置低位链表放在新链表的 i
setTabAt(nextTab, i, ln);
// 设置高位链表，在原有长度上加 n
setTabAt(nextTab, i + n, hn);
// 将旧的链表设置成占位符
setTabAt(tab, i, fwd);
// 继续向后推进
advance = true;
}// 如果是红黑树
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;
}
}
}
}
}
}
