ConcurrentHashMap(JDK1.8)put分析

一、ConcurrentHashMap整体结构

ConcurrentHashMap的数据结构与HashMap差不多，都是Node数组+红黑树+链表；ConcurrentHashMap中table的节点类型有 3 类：

• Node节点，是链表类型的节点；这类节点hash 大于 0 ；

• 在扩容时ConcurrentHashMap会有一个特殊的标志对象：ForwardingNode；hash值： MOVE(-1)

• 在生成红黑树时，会生成一个TreeBin对象放在table中；hash值：TREEBIN（-2）

因此在put值时，会根据table中node的hash判断节点的类型；

• hash>=0，Node是链表类型的节点；

• hash=-2，是TreeBin,表示该table元素是红黑树的节点；

• hash=-1，是ForwardingNode；容器正在扩容

• hash = -3，是ReservationNode，在调用computeIfAbsent方法时可能会使用的占位对象（本文不讨论computeIfAbsent方法，因此只关注前三种对象）

ConcurrentHashMap和HashMap的put值过程有些类似，ConcurrentHashMap的结构也是table + 链表+ 红黑树；在put值时，锁粒度是table的元素；也就是说，当put值时定位到table的第 i 个元素，那么就会给table[i]上锁；

其他线程在put值时也定位到 i 时就需要等待获取锁；如果是其他位置则不需要等待锁，可以进行put操作。

二、计算hash值

ConcurrentHashMap在put值的时候会计算key的hash值，和HashMap类似；在ConcurrentHashMap计算hash值的是spread方法

• 获取key的hashcode

• 利用hashcode的高16位与低16位做" ^ "位运算，让高位也参与位运算，增加离散型

• 再与HASH_BITS位运算，这是int最大正整数:最高位为0，其余位置为1，用来与hash值做 " & "运算；得到的结果保证最高位为0，其余位置不变；这样就保证得到的hash值是一个非负数；

举个例子：

三、initTable

在put值时有几个判断，与HashMap类似

• 判断table是否已经初始化了；如果是没有初始化，就会先初始化table，再往table中添加值；

• 根据hash值，计算出key-value在table的位置i，判断table[i]是否已经插入了值；如果没有插入值，就将该key-value插入到table[i]中；

• 判断table[i]的hash值是否是MOVE,如果是MOVE表示正在扩容(ForwardingNode的hash值就是MOVE)，需要该线程帮忙将旧容器的值移动到新容器中

• 将key-value插入到链表或者红黑树中

• 判断是否需要扩容

3.1、初始化table

因为是多线程环境，就需要考虑到如果有多个线程同时初始化table的情况；假如现在有三个线程：A、B、C 同时判断到table == null。这个时候三个线程都会同时试图来初始化table，如果一个线程抢先修改了 sizeCtl，

他就可以初始化table，其余线程只需要等待初始化完成即可；

• 1、sizeCtl判断是否已经有线程初始化table，如果sizeCtl=-1，表示已经有线程对table初始化；这个时候会调用yield()让出cpu执行权

• 2、通过CAS修改sizeCtl=-1，初始化table

这段代码，有意思的是进入到初始化table的分支时 try{} finally{} 代码块还要判断 table==null，为什么呢 ？

可能是对应这样一个场景：现在有 A ，B，C三个线程都判断table=null进入到initTable，A，B线程先获取到sizeCtl =0，此时A抢先修改到sizeCtl =-1开始初始化，而B线程修改失败，再次通过循环获取sizeCtl =-1，

调用yield()放弃cpu执行权，等待A线程初始化table；A线程初始化完成之后修改sizeCtl 的值，修改后的sizeCtl >0；

而C线程比A，B线程慢一点，当C线程获取到sizeCtl的值时，A线程已经完成了table的初始化sizeCtl >0，C线程获取到的sc=sizeCtl >0,因此不会进入到休眠状态，会尝试修改sizeCtl 的值，这个时候没有其他线程竞争修改值，

因此会修改成功；又会将sizeCtl 的值修改为 -1 ，【此时：sc=A线程初始化之后sizeCtl 的值；sizeCtl =-1】

进入到try代码块，判断table!=null,不会再重新初始化table，进入finally块，sizeCtl =sc；将sizeCtl 的值还原到A线程初始化时候的值；

添加的table=null的判断保证了只会有一个线程能初始化table；

流程：

3.2、思考

在try代码块中判断table是否已经初始化保证了只有一个线程能初始化table，这个 和 DCL(双重检测枷锁，差不多)；这段代码能修改吗 ？

