ConcurrentHashMap源码解析4.transfer()方法(核心)

本文主要是介绍ConcurrentHashMap源码解析4.transfer()方法(核心)，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！

1.扩容流程图

2.链表迁移示意图

3.transfer()方法

总结：

• 第一次进来扩容的线程会创建出一个新表。长度为原来的2倍。
• 迁移元素从后往前(索引从大到小)。
• 迁移完成的桶在当前桶位置放一个ForwardIngNode类型的节点，表示该桶迁移完成
• 迁移时通过hash&n(原长度)就是判断高位来判断在新链表中的索引位置，跟HashMap扩容原理一样。
• 低位链表(树)存储到新表的原索引位置
• 高位链表(树)存储到新表的原索引 + n(原数组长度)的位置
• 迁移元素时会使用synchronized锁定当前桶位，锁对象就是当前桶位的头结点，这是分段锁的思想。
• 迁移时，会根据原桶中的节点创建一个新的节点(除了lastRun机制的节点外)。
• 最后一个扩容的线程在退出时会重新扫描原表判断是否有遗漏的桶没有迁移节点，然后将nextTable赋值给table，然后将nextTable置为NULL，将sizeCtl设置为新数组长度的3/4即扩容阈值。

LastRun机制

LastRun机制就是为了省掉最后连续一段高位相同的节点的创建过程(只有原桶位是链表是才会有lastRun机制)

当最后一段的节点高位相同时，就不必创建了，直接引用这一串节点即可。

源码解析

   /*
    *  两种情况会进入transfer中 
    *  1. 触发扩容的线程 传来的nextTab属性是NULL。
    *  2. 协助扩容的线程 传来的nextTab非NULL。
    */
   private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
         
        /*
         *  n表示扩容之前table数组的长度
         *  stride表示分配给线程任务的步长 (一般为16)
         */
        int n = tab.length, stride;
        
        if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
            stride = MIN_TRANSFER_STRIDE; //16
       
        /*
         *  nextTab == null 说明当前是扩容的线程
         *  需要将nextTable创建出来。  
         */
        if (nextTab == null) {            // initiating
            try {
                
                //创建了一个是原来长度2倍的Node数组
                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;
                return;
            }
            //赋值给nextTable数组，方便协助扩容线程拿到新表。
            nextTable = nextTab;
            //记录迁移数据整体位置的一个标记。index计数是从1开始计算的，从高到底进行计数
            //transferIndex <=0 表示迁移完毕
            transferIndex = n;
        }
       
        //表示新数组的长度
        int nextn = nextTab.length;
       
        /*
         * fwd节点，当某个桶位数据处理完毕后，将此桶位设置为fwd节点，其他写线程或读线程看到		   
         * 后，会有不同的逻辑
         */
        ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
       
        //推进标记
        boolean advance = true;
        //完成标记
        boolean finishing = false; // to ensure sweep before committing nextTab
       
        /*  
         *  i表示分配给当前线程任务，执行到的桶位
         *  bound 表示分配给当前线程任务的下界限制
         *  (迁移从高索引为低索引位开始迁移)
         */
        int i = 0, bound = 0;
       
        //自旋
        for (;;) {
            /*
             * f 表示桶位的头结点
             * fh 头结点的hash值
             */
            Node<K,V> f; int fh;
            
            /*
             *  1.给当前线程分配任务区间
             *  2.维护当前线程任务进入(i表示当前处理的桶位)
             *  3.维护map对象全局范围内的进度
             */   
            while (advance) {
                /*   
                 * nextInde 分配任务的开始下标
                 * nextBound 分配任务的结束下标
                 */
                int nextIndex, nextBound;
                
                /*
                 * CASE1：
                 * 条件一： -- i >= bound
                 *
                 * 成立：表示当前线程的任务尚未完成，还有相应区间的的桶位要处理，--i就让当				
                 * 前线程处理下一个桶位
                 *
                 * 不成立：表示当前线程任务已完成，或者未分配  
                 */
                if (-- i >= bound || finishing)
                    advance = false;
                
                /*
                 *  CASE2
                 *  前置条件：当前线程任务已完成或者未分配
                 *  条件成立：表示对象全局范围内的桶位都分配完毕了，没有区间可分配了，
                 *  设置当前线程的i为-1，跳出循环后，执行退出迁移任务相关的程序。
                 *  
                 *  条件不成立：表示对象全局范围内的桶位尚未分配完毕，还有区间可以分配。
                 */
                else if ((nextIndex = transferIndex) <= 0) {
                    i = -1;
                    advance = false; //推进标记置为false
                }
                /*
                 * CASE3
                 * 前置条件1、当前线程需要分配任务区间 2、全局范围内还有桶位尚未迁移
                 * 条件成立： 说明当前线程分配任务成功， 
                 * 条件失败： 说明分配给的当前线程失败，和其他线程发生了竞争。
                 */
                else if (U.compareAndSwapInt
                         (this, TRANSFERINDEX, nextIndex,
                          nextBound = (nextIndex > stride ?
                                       nextIndex - stride : 0))) {
                    //分配任务成功 设置下界
                    bound = nextBound; 
                    //设置开始下标
                    i = nextIndex - 1; //-1是变为下标
                    advance = false; //分配任务后跳出
                }
            }
//---------------------------START----------------------------------------//                
            /*
             *  处理线程任务完成后，线程退出transfer方法的逻辑
             *  这里只考虑条件1 
             *  i < 0 表示当前线程未分配到任务
             */     
            if (i < 0 || i >= n || i + n >= nextn) {
                //保存sizeCtl的变量
                int sc;
                
                //只有最终完成整个迁移后，才会来到这个逻辑里
                if (finishing) {
                    //将nextTable置为NULL
                    nextTable = null;
                    //迁移后的新表赋值给table
                    table = nextTab;
                    //(n << 1) - (n >>> 1) = 0.75 * 2 * n
                    //下一次的扩容阈值就是新表的0.75
                    sizeCtl = (n << 1) - (n >>> 1);
                    //最终return。
                    return;
                }
                
               /*
                * 当前线程要退出，因为进入transfer()时将sizeCtl的值加了1，所以退出时
                * 必须将sizeCtl的值-1，(扩容时，sizeCtl前16位表示扩容戳，后16位表示
                * (1 + nThread)个线程数)
                * 这里使用CAS的方式尝试更新sizeCtl。
                */
               if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
                    //表示当前线程不是最后一个退出的线程。不需要干最后的活，直接return。
                    if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
                        return;
                    
                    //最后一个线程会一直检查是否有遗漏的桶位没有迁移，然后去做迁移。
                    finishing = advance = true;
                    i = n; // recheck before commit
                }
            }
