ConcurrentHashMap Java8源码分析

ConcurrentHashMap在Java8中的实现改动较大，网上关于ConcurrentHashMap的文章也很少有基于java8的，将个人的一些理解记录下来以供分享。

Node

ConcurrentHashMap底层是通过数组+链表(树)来实现的，数组中存储的就是Node。它与HashMap中的定义很相似，但是有一些差别它对value和next属性设置了volatile同步锁，它不允许调用setValue方法直接改变Node的value域，它增加了find方法辅助map.get()方法。

transient volatile Node<K,V>[] table;

  static class Node<K,V> implements Map.Entry<K,V> {
       //当前节点的hash值
        final int hash;
       //当前节点的Key
        final K key;
       //当前节点的值，保证了可见性
        volatile V val;
       //下一个节点
        volatile Node<K,V> next;
...

TreeNode

树节点类，另外一个核心的数据结构。 当链表长度过长的时候，会转换为TreeNode。 但是与HashMap不相同的是，它并不是直接转换为红黑树，而是把这些结点包装成TreeNode放在TreeBin对象中，由TreeBin完成对红黑树的包装。 而且TreeNode在ConcurrentHashMap继承自Node类，而并非HashMap中的集成自LinkedHashMap.Entry

TreeBin

这个类并不负责包装用户的key、value信息，而是包装的很多TreeNode节点。它代替了TreeNode的根节点，也就是说在实际的ConcurrentHashMap“数组”中，存放的是TreeBin对象，而不是TreeNode对象，这是与HashMap的区别。另外这个类还带有了读写锁。 这里仅贴出它的构造方法。可以看到在构造TreeBin节点时，仅仅指定了它的hash值为TREEBIN常量，这也就是个标识位

ForwardingNode

一个用于连接两个table的节点类。它包含一个nextTable指针，用于指向下一张表。而且这个节点的key value next指针全部为null，它的hash值为-1. 这里面定义的find的方法是从nextTable里进行查询节点，而不是以自身为头节点进行查找。通过名称也很容易理解这个节点的含义这个节点主要作用就是重定向，在resize过程中尽可能不影响对数据对读取。

关键方法

put方法

ConcurrentHashMap中key可为null吗？value可以为null吗？为什么呢？HashMap又是什么情况呢？ 通过下面的代码可以发现，ConcurrentHashMap中的key和value都不能为null,这是因为在Node中计算hash的时候使用的是key.hashCode() ^ val.hashCode()；而在HashMap中是Objects.hashCode(key) ^ Objects.hashCode(value);

主要的处理步骤有：

• 计算key的hash值
• 如果没有初始化需要进行初始化
• 通过按位与进行快速取模计算出桶位置，如果该位置没有元素则通过CAS插入，如果该节点是FWD节点，则帮助完成扩容
• 如果有元素，也不是FWD节点则对该节点进行加锁后插入元素
• 插入元素后，进行元素计数加1，在该步操作中可能会出发扩容操作

final V putVal(K key, V value, boolean onlyIfAbsent) {
      //concurrentHashMap要求key,value都不能为null.为什么呢？这是因为 Node中计算hash的方法是key.hashCode() ^ val.hashCode()；
     //如果key或者value为null在此处计算hash值的时候会出现NPE
        if (key == null || value == null) throw new NullPointerException();
        //计算key的hash值（两次hash计算，通过掩码运算得到内部hash值)
        int hash = spread(key.hashCode());
        int binCount = 0;
        //死循环直到操作成功
        for (Node<K,V>[] tab = table;;) {
            Node<K,V> f; int n, i, fh;
            //此处如果tab=null,或者tab.length-0表示表没有初始化，此处需要完成初始化！
            //这个就是ConcurrentHashMap将初始化延迟到第一次put操作时候完成
            if (tab == null || (n = tab.length) == 0)
                tab = initTable();
           //i=(n-1)&hash这个计算和HashMap相同，这个是对hash值对快速取模操作，i的值即表示该数据存放的桶号
            //如果在i位置的桶中没有元素则通过CAS操作试图将元素插入到该位置，如果插入成功则退出循环！在向空桶中添加元素时是无锁操作
            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
            }
            //如果这个桶的节点的hash值是MOVED则表示该节点在进行扩容操作
            else if ((fh = f.hash) == MOVED)
                tab = helpTransfer(tab, f);
            else {
               //对正常的桶通过synchronized进行加锁，synchronized锁已经进行了很大性能优化
                V oldVal = null;
                synchronized (f) {
                    //重复检查当前桶是否有发生改变,如果没有发生改变才做后续处理
                    if (tabAt(tab, i) == f) {
                         //fn是这个桶的hash值，大于等于0表示链表节点
                        if (fh >= 0) {
                            binCount = 1;
                            //向后遍历，如果存在重复的key则更新value否则插入到链表尾部跳出循环
                            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;
                                   //put操作时候onlyIfAbsent为false，所以此处会进行值更新
                                    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) {
                    //如果链表的长度大于等于8则需要将链表转换为树，可以发现首先是将数据插入到链表中再判断是否需要进行转换为树
                    if (binCount >= TREEIFY_THRESHOLD)
                        treeifyBin(tab, i);
                    if (oldVal != null)
                        return oldVal;
                    break;
                }
            }
        }
        //计数增加1，该操作中可能会出发扩容操作
        addCount(1L, binCount);
        return null;
    }

private final void addCount(long x, int check) {
        CounterCell[] as; long b, s;
       //BASECOUNT 为baseCount属性的偏移量，如果原值是baseCount则更新为b+x;
        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();
        }
        //counterCells为null,check>0 ,s=b+x, 在put方法中x=1,b=baseCount,在resize之前
        //sizeCtl=16*0.75=12
        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);
               //sc小于0表示 sc绝对值-1个线程在进行扩容，sc=0表示还没有初始化，此处不可能为0，如果大于0表示需要下次
               //扩容的阈值
                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);
                }
                //对sizeCtl更新如果成功就扩容，否则计算大小继续循环
                else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                             (rs << RESIZE_STAMP_SHIFT) + 2))
                    transfer(tab, null);
                s = sumCount();
            }
        }
    }
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
        //n为数组大小
        int n = tab.length, stride;
        //NCPU表示机器的CPU核心数,如NCPU=4，第一次n=16,stride=16;否则stride=n>>>3/ncpu
        if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
            stride = MIN_TRANSFER_STRIDE; // subdivide range
       //如果nextTab为空，表示尚未进行扩容处理
        if (nextTab == null) {            // initiating
            try {
                //创建长度为原来两倍的数组，并赋值给nextTab
                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赋值
            nextTable = nextTab;
            //transferIndex时扩容前数组的长度
            transferIndex = n;
        }
        //获取扩容后的数组长度
        int nextn = nextTab.length;
        //创建forwarding节点
        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;
             //只要advance=true就一直循环下去
            while (advance) {
                int nextIndex, nextBound;

                if (--i >= bound || finishing)
                    advance = false;
               //如果transferIndex<=0则退出while循环
                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;
                }
            }
            //第一次 i=nextIndex-1,nextIndex=transferIndex=n=tab.length,即i是原有数组长度减去1
            if (i < 0 || i >= n || i + n >= nextn) {
                int sc;
                if (finishing) {
                    nextTable = null;
                    table = nextTab;
                    sizeCtl = (n << 1) - (n >>> 1);
                    return;
                }
                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
                }
            }
            //如果数组i位置为空则直接fwd节点插入
            else if ((f = tabAt(tab, i)) == null)
                advance = casTabAt(tab, i, null, fwd);
           //如果节点的哈希值为MOVED表示已经处理过了，则将advance设置为true，处理其他的桶，这样有效避免了多线程重复处理过程
            else if ((fh = f.hash) == MOVED)
                advance = true; // already processed
            else {
               //否则对该节点进行加锁
                synchronized (f) {
                    //加锁后再次判断i位置是否是加锁的对象
                    if (tabAt(tab, i) == f) {
                         //ln表示低位节点，hn表示高位节点，在经过一次扩容后，根据桶号计算 h&(n-1)可以知道
                         //该节点要不在原位置，要不就向后移动扩容大小位，ln就是位置不变的元素，hn就是移动扩容
                         //大小的元素
                        Node<K,V> ln, hn;
                        //fh节点的hash值，大于0标志正常链表节点
                        if (fh >= 0) {
                            int runBit = fh & n;
                            Node<K,V> lastRun = f;
                            //从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;
                            }
                            //从f节点开始遍历非最后一个节点,计算出高位节点和低位节点
                            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);
                            }
                            //i是tab中元素的索引，n是tab的长度
                            //在nextTab中i节点中设置ln
                            setTabAt(nextTab, i, ln);
                            //在i+n位置上插入hn
                            setTabAt(nextTab, i + n, hn);
                            //将原有i位置设置为forwarding节点
                            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;
                        }
                    }
                }
            }
        }
    }

   

本文来自网易实践者社区，经作者张伟授权发布。