ConcurrentHashMap源码简单分析

Posted 2021-09-07 SanPiBrother

ConcurrentHashMap源码简单分析

本文只对java8的ConcurrentHashMap源码作一些简单分析，java8的ConcurrentHashMap相对于java7来说，代码变动较大，性能提升比较明显。最大的特点是java7的ConcurrentHashMap通过分段锁来保证线程安全，但是锁的粒度不够精细，java8中通过Synchronized 锁住桶的链表first结点，进一步缩小了竞争的范围。此外还有一些其它有意思的地方，如多线程帮助扩容机制等等，这些值得一探究竟。

容量的计算方式：tableSizeFor方法分析

ConcurrentHashMap中在创建时，会调用tableSizeFor方法进行计算尝试容量大小，这个方法原理就是通过巧妙的位运算，获取最接近入参的2的幂次方数。

简单思考，如果这个数本身不是2的幂次方，如何根据这样的数，计算最接近2的幂次方数呢？首先任意一个数，都可以表示为00...01XXX...XXX, 那么最接近它的2幂次方数肯定是最高位左移一位，低位全0，即：00...10000...000，那么如何得到这样的数呢？仔细观察发现，00...10000...000，其实就是00...01111...111+1得，所以问题就转为了如何求从最高位开始，低位全1的数，而这种数，我们可以通过位运算+或运算得到。

   private static final int tableSizeFor(int c) {
        // 减一是针对入参c恰好是二的幂次方数
        // 此时经移位后输出应该还是c本身
        int n = c - 1;
        //通过移位加或运算，使n最终变成00...01111...111
        // 移16位为止，是因为int型数据，总共32位，移动16刚好
        n |= n >>> 1;
        n |= n >>> 2;
        n |= n >>> 4;
        n |= n >>> 8;
        n |= n >>> 16;
        return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;
    }

举例：一个非2幂次方数
原始入参： 00...01XXX...XXX1
参数减一： 00...01XXX...XXX0
左移一位：00...001XX...XXX0
求或： 00...011XX...XXX0
左移二位： 00...00011X...XXX0
求或： 00...01111x...XXX0
...
发现规律没？每次移n位(n为2的幂次方)，再与移位前的数求或，会使从高位开始的n位变成1，最终
移16位，正好变成： 00...011111...1111，此时原来的数，就变成了从高位开始全1的数，这个时候，获取最接近它的幂次方数就非常简单了，直接加1，
变成：00...011111...1111-->00...100000...0000

ConcurrentHashMap如何正确统计size

我们知道ConcurrentHashMap是支持多线程的，在多线程情况下，是如何做到准确统计map的size呢？

ConcurrentHashMap中，计算size的方法时sumCount()，方法如下：统计的数据来源有两种，一种是来自计数盒子，也就是CounterCell[]这样的数组，CounterCell对象里面存放的是volatile类型的变量；另一种来自baseCount，当然也是volatile类型，保证线程安全。
从该方法可知：只有计数盒子不存在的时候，才会采用baseCount的值作为size返回。

    final long sumCount() {
        CounterCell[] as = counterCells; CounterCell a;
        long sum = baseCount;
        if (as != null) {
            for (int i = 0; i < as.length; ++i) {
                if ((a = as[i]) != null)
                    sum += a.value;
            }
        }
        return sum;
    }

那么什么时候采用的是计数盒子，也就是CounterCell[]数组这种计数方式？什么时候又是采用
baseCount作为计数方式？答案就是：单线程修改计数时，用baseCount，一旦出现了多线程修改计数，那么后面就会抛弃baseCount，一直使用CounterCell[]计数。
增加计数方法为addCount(long x, int check)，源码如下。当然这个方法除了计数外还有一个非常重要的功能，那就是扩容（扩容放到另一个小结讲）。

 private final void addCount(long x, int check) {
        CounterCell[] as; long b, s;
        if ((as = counterCells) != null ||
            !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();
        }
 // 以下省略
}

计数代码中，其实就是通过CAS修改baseCount值，如果失败，就意味着存在并发，此时需要创建CounterCell[] 来保存每个线程添加的元素个数，之所以选择数组，是尽量让不同线程操作不同的volatile变量，以提高效率，注意，数组大小和线程数量无关，初始化大小为2，遵从2倍扩容，只是为了减少CAS修改volatile变量次数，但是不能避免这个问题，同时定位线程在数组中位置，与Map的定位方式类似，只不过是采用线程随机数来&(数组长度-1)。
其实这种思想和LongAdder,Striped64一样，有兴趣的话可以研究一下。

ConcurrentHashMap的get操作凭什么无锁

