JDK1.8 concurrentHashMap设计原理与优化

前面写到了JDK1.8 hashMap优化，那么在jdk1.8中对concurrentHashMap也做了不少的优化，本文也会从源码来分析下这些优化，以及为什么要这么做。首先还是提出需要关注的点:

• 为什么使用CAS+synchronized替换了segemnt+ReentrantLock来保证线程安全
• concurrentHashMap为什么不允许key和value为null
• 如何高效实现在多线程情况下扩容
• 黑科技@sun.misc.Contended

存储结构

jdk1.8concurrentHashMap的存储结构基本和hashMap是一样的了，就是单纯的数组链表；不再是1.7里面的segments数组，每一个segment相当于是一个map，而且每一个segment都实现ReentrantLock。

显然对比起现在的实现，在内存上是有浪费的。

找到真正的桶需要两次hash

如下图所示：

synchronized锁优化

提起synchronized，可能大家都有个印象是这个锁是重量级的，而且性能在线程竞争比较激烈情况下比reentrantLock差。例如下图是JDK1.5对比,确实是这样。 但是这已经是过往了，synchronized作为java语法层面的同步，java虚拟机是可以在线程和对象元数据中记录synchronized中锁的信息的，开发团队是投入了大量的精力进行优化的。

前面提到了可以在对象的元数据里面存储synchronized信息，在对象头里面有个“Mark word”，正是这个Mark word使对象可以表示：未锁定；轻量级锁；重量级锁；偏向锁几种状态。而且还会有自适应性自旋的优化。

回到concurrentHashMap的put方法中,可以发现，当确定了元素要放入的桶之后，如果桶里面没有节点，要么不用加锁，直接CAS尝试将该元素作为头结点；如果里面有了其他的元素，这时才会对头结点f进行加锁，锁住这个链表。也就是说每个桶都是单独的锁，这样不同线程并发争抢到同一个桶的概率已经很小，而且由于自适应自旋机制，就算发生了争抢，也不会将线程挂起，自旋等待即可。实际上这已经是乐观锁的设计思路了。

• reentrantLock本质上是一个悲观锁，就算在非公平模式下，会尝试一次CAS获取锁，如果获取不到，那么会被添加到AQS等待队列再尝试一次，还是获取不到的就会被阻塞住，顺序等待被唤醒。

V putVal(K key, V value, boolean onlyIfAbsent) {
        if (key == null || value == null) throw new NullPointerException();
        int hash = spread(key.hashCode());
        int binCount = 0;
        for (Node<K,V>[] tab = table;;) {
            Node<K,V> f; int n, i, fh;
            if (tab == null || (n = tab.length) == 0)
                tab = initTable();
            else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
                if (casTabAt(tab, i, null,
                             new Node<K,V>(hash, key, value, null)))
                    break; // no lock when adding to empty bin
            }
            else if ((fh = f.hash) == MOVED)
                tab = helpTransfer(tab, f);
            else {
                V oldVal = null;
                synchronized (f) {
                    if (tabAt(tab, i) == f) {
                        if (fh >= 0) {
                            binCount = 1;
                            for (Node<K,V> e = f;; ++binCount) {
                                K ek;
                                if (e.hash == hash &&
                                    ((ek = e.key) == key ||
                                     (ek != null && key.equals(ek)))) {
                                    oldVal = e.val;
                                    if (!onlyIfAbsent)
                                        e.val = value;
                                    break;
                                }
                                Node<K,V> pred = e;
                                if ((e = e.next) == null) {
                                    pred.next = new Node<K,V>(hash, key,
                                                              value, null);
                                    break;
                                }
                            }
                        }
                        else if (f instanceof TreeBin) {
                            Node<K,V> p;
                            binCount = 2;
                            if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
                                                           value)) != null) {
                                oldVal = p.val;
                                if (!onlyIfAbsent)
                                    p.val = value;
                            }
                        }
                    }
                }
                if (binCount != 0) {
                    if (binCount >= TREEIFY_THRESHOLD)
                        treeifyBin(tab, i);
                    if (oldVal != null)
                        return oldVal;
                    break;
                }
            }
        }
        addCount(1L, binCount);
        return null;
    }

高效扩容

扩容是个很消耗性能的点，之前在hashMap中也提到过，最好能够预估需要存入的数据数量，初始化就设定号，尽量避免频繁的扩容，特别是在多线程环境中，更加需要小心谨慎，加上各种“骚操作”才能提升效率，能从根本上避免就最好。

private final void addCount(long x, int check) {
    CounterCell[] as; long b, s;
    //利用CAS方法更新 baseCount 的值
    if ((as = counterCells) != null ||
        !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {// 1
        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))) {
            // 多线程修改baseCount时，竞争失败的线程会执行fullAddCount(x, uncontended),把x的值插入到counterCell类中
            fullAddCount(x, uncontended); // 2
            return;
        }
        if (check <= 1)//这里添加之后，不是树就不用操作了
            return;
        s = sumCount();
    }
    //如果check值大于等于0 则需要检验是否需要进行扩容操作
    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);
            // 如果小于0 说明已经有线程在进行扩容了
            if (sc < 0) {
                // 以下的情况说明已经有在扩容或者多线程进行了扩容，其他线程直接 break 不要进入扩容
                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);
            }
            // 当前线程是唯一的或是第一个发起扩容的线程  此时nextTable=null
            else if (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2))
                transfer(tab, null);
            s = sumCount();
        }
    }
}

final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
    Node<K,V>[] nextTab;int sc;
    //条件：原结点不为空,这个结点是协助结点（ForwardingNode）并且不处于最后
    if (tab != null && (f instanceof ForwardingNode) &&
        (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {
        int rs = resizeStamp(tab.length);
        //sizeCtl是负数，标明还在扩容；tab和现在占用的table不同，说明扩容还在进行当中
        while (nextTab == nextTable && table == tab && (sc = sizeCtl) < 0) {
            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;
}

@sun.misc.Contended

在源码里面CounterCell是被Contended注解的，表示消除伪共享，因为这个值表示桶里面的元素个数，是经常改变的，也就是在缓存中是经常要被置为无效的，如果有其他的变量与其在同一个缓存行，那么会相互影响，降低并发量。

    @sun.misc.Contended static final class CounterCell {
        volatile long value;
        CounterCell(long x) { value = x; }
    }

key value不允许为null

hashMap中是允许key或者value为null的，那么为什么concurrentHashMap不允许呢？其实作者Doug Lea大神亲自解释过：也就是说当get(key)为null，在concurrentHashMap中难以分辨出是不存在这个key对应的value还是本身key对应的value就是null。