多线程（十六ConcurrentHashMap原理扩容）

Posted 2022-02-14

扩容的原理

扩容一般分为2个步骤

1、table数组的扩容，一般是2倍的扩容，这个是单线程操作的。

2、数据的迁移，把旧table中的各个槽中的结点重新分配到新table中。

ConcurrentHashMap在处理rehash的时候，并不会重新计算每个key的hash值，而是利用了一种很巧妙的方法。

1、ConcurrentHashMap内部的table数组的大小必须为2的幂次，原因是让key均匀分布，减少冲突
2、当table数组的大小为2的幂次时，通过key.hash & table.length-1这种方式计算出的索引i，当table扩容后（2倍），新的索引要么在原来的位置i，要么是i+n，n为扩容之前的容量。
3、这种处理方式非常利于扩容时多个线程同时进行的数据迁移操作，因为旧table的各个桶中的结点迁移不会互相影响，将整个table数组划分为很多部分，每一部分包含一定区间的桶，每个数据迁移线程处理各自区间中的结点

什么时候扩容？

链表的结点数目超过一定阈值，就会触发链表 -> 红黑树的转换，执行treeifyBin方法。

/**
 *  链表 -> 红黑树 的转换.
 */
private final void treeifyBin(Node<K, V>[] tab, int index) 
    Node<K, V> b;
    int n, sc;
    if (tab != null) 

        // CASE 1: table的容量 < MIN_TREEIFY_CAPACITY(64)时，直接table扩容，不红黑树转换
        if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
            tryPresize(n << 1); //左移一位，就是*2，扩容2倍

            // CASE 2: table的容量 ≥ MIN_TREEIFY_CAPACITY(64)时，链表 -> 红黑树的转换
        else if ((b = tabAt(tab, index)) != null && b.hash >= 0) 
            synchronized (b) 
                if (tabAt(tab, index) == b) 
                    TreeNode<K, V> hd = null, tl = null;

                    // 遍历链表，建立红黑树
                    for (Node<K, V> e = b; e != null; e = e.next) 
                        TreeNode<K, V> p = new TreeNode<K, V>(e.hash, e.key, e.val, null, null);
                        if ((p.prev = tl) == null)
                            hd = p;
                        else
                            tl.next = p;
                        tl = p;
                    
                    // 封装TreeBin，并链接到table[index]中
                    setTabAt(tab, index, new TreeBin<K, V>(hd));
                
            
        
    

扩容函数tryPresize

private final void tryPresize(int size) 
        //动态调整扩容的大小
        int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :
                tableSizeFor(size + (size >>> 1) + 1);
        int sc;
        while ((sc = sizeCtl) >= 0)  //大于等于0代表，初始化或者扩容后的（table*负载因子）的桶数量
            Node<K,V>[] tab = table; int n;
            //Case1，table没有初始化，先初始化
            if (tab == null || (n = tab.length) == 0) 
                n = (sc > c) ? sc : c;
                //设置SIZECTL为-1代表，初始化或者扩容进行中
                if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) 
                    try 
                        if (table == tab) 
                            Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];//生成新的table
                            table = nt;
                            sc = n - (n >>> 2); //n-(n/4) = 0.75n,也就是负载因子
                        
                     finally 
                        sizeCtl = sc; //设置sizeCtl为扩容后的桶*负载因子
                    
                
            
            //Case2, c<sc说明已经扩容了；n>=MAXIMUM_CAPACITY,说明桶超限了。
            else if (c <= sc || n >= MAXIMUM_CAPACITY)
                break;
            //Case3,开始扩容
            else if (tab == table) 
                int rs = resizeStamp(n);//根据容量n生成一个随机数，唯一标识本次扩容操作
                if (sc < 0)  //sc<0,说明sizeCtl<0,代表有别的线程正在进行扩容
                    Node<K,V>[] nt;
                    //不能协助扩容，退出
                    if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                            sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                            transferIndex <= 0)
                        break;
                    //加入扩容的队伍，同时sizeCtl+1，此时sizeCtl代表扩容的线程数量
                    if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                        transfer(tab, nt);
                
                //设置sizeCtl为负数，代表自己是第一个扩容的线程，CAS操作保证只有一个线程安全
                else if (U.compareAndSwapInt(this, SIZECTL, sc,
                        (rs << RESIZE_STAMP_SHIFT) + 2))
                    transfer(tab, null); //null代表首次扩容
            
        
    