因为只要sizeCtl != 0，就说明已经有线程在初始化table，其余线程都可以等待带线程初始化完成，这样当线程获初始化table的时就不用再判断table是否为null了，因为只能有一个线程能进入这个分支，对table进行初始化；

这样看起来似乎没什么问题，但是忽略了一个问题，在ConcurrentHashMap的构造对象中有一个构造方法可以指定初始容量，而保存初始容量的变量就是sizeCtl ，也就是说如果指定了初始容量sizeCtl 值就 大于0；

而指定初始容量时并不会初始化一个数组；因此不能修改成：(sc = sizeCtl) != 0，如果修改成这样，那么在指定初始容量之后，所有线程都不能初始化table了；同时也可以看到在初始化时利用了sizeCtl的值：

四、通过hash值定位key-value位置

这个分支其实就是判断table[i] = null，就将key-value值包装程成Node对象，放到table[i]中即可；

• i = (n - 1) & hash；与HashMap一样，定位存储下标，因为使用了 & 运算，因此要求table的容量n一定是2的幂次倍；

• casTabAt（）将hash-key-value包装成node添加到table中，添加成功：返回true，进入分支break，结束put操作；不成功：进入下一次循环，继续put操作直到成功为止；

五、判断ConcurrentHashMap是否正在扩容

在上一个分支，如果table[i] != null,就会在接下来的分支中，首先通过node节点的hash值判断，table是否处于扩容状态，如果是扩容状态: hash == MOVE;

当判定在扩容时，会要求这个线程帮忙完成扩容:

扩容暂时在这里不分析，在后面会讲到；因为扩容是添加node导致，在添加元素之后会判断是否需要扩容；在前面提到过，在扩容时会产生：ForwardingNode；它的hash值就是MOVE(-1)；由于扩容实在太复杂，

并且扩容的原因并不在这里；扩容触发条件是添加node之后size到达扩容阈值触发扩容；因此把扩容的部分放在最后；

六、判断节点的类型，并添加节点；

判断节点类型，并添加值：

• 如果是链表，就将值打包成Node，添加到链表中；

• 如果是红黑树，就将值打包成TreeNode添加到红黑树中

这个过程和 HashMap一样；唯一的区别是，在多线程环境下，在确定table的下标之后，会获取table[i]对象的锁，只能有一个线程在table[i]所在的链表或者红黑树put值；其他线程要在table[i]中put值需要等待获取锁；

//	f = tabAt(tab, i = (n - 1) & hash)else {V oldVal = null;synchronized (f) {//获取table[i]的对象锁if (tabAt(tab, i) == f) {//判断是否被修改if (fh >= 0) {//根据node的hash值判断是链表还是红黑树binCount = 1;for (Node<K,V> e = f;; ++binCount) {K ek;// 已经存在key，更新val，返回旧valueif (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;// key在链表中不存在，将key-value打包成Node插入到链表表尾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;//p !=null ,说明在红黑树中已经存在key，只是更新了value；if (!onlyIfAbsent)p.val = value;}}}}if (binCount != 0) {if (binCount >= TREEIFY_THRESHOLD)treeifyBin(tab, i);if (oldVal != null)return oldVal;break;}}}

为了更好的看清上面代码的整体逻辑，去掉部分代码，留下一个大的框架：

		//当有一个线程 ： A ，锁定table[i]时，其余线程如果也是锁定这个位置，就必须等A线程，put完成之后；//获取到table[i]的锁，才能在这个位置put值；synchronized (f) {//判断节点没有变动；如果变动了，就计入下一次循环；//f发生改变的几种情况：//1:有可能链表变成红黑树，//2:可能容器在扩容；//3:如果其他线程在之前进行了remove操作，导致f被删除，这种情况也不能直接put；//还有一种情况：就是key-value的值被修改，这种情况对下一个线程put值没有影响；//因此可以看到在很多地方，对节点进行操作前，都会先判断节点有没有改变；if (tabAt(tab, i) == f) {if(fh>=0){binCount=1;链表，会将新node添加到链表末尾，在这个过程中binCount会记录链表的长度，用来判断是否需要将链表修改为红黑树；还有一种情况是这个key已经存在，就直接更新value；}else if(f instanceof TreeBin){binCount=2;红黑树，这里的binCount就是只用来表示该线程抢到锁，已经put值了；}}}//说明已经添加了node；binCount=0是因为线程没有put值,//f已经被其他线程删除，或者是正在扩容，或者是由链表改成了红黑树。。if (binCount != 0) {if (binCount >= TREEIFY_THRESHOLD)//这里是用来判断链表需不需要改成红黑树;treeifyBin(tab, i);if (oldVal != null)//key已经存在了，会将value更新；并返回旧值oldValuereturn oldVal;break;}