【Java并发】聊聊concurrentHashMap扩容核心流程

扩容

什么时候扩容

• 链表转红黑树。需要判断数组长度，触发扩容
• 调用putAll() , 触发tryPresize() 方法
• 数据量达到阈值

tryPresize-初始化数组

   // 扩容前操作，putAll or 链表转红黑树 
    // size是原数组长度 * 2
    private final void tryPresize(int size) {
        // putAll()使用，插入map的长度 （putAll）
        // size > 最大长度 / 2 就是最大值
        // 否则将长度设置为2的幂次
        // c是初始化数组长度
        int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :
            tableSizeFor(size + (size >>> 1) + 1);
        int sc;
        //sizeCtl
     // -1:正在初始化数组，小于-1：正在扩容，0：代表还没初始化数组，大于0：可能初始化了(代表阈值)，也可能没初始化（初始化的长度）
        // 没有正在进行初始化 或者 正在扩容
        while ((sc = sizeCtl) >= 0) {
            Node<K,V>[] tab = table; int n;
             //数组没有初始化 ，在put方法的时候，其实也走了这样的逻辑。
            if (tab == null || (n = tab.length) == 0) {
               // 在sc和c之间 选择最大值作为初始化长度 
                n = (sc > c) ? sc : c;
                // 初始化 sizeCtl = -1 
                if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
                    try {
                        if (table == tab) {
                            @SuppressWarnings("unchecked")
                            Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                            table = nt;
                            sc = n - (n >>> 2);
                        }
                    } finally {
                        sizeCtl = sc;
                    }
                }
            }
            else if (c <= sc || n >= MAXIMUM_CAPACITY)
                break;
            // 扩容前操作    
            else if (tab == table) {
                 // 计算扩容标识戳 多线程迁移数据
                int rs = resizeStamp(n);
                if (sc < 0) { // 正在扩容 or 初始化
                    Node<K,V>[] nt;
                    if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                        sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                        transferIndex <= 0)
                        break;
                    if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                        transfer(tab, nt);
                }
                // 当前执行扩容，可能是第一个进来执行扩容的线程
                // CAS = SIZECTL  一定是小于 -1
                else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                             (rs << RESIZE_STAMP_SHIFT) + 2))
                    //开始扩容 传输老数组
                    transfer(tab, null);
            }
        }
    }

1.计算表示戳
2.正在扩容线程数

tryPreSize方法-扩容标识戳

    static final int resizeStamp(int n) {
    // 32~64
    // 00000000 00000000 00000000 00011010
    // 计算n在二进制表示时，前面有多少个0
    // 00000000 00000000 10000000 00000000
    // 00000000 00000000 10000000 00011010
    // 前面的操作是基于数组长度等到一个标识，方便其他线程参与扩容
    // 后面的值是为了保证当前扩容戳左移16位之后，一定是一个负数
        return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1));
    }

扩容操作 transfer

    private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
        int n = tab.length, stride;
        // 不超过CPU内核数的线程进行扩容
        // 计算每个线程处理数据的区间大小 最小是16
        if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
            stride = MIN_TRANSFER_STRIDE; // subdivide range
        // 新数组没有初始化 进行初始化 
        // 在原来的基础上扩大两倍  
        if (nextTab == null) {            // 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;
                return;
            }
            nextTable = nextTab;
            
            transferIndex = n;
            // transferIndex old数组的长度
        }
        // n ：老数组长度 
        // stride: 步长
        // nextTable, nextTab : 新数组
        // nextn : 新数组长度
        // transferIndex 线程领取任务时的核心属性
        int nextn = nextTab.length;
        //先声明fwd节点、在老数组迁移数据完成后，将fwd赋值上去
        // 表示已经迁移完的，完成迁移 修改成fwd 
        ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
        boolean advance = true;
        boolean finishing = false; // to ensure sweep before committing nextTab
        for (int i = 0, bound = 0;;) {
            Node<K,V> f; int fh;
            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;
                }
            }
            // 是否扩容完成
            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 = n; // recheck before commit
                }
            }
            // 当前数组为空，直接修改成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 {
                // 针对当前要迁移的节点 加syn🔒
                synchronized (f) {
                    if (tabAt(tab, i) == f) {
                        Node<K,V> ln, hn;
                        if (fh >= 0) {
                            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;
                                }
                            }
                            // == 0 不用迁移
                            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);
                            // 新数组的新位置
                            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);
                                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);
                            advance = true;
                        }
                    }
                }
            }
        }
    }

size如何进行统计？

因为chm是put线程安全，且同一时刻可能多个线程操作数据，那么如何统计有多少个元素呢，如果使用cas的方式，也就是atoxxx的。在并发量小的时候可以计算。当并发量比较高的时候，性能就下降了。chm很巧妙的通过并发的粒度来做的处理。
当并发量竞争小的时候通过size 进行统计，当并发竞争高cas失败的时候，每个链表维护一个一个cell,当调用size 统计的时候，在逐个统计。但是可能存在数据不准确，是一种弱一致性。

    private final void addCount(long x, int check) {
        CounterCell[] as; long b, s;
         // 统计元素的个数
        if ((as = counterCells) != null ||
            // cas 修改baseCount 
            !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
            CounterCell a; long v; int m;
            boolean uncontended = true;
            if (as == null || (m = as.length - 1) < 0 ||
                (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
                !(uncontended =
                  U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
                  // 添加元素个数
                fullAddCount(x, uncontended);
                return;
            }
            if (check <= 1)
                return;
            s = sumCount();
        }
        // 是否要做扩容
        if (check >= 0) {
            Node<K,V>[] tab, nt; int n, sc;
            // 第一次扩容的场景
            while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
                   (n = tab.length) < MAXIMUM_CAPACITY) {
                int rs = resizeStamp(n);
                if (sc < 0) {
                    if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                        sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                        transferIndex <= 0)
                        break;
                    if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                        transfer(tab, nt);
                }
                else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                             (rs << RESIZE_STAMP_SHIFT) + 2))
                    transfer(tab, null);
                s = sumCount();
            }
        }
    }

    final long sumCount() {
        CounterCell[] as = counterCells; CounterCell a;
        long sum = baseCount;
        //as != null 说明竞争高，使用countterCell 数组统计，否则直接返回sum的和
        if (as != null) {
            for (int i = 0; i < as.length; ++i) {
                if ((a = as[i]) != null)
                    sum += a.value;
            }
        }
        return sum;
    }

get()

在查询数据时，会先判断当前key对应的value，是否在数组上。

其次会判断当前位置是否属于特殊情况：数据被迁移、位置被占用、红黑树结构

最后判断链表上是否有对应的数据。

找到返回指定的value，找不到返回null即可

    public V get(Object key) {
        Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
        // 和put时一样的操作，先计算k的hashcode 
        int h = spread(key.hashCode());
        // 对应的下标不为空
        if ((tab = table) != null && (n = tab.length) > 0 &&
            (e = tabAt(tab, (n - 1) & h)) != null) {
            // 第一个元素就是 直接返回
            if ((eh = e.hash) == h) {
                if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                    return e.val;
            }
            // 树结构查找
            else if (eh < 0)
                return (p = e.find(h, key)) != null ? p.val : null;
            // 链表结构查找
            while ((e = e.next) != null) {
                if (e.hash == h &&
                    ((ek = e.key) == key || (ek != null && key.equals(ek))))
                    return e.val;
            }
        }
        return null;
    }

针对于chm，主要从几个方面学习

• 1.7 8的区别
• put核心流程
• get核心流程
• 扩容流程
• size流程