//------------------------------END-------------------------------------------            
//下面的代码表示线程处理一个桶位数据的迁移工作，处理完毕后设置advance为true，表示继续推进，一直循环下去直接任务完毕。
            
            //来到下面这一堆代码(CASE2 - 4)的前置条件，当前线程任务尚未处理完，正在进行中
            
            /*
             * CASE2： 当前桶位未存放数据，只需要将此处设置为fwd节点即可
             */
            else if ((f = tabAt(tab, i)) == null)
                //CAS操作，期望为NULL，设置为fwd节点。
                advance = casTabAt(tab, i, null, fwd);
            
            /*
             * CASE3 
             * 哈希值为MOVED的节点是FWD节点，表示当前桶位的数据已经被迁移完毕了。
             */
            else if ((fh = f.hash) == MOVED)
                advance = true; // already processed
            
            /*
             * CASE4
             * 前置条件，当前桶位有数据，且头节点不是fwd节点，说明当前桶位的节点就是需要迁			   
             * 移的节点
             */
            else {
                //加锁 锁住当前桶位头结点(分段锁思想)
                synchronized (f) {
                    //再次判断。保证在加锁时当前桶位的头结点被其他线程修改了。
                    if (tabAt(tab, i) == f) {
                        /*
                         * ln 表示低位链表 
                         * hn 表示高位链表
                         */
                        Node<K,V> ln, hn;
                        
                        /*
                         * 条件成立：表示当前桶位是链表 
                         */
                        if (fh >= 0) {
                            
                           /*
                            * 这里是lastRun机制
                            * 可以获取当前链表末尾连续高位不变的一串node，为了方便迁移
                            */ 
                           int runBit = fh & n;
                           Node<K,V> lastRun = f;
                           //遍历链表，
                           for (Node<K,V> p = f.next; p != null; p = p.next) {
                                int b = p.hash & n;
                                //当前节点的高位与lastRun节点的哈希值不一致，就更新
                                //lastRun，最终lastRun指向一串高位相同的节点
                                //这一串节点一定是链表的最后一段。
                                if (b != runBit) {
                                    runBit = b;
                                    lastRun = p;
                                }
                            }
                            
                            //如果lastRun指向的一串节点的高位是0，将ln也引用这一串节点
                            if (runBit == 0) {
                                ln = lastRun;
                                hn = null;
                            }
                            //否则将hn(高位链表)引用这一串节点
                            else {
                                hn = lastRun;
                                ln = null;
                            }
                            
                          /*
                           * 类似HashMap将链表分为高位链表和低位链表，低位链表在扩容之
                           * 后的索引与原哈希表中的索引位置一样，高位链表的索引位置是
                           * 原索引位置+n
                           */
                          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);
                            }
                            //低位链设置到新哈希表的原索引位置(i)
                            setTabAt(nextTab, i, ln);
                            //高位链设置到新哈希表的i + n位置
                            setTabAt(nextTab, i + n, hn);
                            //设置原表的位置为fwd节点，表示当前桶位迁移完毕。
                            setTabAt(tab, i, fwd);
                            advance = true; 
                        }
                        
                        /*
                         *   当前节点是红黑树节点。
                         *   
                         */
                        else if (f instanceof TreeBin) {
                            //转换头结点为TreeBin节点。
                            TreeBin<K,V> t = (TreeBin<K,V>)f;
                            
                            //低位双向链表
                            TreeNode<K,V> lo = null, loTail = null;
                            //高位双向链表
                            TreeNode<K,V> hi = null, hiTail = null;
                            
                          
                          /*
                           * lc 表示低位链表中的元素
                           * hc 表示高位链表中的元素
                           */  
                          int lc = 0, hc = 0;
                            
                          for (Node<K,V> e = t.first; e != null; e = e.next) {
                                //表示循环处理当前元素的hash值
                                int h = e.hash;
                              
                                //使用当前节点构建出来的新的TreeNode节点
                                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;
                                }
                            }
                            
                          /*
                           * 判断低位链表的长度是否达到了树转链表的阈值(长度小于等于6)
                           * 如果达到了调用untreeify()方法将双向链表转为单向链表
                           * 没有达到，
                           *    判断高位链表是否有节点
                           *        有节点，将低位链表转为树.
                           *        没有节点，说明原树节点全都是低位链表，直接将原来的
                           *        树的头结点放到新表的头结点即可
                           */
                            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;
                            
                            /*
                             * 指定迁移 + 将原表中的桶位设置为fwd节点。
                             */
                            setTabAt(nextTab, i, ln);
                            setTabAt(nextTab, i + n, hn);
                            setTabAt(tab, i, fwd);
                            advance = true;
                        }
                    }
                }
            }
        }
    }

