并发编程——ConcurrentHashMap#transfer() 扩容逐行分析

时间 2019-11-16

标签并发编程 concurrenthashmap#transfer concurrenthashmap transfer 扩容逐行分析栏目 Java 繁體版

原文原文链接

前言

ConcurrentHashMap 是并发中的重中之重，也是最经常使用的数据结果，以前的文章中，咱们介绍了 putVal 方法。并发编程之 ConcurrentHashMap（JDK 1.8） putVal 源码分析。其中分析了 initTable 方法和 putVal 方法，但也留下了一句话：java

这篇文章仅仅是 ConcurrentHashMap 的开头，关于 ConcurrentHashMap 里面的精华太多，值得咱们好好学习。node

说道精华，他的扩容方法绝对是精华，要知道，ConcurrentHashMap 扩容是高度并发的。算法

今天来逐行分析源码。编程

先说结论

首先说结论。源码加注释我会放在后面。该方法的执行逻辑以下：数组

经过计算 CPU 核心数和 Map 数组的长度获得每一个线程（CPU）要帮助处理多少个桶，而且这里每一个线程处理都是平均的。默认每一个线程处理 16 个桶。所以，若是长度是 16 的时候，扩容的时候只会有一个线程扩容。多线程
初始化临时变量 nextTable。将其在原有基础上扩容两倍。并发
死循环开始转移。多线程并发转移就是在这个死循环中，根据一个 finishing 变量来判断，该变量为 true 表示扩容结束，不然继续扩容。ide

3.1 进入一个 while 循环，分配数组中一个桶的区间给线程，默认是 16. 从大到小进行分配。当拿到分配值后，进行 i-- 递减。这个 i 就是数组下标。（其中有一个 bound 参数，这个参数指的是该线程这次能够处理的区间的最小下标，超过这个下标，就须要从新领取区间或者结束扩容，还有一个 advance 参数，该参数指的是是否继续递减转移下一个桶，若是为 true，表示能够继续向后推动，反之，说明尚未处理好当前桶，不能推动) 3.2 出 while 循环，进 if 判断，判断扩容是否结束，若是扩容结束，清空临死变量，更新 table 变量，更新库容阈值。若是没完成，但已经没法领取区间（没了），该线程退出该方法，并将 sizeCtl 减一，表示扩容的线程少一个了。若是减完这个数之后，sizeCtl 回归了初始状态，表示没有线程再扩容了，该方法全部的线程扩容结束了。（这里主要是判断扩容任务是否结束，若是结束了就让线程退出该方法，并更新相关变量）。而后检查全部的桶，防止遗漏。 3.3 若是没有完成任务，且 i 对应的槽位是空，尝试 CAS 插入占位符，让 putVal 方法的线程感知。 3.4 若是 i 对应的槽位不是空，且有了占位符，那么该线程跳过这个槽位，处理下一个槽位。 3.5 若是以上都是否是，说明这个槽位有一个实际的值。开始同步处理这个桶。 3.6 到这里，都尚未对桶内数据进行转移，只是计算了下标和处理区间，而后一些完成状态判断。同时，若是对应下标内没有数据或已经被占位了，就跳过了。源码分析
处理每一个桶的行为都是同步的。防止 putVal 的时候向链表插入数据。 4.1 若是这个桶是链表，那么就将这个链表根据 length 取于拆成两份，取于结果是 0 的放在新表的低位，取于结果是 1 放在新表的高位。 4.2 若是这个桶是红黑数，那么也拆成 2 份，方式和链表的方式同样，而后，判断拆分过的树的节点数量，若是数量小于等于 6，改形成链表。反之，继续使用红黑树结构。 4.3 到这里，就完成了一个桶从旧表转移到新表的过程。性能

好，以上，就是 transfer 方法的整体逻辑。仍是挺复杂的。再进行精简，分红 3 步骤：

计算每一个线程能够处理的桶区间。默认 16.
初始化临时变量 nextTable，扩容 2 倍。
死循环，计算下标。完成整体判断。
1 若是桶内有数据，同步转移数据。一般会像链表拆成 2 份。

大致就是上的的 3 个步骤。

再来看看源码和注释。

再看源码分析

源码加注释：

/** * 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;
                    }
                }
            }
        }
    }
}
复制代码

代码加注释比较长，有兴趣能够逐行对照，有 2 个判断楼主看不懂为何这么判断，知道的同窗能够提醒一下。

而后，说说精华的部分。

Cmap 支持并发扩容，实现方式是，将表拆分，让每一个线程处理本身的区间。以下图：

假设总长度是 64 ，每一个线程能够分到 16 个桶，各自处理，不会互相影响。

而每一个线程在处理本身桶中的数据的时候，是下图这样的：

扩容前的状态。

当对 4 号桶或者 10 号桶进行转移的时候，会将链表拆成两份，规则是根据节点的 hash 值取于 length，若是结果是 0，放在低位，不然放在高位。

所以，10 号桶的数据，黑色节点会放在新表的 10 号位置，白色节点会放在新桶的 26 号位置。

下图是循环处理桶中数据的逻辑：

处理完以后，新桶的数据是这样的：

总结

transfer 方法能够说很牛逼，很精华，内部多线程扩容性能很高，

经过给每一个线程分配桶区间，避免线程间的争用，经过为每一个桶节点加锁，避免 putVal 方法致使数据不一致。同时，在扩容的时候，也会将链表拆成两份，这点和 HashMap 的 resize 方法相似。

而若是有新的线程想 put 数据时，也会帮助其扩容。鬼斧神工，使人赞叹。