ConcurrentHashMap源码解析（超级详细版本）

concurrentHashMap概述

concurrentHashMap是HashMap的线程安全版本，是juc包中提供的。利用volatile关键字和cas还有synchronized关键字对hashMap进行改造，使其变为线程安全的。其很多特性和hashMap是一样的，hashMap的源码可以去看我之前的博客：https://www.shouxicto.com/article/129.html

但是与hashMap区别不只是线程安全方面，还有就是concurrentHashMap不允许key和value为空，这点在后面源码解析也可以看到，本文章主要分析JDK1.8及之后的版本，1.7之前的版本会提一点，不会对源码进行详细的解析。

数据结构图解

自我感觉学一个东西首先要了解他的架构图，包括框架，这样我们可以有一个整体的框架，学习起来也会更轻松。接下来我们来看concurrentHashMap的数据结构图，下面两张图片是从别人博客里面摘的（ps：感觉很多博客都是这两张图，就不注明出处了，我：拿来吧你0.0）

1.8：

1.7：

如图，1.8版本之后，concurrentHashMap和hashMap数据结构基本是一致的，唯一的区别就是concurrentHashMap是在操作链表或者红黑树时对node数组中的每一个链表或红黑树加锁，其他时候利用cas来保证线程同步。并且利用volatile关键字，保证了table数组、标索引等的可见性，结合cas，在不加锁的情况下保证线程同步，相较于1.7，还引入了红黑树，极大的优化了速度。

1.7版本之前，concurrentHashMap是利用分段锁，将entry数组分为多个segment数组，然后当我们插入数据时对segment加锁，相较于hashTable的全部加锁速度快了不少，并且能够至此一定的并发操作，但是速度方面还是不够优秀。

源码解析

1. GET方法

get方法比较简单，相对于put方法，只需要我们通过valotile关键字保证可见性就可以了，所以和hashMap差别不大

在读这get源码之前，我们需要知道一个方法，这个方法在put中也被用到，我们需要理解他是干什么的:

staticfinal<K,V>Node<K,V>tabAt(Node<K,V>[]tab,inti){return(Node<K,V>)U.getObjectVolatile(tab,((long)i<<ASHIFT)+ABASE);}

这段代码主要是用来取table数组中的对应下标的元素的，可以看到调用了一个getObjectVolatile()方法,这个方法是Usafe类提供的一个方法，这个类提供了一些原子操作，是通过我们的本地方法实现的，也就是调用了C++代码

然后是spread()方法，可以看到这个和我们hashmap中的hash()方法是一样的，只是和HASH_BITS进行了一次与运算，这个是用来使最高位为0，保证我们的哈希值为正数，代表是一个链表，若为1则为负数，代表一个红黑树，这个后面判断红黑树的时候会用到。

staticfinalintHASH_BITS=0x7fffffff;//usablebitsofnormalnodehashstaticfinalintspread(inth){return(h^(h>>>16))&HASH_BITS;}

然后我们来看get的源码：

publicVget(Objectkey){Node<K,V>[]tab;Node<K,V>e,p;intn,eh;Kek;inth=spread(key.hashCode());if((tab=table)!=null&&(n=tab.length)>0&&(e=tabAt(tab,(n-1)&h))!=null){if((eh=e.hash)==h){if((ek=e.key)==key||(ek!=null&&key.equals(ek)))returne.val;}elseif(eh<0)return(p=e.find(h,key))!=null?p.val:null;while((e=e.next)!=null){if(e.hash==h&&((ek=e.key)==key||(ek!=null&&key.equals(ek))))returne.val;}}returnnull;}

首先我们通过spread()方法计算出传入的key的hash值，如果table不为空并且(n - 1) & h对应位置的节点不为空（n是table数组长度），

那么就先判断对应位置根节点的hash值是否等于我们计算出的key值的hash值，如果相等就直接返回根节点的val值，否则就进行下一步

若根接节点的hash值小于0，若小于0代表为红黑树，那么就采用红黑树的方式向下查找，找到有相同的值就返回，否则返回null。

若接节点的hash值大于等于0，遍历node链表向下找，同样找到就返回值，否则返回null。

到此，get方法结束。

