ConcurrentHashMap的源码分析-transfer
作者：互联网
扩容是ConcurrentHashMap的精华之一，扩容操作的核心在于数据的转移，在单线程环境下数据的转移很简单，无非就是把旧数组中的数据迁移到新的数组。但是这在多线程环境下，在扩容的时候其他线程也可能正在添加元素，这时又触发了扩容怎么办？可能大家想到的第一个解决方案是加互斥锁，把转移过程锁住，虽然是可行的解决方案，但是会带来较大的性能开销。因为互斥锁会导致所有访问临界区的线程陷入到阻塞状态，持有锁的线程耗时越长，其他竞争线程就会一直被阻塞，导致吞吐量较低。而且还可能导致死锁。
而ConcurrentHashMap并没有直接加锁，而是采用CAS实现无锁的并发同步策略，最精华的部分是它可以利用多线程来进行协同扩容
简单来说，它把Node数组当作多个线程之间共享的任务队列，然后通过维护一个指针来划分每个线程锁负责的区间，每个线程通过区间逆向遍历来实现扩容，一个已经迁移完的bucket会被替换为一个ForwardingNode节点，标记当前bucket已经被其他线程迁移完了。接下来分析一下它的源码实现
1、fwd:这个类是个标识类，用于指向新表用的，其他线程遇到这个类会主动跳过这个类，因为这个类要么就是扩容迁移正在进行，要么就是已经完成扩容迁移，也就是这个类要保证线程安全，再进行操作。
2、advance:这个变量是用于提示代码是否进行推进处理，也就是当前桶处理完，处理下一个桶的标识
3、finishing:这个变量用于提示扩容是否结束用的
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
//将 (n>>>3相当于 n/8) 然后除以 CPU核心数。如果得到的结果小于 16，那么就使用 16
// 这里的目的是让每个 CPU 处理的桶一样多，避免出现转移任务不均匀的现象，如果桶较少的话，默认一个 CPU（一个线程）处理 16 个桶，也就是长度为16的时候，扩容的时候只会有一 个线程来扩容
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
//nextTab未初始化，nextTab是用来扩容的node数组
if (nextTab == null) { // initiating
try {
@SuppressWarnings("unchecked")
//新建一个n<<1原始table大小的nextTab,也就是32
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
nextTab = nt;//赋值给nextTab
} catch (Throwable ex) { // try to cope with OOME
sizeCtl = Integer.MAX_VALUE; //扩容失败，sizeCtl使用int的最大值
return;
}
nextTable = nextTab; //更新成员变量
transferIndex = n;//更新转移下标，表示转移时的下标
}
int nextn = nextTab.length;//新的tab的长度
// 创建一个 fwd 节点，表示一个正在被迁移的Node，并且它的hash值为-1(MOVED)，也就是前面我们在讲putval方法的时候，会有一个判断MOVED的逻辑。它的作用是用来占位，表示原数组中位置i处的节点完成迁移以后，就会在i位置设置一个fwd来告诉其他线程这个位置已经处理过了，具体后续还会在讲
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
// 首次推进为 true，如果等于 true，说明需要再次推进一个下标（i--），反之，如果是false，那么就不能推进下标，需要将当前的下标处理完毕才能继续推进
boolean advance = true;
//判断是否已经扩容完成，完成就return，退出循环
boolean finishing = false; // to ensure sweep before committing nextTab
通过for自循环处理每个槽位中的链表元素，默认advace为真，通过CAS设置transferIndex属性值，并初始化i和bound值，i指当前处理的槽位序号，bound指需要处理的槽位边界，先处理槽位15的节点；
for (int i = 0, bound = 0;;) {
// 这个循环使用CAS不断尝试为当前线程分配任务
// 直到分配成功或任务队列已经被全部分配完毕
// 如果当前线程已经被分配过bucket区域
// 那么会通过--i指向下一个待处理bucket然后退出该循环
Node<K,V> f; int fh;
while (advance) {
int nextIndex, nextBound;
//--i表示下一个待处理的bucket，如果它>=bound,表示当前线程已经分配过
bucket区域
if (--i >= bound || finishing)
advance = false;
else if ((nextIndex = transferIndex) <= 0) {//表示所有bucket已经
被分配完毕
i = -1;
advance = false;
}
//通过cas来修改TRANSFERINDEX,为当前线程分配任务，处理的节点区间为
(nextBound,nextIndex)->(0,15)
else if (U.compareAndSwapInt(this, TRANSFERINDEX, nextIndex,nextBound = (nextIndex > stride ? nextIndex - stride : 0))) {
bound = nextBound;//0
i = nextIndex - 1;//15
advance = false;
}
}
//i<0说明已经遍历完旧的数组，也就是当前线程已经处理完所有负责的bucket
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
if (finishing) {//如果完成了扩容
nextTable = null;//删除成员变量
table = nextTab;//更新table数组
sizeCtl = (n << 1) - (n >>> 1);//更新阈值(32*0.75=24)
return;
}
// sizeCtl 在迁移前会设置为 (rs << RESIZE_STAMP_SHIFT) + 2 (详细介绍点击这里)
// 然后，每增加一个线程参与迁移就会将 sizeCtl 加 1，
// 这里使用 CAS 操作对 sizeCtl 的低16位进行减 1，代表做完了属于自己的任务
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
第一个扩容的线程，执行transfer方法之前，会设置 sizeCtl = (resizeStamp(n) << RESIZE_STAMP_SHIFT) + 2)
后续帮其扩容的线程，执行transfer方法之前，会设置 sizeCtl = sizeCtl+1
每一个退出transfer的方法的线程，退出之前，会设置 sizeCtl = sizeCtl-1
那么最后一个线程退出时：必然有sc == (resizeStamp(n) << RESIZE_STAMP_SHIFT) + 2)，即 (sc - 2) == resizeStamp(n) << RESIZE_STAMP_SHIFT
// 如果 sc - 2 不等于标识符左移 16 位。如果他们相等了，说明没有线程在帮助他们扩容了。也就是说，扩容结束了。
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
return;
// 如果相等，扩容结束了，更新 finising 变量finishing = advance = true;
// 再次循环检查一下整张表i = n; // recheck before commit
}
}
// 如果位置 i 处是空的，没有任何节点，那么放入刚刚初始化的 ForwardingNode ”空节点“
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
//表示该位置已经完成了迁移，也就是如果线程A已经处理过这个节点，那么线程B处理这个节点
时，hash值一定为MOVED
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
}
}
标签：扩容,Node,ConcurrentHashMap,transfer,bound,源码,线程,nextIndex,nextTab 来源： https://blog.csdn.net/Leon_Jinhai_Sun/article/details/112062447