ConcurrentHashMap

一 1.7版本的实现方式

• 采用分段锁机制，底层有两个对象，一个是Segment对象，另一个是HashEntry对象
• 一个Segment就是一把锁，它继承自ReentrantLock，每个Segment锁住若干个桶(不一定是一个)。
• 一个HashEntry是一个链表，里面包含Key-Value和next等信息。
• 一个ConcurrentHashMap里包含一个HashEntry数组和一个Segment数组。

二 ConcurrentHashMap的核心字段/内部类

字段/类名	类型	含义	备注
table	Node[]	底层保存数据的数组	初始化时为null
sizeCtl	volatile int	用于控制初始化和扩容，处理并发问题	这个值为长度没有关系
nextTable	Node[]	通常情况下为null，扩容时使用	-
ForwardingNode	内部类	-	在扩容时用，扩容后，需要把table中的数据转移到nextTable中，转移的过程中可能存在并发问题，如果table中某个元素的这个值，说明这个元素正在被迁移。

• sizeCtl的几种值及意义：
• -1：表示正在初始化
• -N: 表示有N-1个线程在扩容
• 其他情况下，如果table未被初始化，则表示初始化的长度，如果已经初始化，则表示table的容量。

三 初始化过程

ConcurrentHashMap的初始化没有初始化table，只是计算了sizeCtl：

• 使用无参构造方法时，sizeCtl = 0，后续table初始化时使用默认容量16
• 使用带有initialCapacity，sizeCtl是一个其乘以0.75表这个initialCapacity大的最小2的整数次幂。

table的初始化放在第一次put的操作中，并且使用sizeCtl的原子操作来控制并发：

    private final Node<K,V>[] initTable() {
        Node<K,V>[] tab; int sc;
        while ((tab = table) == null || tab.length == 0) {
            if ((sc = sizeCtl) < 0)
                Thread.yield(); // lost initialization race; just spin
            else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
              //do real initTable
               break;
            }
        }
        return tab;
    }

四 put操作

4.1 hashCode到table下标的映射方式：

h = key.hashCode();
(h ^ (h >>> 16)) & HASH_BITS; //HASH_BITS=0x7fffffff
index = h ^ (table.length - 1);

4.2 put操作的并发控制

分以下几种情况：

4.2.1 table[index] == null
说明这个节点重来未被插入过元素，直接CAS把这个节点值从null更新到新插入的值：

          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
            }

4.2.2 table[index] instanceof ForwardingNode
说明正在被扩容中，此时会去帮助一起扩容。

扩容完成后会令table = nextTable，然后重新再执行一次插入操作

4.2.3 table[index] != null
也就是说table对应下标的已经有其他元素了，这时需要把新元素加到链尾

如果链表长度>=8，则需要转换成红黑树

如果已经是红黑树，则加入红黑树中

这种情况下的并发控制：直接synchronized队头结点(或者红黑树的根结点)

 synchronized (f) {
                    if (tabAt(tab, i) == f) {
                    }
}

f即为table中对应下标的元素，f是在前面的判断中付值的，这里之所以要再判断if (tabAt(tab, i) == f) ，是因为这个位置的元素可能已经被修改了(例如其他线程扩容了)

取table中的元素并不是采用table[index] 而是用UNSAFE.getObjectVolatile, 虽然table是volatile修饰的，但不能保证线程每次都拿到table中的最新元素，Unsafe.getObjectVolatile可以直接获取指定内存的数据，保证了每次拿到数据都是最新的。

从网上查到的观点：volatile修饰数组时，只能保存数组引用的可见性，不能保存数组中元素的可见性，所以这里table虽然使用了volatile修饰，但是仍然不能保存table[index]可以拿到最新的值。

五扩容操作

参考文章https://www.jianshu.com/p/f6730d5784ad

分为以下几步：

• 第一个触发扩容的线程初始化nextTable数组，长度为原数组的两倍。同时使用CAS sizeCtl的值以确保nextTable只会初始化一次。
• 初始化完成后开始迁移数据，从旧数组的最后一个元素开始迁移。这个过程可以多个线程一起执行。
• 开始迁移之前，要先判断旧数组中的元素，如果为null，则直接跳过，如果为ForwardingNode则说明有其他线程正在迁移。
• 否则说明该位置节点没有迁移，CAS操作把该位置节点设置成ForwardingNode，表明当前线程拿到了这个位置的迁移权限。
• 开始迁移，遍历该链的所有节点，使用节点的hashCode & n 把节点分成两类。把结果为1的和结果为0的分别做成hn和ln两个链表，把ln放在nextTable的旧位置上(原来是第5个桶就还放在第5个桶上)，把hn向后移动n个桶(就是放在nextTable的5+n的位置上)

可以看出来，扩容操作的并发控制分为两点：

• 一是使用sizeCtl的CAS操作保证nextTable只会初始化一次
• 二是使用把旧数组的元素CAS为ForwardingNode的方式来保证一个桶只会被一个线程迁移。

六 如何统计数量

Map有个size()方法会返回当前的数量，在HashMap中直接使用一个size类型记录。但是在ConcurrentHashMap中要考虑高并发的情况，首先肯定不能像HashMap一样用一个普通的int字段统计。

可以使用一个原子类来记录，或者普通int字段加CAS操作，但是这样在高并发的情况下乐观锁的竞争压力太大，性能不行。

ConcurrentHashMap在低并发的情况下就是这种方式，使用一个volatile long型的baseCount的CAS操作来计数。但是在高并发的情况下（如果baseCount的CAS操作出现竞争，就会转用高并发的方式）会有不同的方案。

在高并发时ConcurrentHashMap使用了一组计数器来计数，每个线程第一次操作计数时都会随机选定一个计数器，然后使用这个计数器进行CAS操作。该线程以后再次操作计数时，都会使用同一个计数器（这是为了减少CAS的竞争）。

计数器数组及里面的计数器的初始化都需要做并发控制，确保只初始化一次，ConcurrentHashMap中使用cellsBusy来控制并发