ConcurrentHashMap浅析

ConcurrentHashMap：在HashMap的基础上加上分段锁

• 关于HashMap，可以参考另一篇文章
  https://www.jianshu.com/p/68b5f23e45bd

底层结构

ConcurrentHashMap默认结构.png

ConcurrentHashMap的初始化：

• 不同于HashMap，ConcurrentHashMap的初始化有两步

  • 1、初始化段Segment[]

    • 默认大小：DEFAULT_CONCURRENCY_LEVEL=16

  • 2、初始化数组Entry[]

    • 默认大小：DEFAULT_INITIAL_CAPACITY=16

  • 我们可以发现分段锁和数组的初始化个数是一样的，这就意味着每一个数组元素对应着一把锁

    
    image.png

  • 这种结构的好处是具有较高的并发能力，但是有个显而易见的缺点：Segment太多了

  插入过程

  • 前面的初始化结果是申请到了空间，并没有直接new出segment
  • 当我们插入的时候，如果没有则new一个segment(实际上也是一个小HashMap)

//初始化之后的segment,会先算出ssize（算出来之后就不会改变）
//ssize就是找到大于concurrencylevel 的最小的2次方值
new segment(){
      new entry[DEFAULT_INITIAL_CAPACITY/ssize];//这个算出来一定是整除的
}

• entry数组的大小由数组总大小除以level，之后每次new segment的时候就直接用之前算好的大小即可（例如：s0.entry.size()）

• 接下来（前文提到segment也是一个小HashMap）：

  
  image.png

---

Java8中的改变

• 和HashMap相同都多了一个红黑树的结构，红黑树的优势在于插入和查询效率都还可以，都是O(logn)层次，可以避免只采用链表时的老HashMap中链表过长的而导致查询效率地下的情况。
• 后续更新学习中。。。。。。

---

研究了一下源码
8中将segment取消了（为了兼容segment类还保留但是没用到），而是采用CAS算法，这无疑会提升效率，因为少操作一层。换句话说，8将segment和数组合二为一了。

CAS

• 这边查出tab数组中的一个节点是空的，就会调用CAS算法U.compareAndSwapObject（...）在Unsafe类中定义，其目的是比较该元素是否为空，若为空则new一个Node放入，若不为空（但上面查出是空，则证明有别的线程干了这件事）

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
            }

• 假如检测到这种情况，则直接break跳出循环，再执行for (Node<K,V>[] tab = table;;)此时该位置上有元素，则会跳到正常条件中执行put操作（插入或者覆盖）：

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;
                }
            }

这个特性可以更加有效防止多线程死循环的情况

除此之外，判断中还有一段代码

//假如检测到扩容，则帮忙
else if ((fh = f.hash) == MOVED)    tab = helpTransfer(tab, f);

我们可以看看helpTransfer都干了啥

    /**
     * Helps transfer if a resize is in progress.
     */
    final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
        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) {
                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;
    }

• 则可以得出结论，当该线程发现扩容时，执行此方法，帮助一起扩容，这意为着，多线程情况下扩容操作不会带来隐患，反而会提升ConcurrentHashMap的运行速度。

  
  多线程同时扩容.png