ConcurrentHashMap源码解读三
今天首先讲解helpTransfer方法
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) { Node<K,V>[] nextTab; int sc;
//如果table不是空，且node节点是转移类型，数据校验，且node节点得nextTable（新table）不是空，同样也是数据校验，那么就尝试帮助扩容。if (tab != null && (f instanceof ForwardingNode) && (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) { int rs = resizeStamp(tab.length);//根据length得到一个标识符号
//如果nextTab没有被并发修改，且tab也没有被并发修改，且sizeCtl<0(说明还在扩容) while (nextTab == nextTable && table == tab && (sc = sizeCtl) < 0) {
//如果sizeCtl无符号右移16位不等于rs，sc前16位如果不等于标识符，则标识符变化挂了
//或者sizeCtl == rs+1，扩容结束了，不再有线程进行扩容。
//或者sizeCtl达到最大帮助线程得数量，或者转移下标正在调整，都代表扩容结束。break；返回table。 if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || sc == rs + MAX_RESIZERS || transferIndex <= 0) break;
//否则，进行扩容 if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) { transfer(tab, nextTab); break; } } return nextTab; } return table; }
接下来就是transfer部分
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) { int n = tab.length, stride;//stride为每个cpu所需要处理得桶个数。 if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)//如果是多cpu，那么每个线程划分任务，最小任务量是16个桶位的迁移 stride = MIN_TRANSFER_STRIDE; // subdivide range if (nextTab == null) {//如果新的table没有初始化，那么就初始化一个大小为原来二倍的数组为新数组，被nextTab引用 // initiating try { @SuppressWarnings("unchecked") Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];// nextTab = nt; } catch (Throwable ex) { // try to cope with OOME sizeCtl = Integer.MAX_VALUE;//如果扩容出现错误，则sizeCtl赋值为int最大值返回return; } nextTable = nextTab; transferIndex = n;//记录线程开始迁移的桶位，从后往前迁移。 } int nextn = nextTab.length;//记录新数组的末尾。
//当旧数组的某个桶位null或者这个桶的元素已经被全部转移到新数组中，那么就在旧数组中放一个这个
//当其他线程想往旧数组中put元素的时候，如果put元素的index处存的是这个，那么就调用helpTransfer让这个线程一起进行扩容 ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);//已经迁移的桶位，会用这个节点占位（这个节点的hash值为-1--MOVED） boolean advance = true;//数组一层层推进的标识符，advance为true就说明这个桶处理完了，有三种情况可以让advance为true。一是旧数组对应位置为null并通过cas写入一个fwd，二是旧数组对应位置就是一个fwd。三是就数组对应的位置已经转移成功 boolean finishing = false; // to ensure sweep before committing nextTab//扩容结束的标识符 for (int i = 0, bound = 0;;) { Node<K,V> f; int fh;
//while的作用就是确定每个线程需要处理的桶的下标，并且通过i--来遍历hash表中的每个桶，
//假如旧数组有32个桶模板transferferIndex就是32，当第一个线程进来的时候，nextIndex=transferIndex = 32，nextBound = nextIndex-stride = 16，而transferIndex也通过cas被调整为16，所以第一个线程处理桶的范围是从16到31号桶。
//所以第二个线程进来，nextIndex = transferIndex = 16，nextBound = 0
//所以第二个线程处理桶的范围是从0到第15号桶。 while (advance) { int nextIndex, nextBound;
//过了第一次之后i-->=bound,在这里就跳出循环了，这里的finishing是为了二次循环检查用的，让最后一个线程遍历整个数组，但是如果多线程的时候i--可能会使i跳出这个线程对应的范围，所以用finishing保证i可以一直--。 if (--i >= bound || finishing) advance = false;
//当所有的线程都分配完对应需要转移的桶的范围后，transferindex就为0，所以当某一个线程完成了自己的任务后，nextIndex=transferIndex=0，所以i被设置为-1，跳出这个whlle。 else if ((nextIndex = transferIndex) <= 0) { i = -1; advance = false; }
//第一次进入这个扩容方法时，是到这里来的，因为前面for循环i=0，所以前面的两个判断都跳过，直接来到这里。这里就是在分配当前线程所需要转移的范围。 else if (U.compareAndSwapInt (this, TRANSFERINDEX, nextIndex, nextBound = (nextIndex > stride ? nextIndex - stride : 0))) { bound = nextBound;//线程负责桶区间当前最小下标 i = nextIndex - 1;//线程负责桶区间当前最大下标 advance = false; } }
//如果没有更多的需要迁移的桶位，就进入该if if (i < 0 || i >= n || i + n >= nextn) { int sc; if (finishing) {//扩容结束后，保存新数组，并重新计算扩容阈值，赋值给sizeCtl nextTable = null;//删除成员变量 table = nextTab;//更新table sizeCtl = (n << 1) - (n >>> 1);//更新阈值，就是将阈值变为旧数组的1.5倍，因为旧阈值是旧数组的0.75倍，旧数组扩容2倍对应的阈值也就是扩容两倍，即1.5倍。 return; }
//扩容任务线程数减1 if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) { if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)//判断当前所有扩容任务线程是否都执行完成 return; finishing = advance = true;//所有扩容线程都执行完，标识结束 i = n; // recheck before commit//再次循环检查一个表 } }
//当前迁移的桶位没有元素，那就不用迁移了，直接在该位置通过cas添加一个fwd节点 else if ((f = tabAt(tab, i)) == null) advance = casTabAt(tab, i, null, fwd); else if ((fh = f.hash) == MOVED)//当前节点已经被迁移 advance = true; // already processed else { synchronized (f) { //当前节点需要迁移，加锁迁移，保证多线程安全//加锁防止在这个桶位迁移的时候，别的线程对这个桶位进行元素添加。 if (tabAt(tab, i) == f) {// 判断 i 下标处的桶节点是否和 f 相同Node<K,V> ln, hn;//ln低位桶，hn高位桶 if (fh >= 0) {// 如果 f 的 hash 值大于 0 。TreeBin 的 hash 是 -2
// 对老长度进行与运算（第一个操作数的的第n位与上第二个操作数的第n位如果都是1，那么结果的第n位也为1，否则为0） // 由于 Map 的长度都是 2 的次方（000001000 这类的数字），那么取于 length 只有 2 种结果，一种是 0，一种是1 // 如果是结果是0 ，Doug Lea 将其放在低位，反之放在高位，目的是将链表重新 hash，放到对应的位置上，让新的取于算法能够击中他。int runBit = fh & n;//就比如n是32，那么就是000010000，就只有一个地方为1.所以fn是大于0的数，他的二进制与上n的话，结果只有000010000和00000000； Node<K,V> lastRun = f;// 尾节点，且和头节点的 hash 值取于不相等for (Node<K,V> p = f.next; p != null; p = p.next) {//开始从这个桶的头节点之后的节点开始遍历元素 int b = p.hash & n;
//其实这里这个for循环，就是为了求出lastRun最后指向的是是链表中的哪个节点。那么在下面的另一个for循环中，lastRun后面的就跟这个lastRun节点是一样的，就可以不用遍历，直接把lastRun这个头节点接过去就行。省了一些时间 if (b != runBit) { runBit = b; lastRun = p; } } if (runBit == 0) {//如果最后更新的runBit是0，设置低位节点 ln = lastRun; hn = null; } else { hn = lastRun;// 反之，设置高位节点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; if ((ph & n) == 0)
//注意创建node节点的最后一个参数ln指代的是next，也就是说，我们不再是从头到尾节点，而是从尾节点开始往头节点走。当开始遍历到第一个元素的时候，那么就把刚刚得到的ln设置尾新元素的next。 ln = new Node<K,V>(ph, pk, pv, ln);// 如果是0 ，那么创建低位节点else hn = new Node<K,V>(ph, pk, pv, hn);// 1 则创建高位} setTabAt(nextTab, i, ln);//ln挂到新数组的原下标 setTabAt(nextTab, i + n, hn);//hn挂到新数组的原下标+老数组长度，跟HashMap一样。 setTabAt(tab, i, fwd);//把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); if ((h & n) == 0) { if ((p.prev = loTail) == null) lo = p; else loTail.next = p; loTail = p; ++lc; } else { if ((p.prev = hiTail) == null) hi = p; else hiTail.next = p; hiTail = p; ++hc; } } ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) ://如果长度小于等于6，则将红黑树转换成链表 (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; } } } } } }
接下来借用这个博客的图https://www.jianshu.com/p/2829fe36a8dd来更好理解上述的链表转移
可以看到四号桶和10号桶的元素转移时都会把链表拆成两份，规则就是上述根据hash值取与旧表的长度，如果结果是0，放在低位，否则放在高位。假如只看10号桶，那么黑丝的会放在新表的10号位置，白色节点会放在10+16也就是26的位置。
从这个for循环中可以发现。如果节点是上面图的一样，那么最后9和10是黑色，因此lastRun就是9。可以从1开始for循环验证。
for (Node<K,V> p = f.next; p != null; p = p.next) { int b = p.hash & n; if (b != runBit) { runBit = b; lastRun = p; } }
经过第二个for循环时
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) ln = new Node<K,V>(ph, pk, pv, ln); else hn = new Node<K,V>(ph, pk, pv, hn); }
这里可以看到，还未到lastRun的节点都是倒序的。比如黑色的7-4-2-1。原图数字标错，将4误标为了3。白色的就是8-6-5-3。当循环到p为9时，也就是lastRun时，那么9包括后面的节点，也就是10，直接插入到1之后。所以第一个for循环就是省了遍历9，10这个节点的时间。处理之后的数组结构就是这样的
总结：转移旧数组是从右至左开始转移的。数组扩容是按两倍进行扩容的。阈值也就相应的变为原来的两倍。链表的转移是有一个小优化来节省时间的。就是得到lastRun节点。然后是反向的。转移后的桶位，低位链就是原来的位置，高位链就是原来的位置加旧数组长度的位置。
版权声明：本文为lingchenstudy原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。