3.helpTransfer()

在put操作时，会判断当前桶位是否是fwd节点，如果是的话，就会进入**helpTransfer()**方法尝试帮助扩容。

  // -------------- put()方法中的片段--------------------------------------
  else if ((fh = f.hash) == MOVED)
                tab = helpTransfer(tab, f);
  //---------------------------------------
	
    /*
     * tab表示原哈希表
     * f 表示当前桶位的头节点。
     */
    final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
        
        /*
         * nextTab 就是引用的nextTable.
         * sc 保存的sizeCtl
         */
        Node<K,V>[] nextTab; int sc;
        
        
        //三个条件都是恒成立
        if (tab != null && (f instanceof ForwardingNode) &&
            (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {
            
            //拿当前表的长度获取扩容戳
            int rs = resizeStamp(tab.length);
            
            //判断扩容是否已经完成。
            while (nextTab == nextTable && table == tab &&
                   (sc = sizeCtl) < 0) {
                
                    /*
                     * 当前线程是否可以进行扩容的条件。
                     * 
                     * 条件一： 
                     *  (sc >>> RESIZE_STAMP_SHIFT) != rs 
                     *   true表示当前线程获得的扩容唯一表示戳 非 本批次扩容 
                     *   false表示当前线程获得的扩容唯一表示戳 是 本批次扩容
                     *  
                     * 条件二：
                     *   sc == rs + 1 (这里有BUG，后续JDK版本已经修改)
                     *   实际上想表达的是 sc == rs << 16 + 1
                     *   true 表示扩容完毕，当前线程不需要参与
                     *   false 表示扩容还在进行中，当前线程可以参与
                     *
                     *  条件三：
                     *    sc == rs + MAX_RESIZERS (这里有BUG，后续JDK版本已修改)
                     *    实际想表达的是 sc == rs << 16 + MAX_RESIZERS
                     *   true 表示当前参与的扩容线程数已经到了最大数 当前线程不需要参与
                     *   false 表示当前当前参与的扩容线程数未到达最大数 当前线程可以参与
                     *   
                     *   条件四：
                     *   transferIndex <= 0    
                     *   true 表示任务已经被分配完了。
                     *   false 表示还有任务
                     */
                
                if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                    sc == rs + MAX_RESIZERS || transferIndex <= 0)
                    break;
                
                //CAS尝试更新扩容线程数 然后加入到扩容中去。
                if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
                    transfer(tab, nextTab);
                    break;
                }
            }
            return nextTab; 
        }
        return table;
    }

这篇关于ConcurrentHashMap源码解析4.transfer()方法(核心)的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！