C/C++教程

ConcurrentHashMap源码解读三

本文主要是介绍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节点。然后是反向的。转移后的桶位,低位链就是原来的位置,高位链就是原来的位置加旧数组长度的位置。

这篇关于ConcurrentHashMap源码解读三的文章就介绍到这儿,希望我们推荐的文章对大家有所帮助,也希望大家多多支持为之网!