2. PUT方法

put方法相对就比较麻烦了，代码如下：

publicVput(Kkey,Vvalue){returnputVal(key,value,false);}

这里调用了putVal()方法，哪个false参数是代表当put的值重复时要不要替换，false代表要替换。然后我们看putVal()方法：

finalVputVal(Kkey,Vvalue,booleanonlyIfAbsent){//判断key和value是否为空，为空直接抛出空指针异常if(key==null||value==null)thrownewNullPointerException();//计算hash值，和put方法中一致inthash=spread(key.hashCode());//用于记录链表长度intbinCount=0;for(Node<K,V>[]tab=table;;){Node<K,V>f;intn,i,fh;//如果table未初始化，就先初始化tableif(tab==null||(n=tab.length)==0)tab=initTable();//如果要插入的table数组下标位置为空，则直接利用cas插入一个node即可elseif((f=tabAt(tab,i=(n-1)&hash))==null){if(casTabAt(tab,i,null,newNode<K,V>(hash,key,value,null)))break;//nolockwhenaddingtoemptybin}//如果table数组正在扩容，那么就需要当前线程帮助其扩容elseif((fh=f.hash)==MOVED)tab=helpTransfer(tab,f);//如果下标位置有节点，并且没有在扩容else{//记录要被替换的老数据VoldVal=null;//对table数组对应下表位置的根节点加锁synchronized(f){if(tabAt(tab,i)==f){//如果根节点hash大于等于0，代表为链表if(fh>=0){binCount=1;//对链表进行遍历，binCount记录链表长度for(Node<K,V>e=f;;++binCount){Kek;//如果有相同的节点if(e.hash==hash&&((ek=e.key)==key||(ek!=null&&key.equals(ek)))){//将老数据进行保存oldVal=e.val;//根据参数条件，将节点的value值进行替换if(!onlyIfAbsent)e.val=value;break;}Node<K,V>pred=e;//如果没有相同节点，并且遍历到了最后一个节点if((e=e.next)==null){//将节点拼接在链表最后面pred.next=newNode<K,V>(hash,key,value,null);break;}}}//如果根节点是树节点，用树的方式进行插入elseif(finstanceofTreeBin){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){//判断链表长度，如果大于TREEIFY_THRESHOLD，也就是大于8if(binCount>=TREEIFY_THRESHOLD)//尝试进行扩容treeifyBin(tab,i);//如果是替换，那么返回替换前的valueif(oldVal!=null)returnoldVal;break;}}}//添加计数，判断是否需要扩容，如果正在扩容，则帮助扩容，扩容后再次判断是否需要再次扩容。//根据sizeCtl扩容阈值(注意还有一个SIZECTL，这两个并不是同一个东西)//判断是否达到阈值需要进行扩容，需要扩容的话，//判断是否正在扩容，正在扩容则帮助扩容addCount(1L,binCount);returnnull;}

我们来看一下这个addcount()方法：

//从putVal传入的参数是1，binCount，binCount默认是0，只有hash冲突了才会大于1.且他的大小是链表的长度（如果不是红黑数结构的话）。privatefinalvoidaddCount(longx,intcheck){CounterCell[]as;longb,s;//如果计数盒子不是空或者//如果修改baseCount失败if((as=counterCells)!=null||!U.compareAndSwapLong(this,BASECOUNT,b=baseCount,s=b+x)){CounterCella;longv;intm;booleanuncontended=true;//如果计数盒子是空（尚未出现并发）//如果随机取余一个数组位置为空或者//修改这个槽位的变量失败（出现并发了）//执行fullAddCount方法。并结束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();}//如果需要检查,检查是否需要扩容，在putVal方法调用时，默认就是要检查的。if(check>=0){Node<K,V>[]tab,nt;intn,sc;//如果map.size()大于sizeCtl（达到扩容阈值需要扩容）且//table不是空；且table的长度小于1<<30。（可以扩容）while(s>=(long)(sc=sizeCtl)&&(tab=table)!=null&&(n=tab.length)<MAXIMUM_CAPACITY){//根据length得到一个标识intrs=resizeStamp(n);//如果正在扩容if(sc<0){//如果sc的低16位不等于标识符（校验异常sizeCtl变化了）//如果sc==标识符+1（扩容结束了，不再有线程进行扩容）（默认第一个线程设置sc==rs左移16位+2，当第一个线程结束扩容了，就会将sc减一。这个时候，sc就等于rs+1）//如果sc==标识符+65535（帮助线程数已经达到最大）//如果nextTable==null（结束扩容了）//如果transferIndex<=0(转移状态变化了)//结束循环if((sc>>>RESIZE_STAMP_SHIFT)!=rs||sc==rs+1||sc==rs+MAX_RESIZERS||(nt=nextTable)==null||transferIndex<=0)break;//如果可以帮助扩容，那么将sc加1.表示多了一个线程在帮助扩容if(U.compareAndSwapInt(this,SIZECTL,sc,sc+1))//扩容transfer(tab,nt);}//如果不在扩容，将sc更新：标识符左移16位然后+2.也就是变成一个负数。高16位是标识符，低16位初始是2.elseif(U.compareAndSwapInt(this,SIZECTL,sc,(rs<<RESIZE_STAMP_SHIFT)+2))//更新sizeCtl为负数后，开始扩容。transfer(tab,null);s=sumCount();}}}

x 参数表示的此次需要对表中元素的个数加几。check 参数表示是否需要进行扩容检查，大于等于0 需要进行检查，而我们的 putVal 方法的 binCount 参数最小也是 0 ，因此，每次添加元素都会进行检查。（除非是覆盖操作），下面是详细步骤：

判断计数盒子（counterCells）属性是否是空，如果是空，就尝试修改 baseCount 变量，对该变量进行加 X。

如果计数盒子不是空，或者修改 baseCount 变量失败了，则放弃对 baseCount 进行操作。

如果计数盒子是 null 或者计数盒子的 length 是 0，或者随机取一个位置是 null，那么就对刚刚的元素进行 CAS 赋值。

如果赋值失败，或者满足上面的条件，则调用 fullAddCount 方法重新死循环插入。

这里如果操作 baseCount 失败了（或者计数盒子不是 Null），且对计数盒子赋值成功，那么就检查 check 变量，如果该变量小于等于 1. 直接结束。否则，计算一下 count 变量。

如果 check 大于等于 0 ，说明需要对是否扩容进行检查。

如果 map 的 size 大于 sizeCtl（扩容阈值），且 table 的长度小于 1 << 30，那么就进行扩容。

根据 length 得到一个标识符，然后，判断 sizeCtl 状态，如果小于 0 ，说明要么在初始化，要么在扩容。

如果正在扩容，那么就校验一下数据是否变化了（具体可以看上面代码的注释）。如果检验数据不通过，break。

如果校验数据通过了，那么将 sizeCtl 加一，表示多了一个线程帮助扩容。然后进行扩容。

如果没有在扩容，但是需要扩容。那么就将 sizeCtl 更新，赋值为标识符左移 16 位 —— 一个负数。然后加 2。 表示，已经有一个线程开始扩容了。然后进行扩容。然后再次更新 count，看看是否还需要扩容。

到此，我们的put方法结束了。

1.7与1.8的区别

项目1.71.8同步机制分段锁，每个segment继承ReentrantLockCAS + synchronized保证并发更新数据结构数组+链表数组+链表+红黑树节点HashEntryNodeput操作多个线程同时竞争获取同一个segment锁，获取成功的线程更新map；失败的线程尝试多次获取锁仍未成功，则挂起线程，等待释放锁访问相应的bucket时，使用sychronizeded关键字，防止多个线程同时操作同一个bucket，如果该节点的hash不小于0，则遍历链表更新节点或插入新节点；如果该节点是TreeBin类型的节点，说明是红黑树结构，则通过putTreeVal方法往红黑树中插入节点；更新了节点数量，还要考虑扩容和链表转红黑树

总结

concurrentHashMap和hashMap相比，由于考虑了高并发，整体更加复杂，并且也更难以理解。本文只对put和get方法进行了详细的分析，其他方法相对来说比较简单，想要了解的朋友可以自己去查看源码。另外本次源码解析有错误的，欢迎各位指正，谢谢。