ConcurrentHashMap在多线程的情况下，是如何保证不依赖锁而get到正确的数据呢？
首先并发分两种场景考虑：
1、多线程在对ConcurrentHashMap扩容后，正在对数据进行迁移，此时get的数据恰好在被迁移。
2、多个线程都在修改ConcurrentHashMap中同一个数据，如何不会get到脏数据。
针对这些疑问，扒一扒源码如下：

    public V get(Object key) {
        Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
        int h = spread(key.hashCode());
        //先通过key的Hash值定位到所查元素所在桶的位置
        if ((tab = table) != null && (n = tab.length) > 0 &&
            (e = tabAt(tab, (n - 1) & h)) != null) {
            if ((eh = e.hash) == h) {
                // 恰好桶中第一个元素就满足，直接返回
                //比较方式先==后equals,效率高，注意要equals的重写
                if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                    return e.val;
            }
            //eh=-1，说明当前的数据正被迁移，调用ForwardingNode的find方法在新的数组中查找
            // eh=-2，说明该节点是一个TreeBin，此时调用TreeBin的find方法遍历红黑树
            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;
    }

看完get方法的原码，对于上面场景1的问题，就知道答案了，当扩容后，数据被迁移时，旧的数组中的结点会被替换为ForwardingNode，Hash值置为-1，且其内部维护了一个nextTable指向新的数组，当我们发现需要查询的桶的Hash值为-1时，就会去新的数组中查找。
对于场景2的问题，使用volatile变量就能解决，注意Node中保存的value使用了volatile修饰，可以保证线程之间可见性。

ConcurrentHashMap 扩容探究

ConcurrentHashMap 在1.8中，扩容效率得以提升，除了正常扩容流程外，如果其它线程在put的时候，发现Map正在扩容，也会帮助扩容。迁移中如何确定旧的结点在新的桶中位置，放后面讲。
先分析一下正常的扩容源码：还是熟悉的addCount方法，前面介绍过它的计数功能，接下来介绍它的扩容功能。

 private final void addCount(long x, int check) {
        // 上面计数代码前面已探究，这里省略
        if (check >= 0) {
            Node<K,V>[] tab, nt; int n, sc;
            //当前size大于等于阈值，table已初始化且不超过极限容量（2的30次方）
            while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
                   (n = tab.length) < MAXIMUM_CAPACITY) {
                 //得到size的标识符
                int rs = resizeStamp(n);
                // 如果正在扩容
                if (sc < 0) {
                     // sc >>> RESIZE_STAMP_SHIFT) != rs，因为rs是根据size计算出来的，sc又是rs左移16位加上当前扩容线程数+1得到，如果sc高16位不等于rs，很可能是size已变
                     // sc == rs + 1 表示扩容结束，因为第一个线程扩容时：sc 等于rs << RESIZE_STAMP_SHIFT) + 2
                     // sc == rs + MAX_RESIZERS 扩容线程是否已达到上限
                     //nextable 为 null 或 transferIndex <= 0表示扩容已结束
                    if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                        sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                        transferIndex <= 0)
                        break;
                    //对于扩容已开始情况，在允许扩容时，开始帮助扩容，sc+1表示扩容线程数加1
                    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();
            }
        }
    }

接下来就是调用transfer方法真正进行扩容，transfer方法是在巨长无比，所以省略部分

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
        int n = tab.length, stride;
        //根据CPU数计算当前线程需要负责多少个桶的数据迁移
        if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
            stride = MIN_TRANSFER_STRIDE; // subdivide range
        //定义一个新的桶数组，容量为旧的2倍
        if (nextTab == null) {            // initiating
           //代码略
        }
        int nextn = nextTab.length;
        //定义迁移结点，当旧的数组中桶被迁移时，会被替换为迁移结点，其Hash值固定为-1
        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;
            //这里用于获取当前线程需要处理的数组中桶的坐标范围i~bound
            while (advance) {
               //略
            }
            // 判断当前线程负责的范围的桶是否已完成迁移
            if (i < 0 || i >= n || i + n >= nextn) {
              //如果已完成，则旧的table替换为扩容后的table
             // 扩容线程数减1
            }
            // 旧的桶本身为null，则直接替换为迁移结点，hash值为-1
            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) {
                        Node<K,V> ln, hn;
                        // 头结点Hash值大于0，桶为链表，进行迁移
                        if (fh >= 0) {
                           //略
                        }
                        // 桶为红黑树时，进行迁移
                        else if (f instanceof TreeBin) {
                            //略
                        }
                    }
                }
            }
        }
    }

扩容中，数据迁移时，如何确定Node在新的桶中位置？

ConcurrentHashMap在扩容后的数据迁移时，不需要重新Hash确定位置。

简单说一下它是怎么做的?
在put一个元素的时候，是hash&(n-1), n为旧的size，当扩容后（2倍扩容，n<<1），则计算hash&n 后的值，判断是0，还是大于0，如果为0，则当前元素在新的容器中位置保持不变，如果为大于0，则是将当前位置值+n（n为扩容前size）
这是个巧妙的位运算，如果还无法理解，就带值算一下就明白了

杂记

1、ConcurrentHashMap中链表达到阈值8的时候，并不一定会转换为红黑树，而是会判断当前size是否大于等于MIN_TREEIFY_CAPACITY （64），是才会转为树，否则只是扩容

 private final void treeifyBin(Node<K,V>[] tab, int index) {
        Node<K,V> b; int n, sc;
        if (tab != null) {
            if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
                tryPresize(n << 1);
          ...
}

当树的深度小于等于6的时候，就重新转为链表。
ps 网上说阈值8基于松分布和负载因子得到的，没考究过。

2、ConcurrentHashMap在new出来后，并不会初始化里面的Node<K,V>[]数组，而是在put的时候，才去初始化，懒加载。