说一说ConcurrentHashMap的实现原理

得分点

​ 数组+链表+红黑树、锁头节点

参考答案

标准回答

​ 在JDK8中，ConcurrentHashMap的底层数据结构与HashMap一样，也是采用“数组+链表+红黑树”的形式。同时，它又采用锁定头节点的方式降低了锁粒度，以较低的性能代价实现了线程安全。底层数据结构的逻辑可以参考HashMap的实现，下面我重点介绍它的线程安全的实现机制。

1. 初始化数组或头节点时，ConcurrentHashMap并没有加锁，而是CAS的方式进行原子替换（原子操作，基于Unsafe类的原子操作API）。
2. 插入数据时会进行加锁处理，但锁定的不是整个数组，而是槽中的头节点。所以，ConcurrentHashMap中锁的粒度是槽，而不是整个数组，并发的性能很好。
3. 扩容时会进行加锁处理，锁定的仍然是头节点。并且，支持多个线程同时对数组扩容，提高并发能力。每个线程需先以CAS操作抢任务，争抢一段连续槽位的数据转移权。抢到任务后，该线程会锁定槽内的头节点，然后将链表或树中的数据迁移到新的数组里。
4. 查找数据时并不会加锁，所以性能很好。另外，在扩容的过程中，依然可以支持查找操作。如果某个槽还未进行迁移，则直接可以从旧数组里找到数据。如果某个槽已经迁移完毕，但是整个扩容还没结束，则扩容线程会创建一个转发节点存入旧数组，届时查找线程根据转发节点的提示，从新数组中找到目标数据。

加分回答

​ ConcurrentHashMap实现线程安全的难点在于多线程并发扩容，即当一个线程在插入数据时，若发现数组正在扩容，那么它就会立即参与扩容操作，完成扩容后再插入数据到新数组。在扩容的时候，多个线程共同分担数据迁移任务，每个线程负责的迁移数量是 (数组长度 >>> 3) / CPU核心数。

​ 也就是说，为线程分配的迁移任务，是充分考虑了硬件的处理能力的。多个线程依据硬件的处理能力，平均分摊一部分槽的迁移工作。另外，如果计算出来的迁移数量小于16，则强制将其改为16，这是考虑到目前服务器领域主流的CPU运行速度，每次处理的任务过少，对于CPU的算力也是一种浪费。

延伸阅读

​ 插入数据的关键代码如下：

final V putVal(K key, V value, boolean onlyIfAbsent) {
    ...
    for (Node<K,V>[] tab = table;;) {
        Node<K,V> f; int n, i, fh;
        // 1.若数组还未初始化，则初始化数组。
        if (tab == null || (n = tab.length) == 0)
            tab = initTable();
        // 2.若该位置头节点为空，则初始化头节点。
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
            if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null)))
                break;
        }
        // 3.若头节点哈希值为MOVED，则表示数组正在扩容，则当前线程也去参与扩容。
        else if ((fh = f.hash) == MOVED)
            tab = helpTransfer(tab, f);
        // 4.若该位置头节点非空，则追加一个新节点。
        else {
            V oldVal = null;
            // 加锁，锁的不是整个数组，而是头节点。
            synchronized (f) {
                if (tabAt(tab, i) == f) {
                    // 头节点为链表节点（参考状态常量）
                    if (fh >= 0) {
                        ...
                    }
                    // 头节点为红黑树节点
                    else if (f instanceof TreeBin) {
                        ...
                    }
                }
            }
            // 当链表中节点数达到8时，则扩容数组或者转为红黑树。
            if (binCount != 0) {
                if (binCount >= TREEIFY_THRESHOLD)
                    treeifyBin(tab, i);
                ...
            }
        }
    }
    // 计数，若发现数组过小则进行扩容。
    addCount(1L, binCount);
    return null;
}

​ 数组扩容的关键代码如下：

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
    ...
    // 每个CPU处理的节点数（步长）
    if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
        // 最少设定为16
        stride = MIN_TRANSFER_STRIDE;
    if (nextTab == null) {
        ...
        // 扩容至2倍
         Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
        ...
    }

    // 特殊节点，标识该位置的元素已被转移，并指向新的数组。
    ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);

    // bound为遍历的边界，因为每个线程负责转移一部分节点，不是全部。
    for (int i = 0, bound = 0;;) {
        Node<K,V> f; int fh;
        // advance表示从transferIndex-1遍历到bound位置的过程中，是否需要继续。
        // 自旋
        while (advance) {
            ...
            // 抢任务（CAS修改TRANSFERINDEX）
            else if (U.compareAndSwapInt
                     (this, TRANSFERINDEX, nextIndex,
                      nextBound = (nextIndex > stride ?
                                   nextIndex - stride : 0))) {
                ...
            }
        }
        // i越界，整个集合遍历完成。
        if (i < 0 || i >= n || i + n >= nextn) {
            // 转移完成
            if (finishing) {
                // 替换旧数组
                nextTable = null;
                table = nextTab;
                // 将sizeCtl设置为扩容后的1.5倍
                sizeCtl = (n << 1) - (n >>> 1);
                return;
            }
            ...
        }
        // tab[i]迁移完毕
        else if ((f = tabAt(tab, i)) == null)
            advance = casTabAt(tab, i, null, fwd);
        // tab[i]正在迁移
        else if ((fh = f.hash) == MOVED)
            advance = true;
        else {
            synchronized (f) {
                ...
            }
        }
    }
}