数据迁移方法transfer

/**
     * tab,    旧table
     * nextTAB 扩容后的table
     */
    private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) 
        int n = tab.length, stride;
        //每个线程负责迁移table一个区间段的桶的个数，最少是16个
        if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
            stride = MIN_TRANSFER_STRIDE; // subdivide range
        //首次扩容
        if (nextTab == null)             // initiating
            try 
                //创建新的table,默认n*2
                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 = nextTab;
            transferIndex = n; //要迁移的桶的个数
        
        int nextn = nextTab.length;
        //创建扩容节点，当某个桶迁移完成后，放入table[i],标记桶扩容完成
        ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
        boolean advance = true; //为true,表示当前桶迁移完成，可以继续处理下一个桶
        boolean finishing = false; // 最后一个数据迁移的线程将该值置为true，进行扩容的收尾工作
        //i桶索引，bound就是线程要处理的另一个区间边界
        for (int i = 0, bound = 0;;) 
            Node<K,V> f; int fh;
            //定位本轮处理区间【transferIndex-1，transferIndex-stride】
            while (advance) 
                int nextIndex, nextBound;
                if (--i >= bound || finishing)
                    advance = false;
                else if ((nextIndex = transferIndex) <= 0) 
                    i = -1;
                    advance = false;
                
                else if (U.compareAndSwapInt
                        (this, TRANSFERINDEX, nextIndex,
                                nextBound = (nextIndex > stride ?
                                        nextIndex - stride : 0))) 
                    bound = nextBound;
                    i = nextIndex - 1;
                    advance = false;
                
            
            //Case1,最后一个迁移线程或者是线程出现了冲突，导致了i<0
            if (i < 0 || i >= n || i + n >= nextn) 
                int sc;
                if (finishing)  //迁移完成
                    nextTable = null;
                    table = nextTab;
                    sizeCtl = (n << 1) - (n >>> 1);
                    return;
                
                //扩容线程数减1,当前线程任务已执行完成
                if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) 
                    //判断是否最后一个迁移线程，不是则退出
                    if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
                        return;
                    finishing = advance = true;
                    //最后一个线程，还原i值，重新进行检查，是否全部迁移完成，应该所有桶都是ForwardingNode
                    i = n; // recheck before commit
                
            
            // CASE2：旧桶本身为null，不用迁移，放一个ForwardingNode
            else if ((f = tabAt(tab, i)) == null)
                advance = casTabAt(tab, i, null, fwd);
            //CASE3：该旧桶已经迁移完成，直接跳过，hash==moved 代表ForwardingNode
            else if ((fh = f.hash) == MOVED)
                advance = true; // already processed
            else 
                // CASE4：该旧桶未迁移完成，进行数据迁移，加锁
                synchronized (f) 
                    if (tabAt(tab, i) == f) 
                        Node<K,V> ln, hn;
                        if (fh >= 0)  //桶是链表，迁移链表

                            /**
                             * 下面的过程会将旧桶中的链表分成两部分：ln链和hn链
                             * ln链会插入到新table的槽i中，hn链会插入到新table的槽i+n中
                             */
                            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;
                                if (b != runBit) 
                                    runBit = b;
                                    lastRun = p;
                                
                            
                            if (runBit == 0) 
                                ln = lastRun;
                                hn = null;
                            
                            else 
                                hn = lastRun;
                                ln = null;
                            
                            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);
                            
                            setTabAt(nextTab, i, ln);  // ln链表存入新桶的索引i位置
                            setTabAt(nextTab, i + n, hn); // hn链表存入新桶的索引i+n位置
                            setTabAt(tab, i, fwd); // 设置ForwardingNode占位
                            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;
                            /**
                             * 先以链表方式遍历，复制所有结点，然后根据高低位组装成两个链表；
                             * 然后看下是否需要进行红黑树转换，最后放到新table对应的桶中
                             */
                            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) :
                                    (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);// 设置ForwardingNode占位
                            advance = true; // 表示当前旧桶的结点已迁移完毕