Java并发进阶系列：深度讨论官方关于jdk1.8ConcurrentHashMap的computeIfAbsent源代码修复逻辑

在文章中《深度解析官方关于jdk1.8的resizeStamp的bug处理过程》，我们讨论关于CHM的核心设计——resizeStam需要修复的处理过程，本文再次基于openJDK的bugs讨论组提出的CHM源代码另外一个会造成死循环的bug，默认读者已经掌握CHM的核心源代码实现，否则无法从本文的讨论中获益。文章前部分先把computeIfAbsent的bug成因分析清楚，再来介绍官网ConcurrentHashMap.computeIfAbsent stuck in an endless loop的讨论过程，这样更容易看懂相关内容。研究openJDK官方公布的相关源码bug有何“收益”：虽然这些bug不是特别严重，修复起来也即几行代码，但如果想要解决这种看似“简单的bug”，要求对CHM设计原理、类、方法实现细节足够熟悉，也就是说，你要具备（至少在这个bug上下文的类、方法范围内）和源代码设计者同等思考视角才能去挖掘bug的本质原因并提出合理的修复建议。换句话说，你研究的不是这个bug本身，而是深入精通整个类的源代码实现，这种高级收益在日常业务开发几乎无法获得。认识computeIfAbsent用法理解computeIfAbsent在一些场合下的用法，有助于帮助切入源代码分析。computeIfAbsent使用场景1：packageconcurrent.demo;importjava.util.concurrent.ConcurrentHashMap;publicclassDemo1{staticintcomputeKeyLength(Stringkey){// 计算key的长度，将其作为该key对应的valuereturnkey.length();}publicstaticvoidmain(String[]args){ConcurrentHashMapString,Integermap=newConcurrentHashMap();map.put("foo",1);map.computeIfAbsent("foobar",key-computeKeyLength(key));System.out.println(map);//输出 {foobar=6, foo=1}}computeIfAbsent字面意思：如果key不在map里面，那么就使用给定的匿名函数（也叫映射函数）将key对应的value“计算出来”。（匿名函数也即lambda语法是jdk1.8语法新特性，这一点不必多说）按这个思路可以有以下解释：因为字符串"foobar"这个key不在map里面，因此把它放入map的同时，对应的value要用给定的函数computeKeyLength计算出来，例如这里调用computeKeyLength计算值为6，因此有key=foobar，value=6，将其放入map中。map.put("bar",10);map.computeIfAbsent("bar",key-computeKeyLength(key));System.out.println(map);// 输出：{bar=10}若key已经在map时，value不会被computeKeyLength(key)的计算值6所覆盖。当然此demo做了一个不优雅的示范：既然可用匿名函数的写法去写逻辑，就没必要基于方法computeKeyLength去封装多一层，最简便写法如下：map.computeIfAbsent("foobar",key-key.length());注意这个key可以作为匿名函数的入参去参与到计算value，也可以不作为匿名函数的入参，如下：map.computeIfAbsent("foobar",key-10);显然foobar=10。computeIfAbsent使用场景2：并发场景下的频率统计：该demo方法其实在并发计数器LongAdder这个类的源码注释里面，Doug Lea已经告诉我们一个经典的场景恰好需要使用computeIfAbsent方法LongAdders can be used with a java.util.concurrent.ConcurrentHashMap to maintain a scalable frequency map (a form of histogram or multiset). For example, to add a count to a ConcurrentHashMapString,LongAdder freqs, initializing if not already present, you can use freqs.computeIfAbsent(k - new LongAdder()).increment();packageconcurrent.demo;importjava.util.concurrent.ConcurrentHashMap;importjava.util.concurrent.atomic.LongAdder;publicclassDemo2{publicstaticvoidmain(String[]args){ConcurrentHashMapString,LongAddermap=newConcurrentHashMap();String[]strings={"foo","bar","foo","foo"};for(Stringkey:strings){map.computeIfAbsent(key,k-newLongAdder()).increment();}System.out.println(map);//输出 {bar=1, foo=3}}}该demo虽然只是用单个线程去执行computeIfAbsent，但逻辑是清晰的：实现对字符串出现次数进行统计关于LongAdder的分析，它内部其实有一个像ConcurrentHashMap的fullAddCount并发计数逻辑，这里不再讨论，有关研究可参考本博客的文章《Java并发进阶系列：LongAdder高并发计数性能分析》computeIfAbsent方法源码解析这部内容要求读者已经掌握jdk1.8的ConcurrentHashMap设计及其关键方法的源代码实现逻辑，否则将难以理解其含义。本节所提的computeIfAbsent是未修复前的版本，这里并不会详细解析computeIfAbsent每一行代码，因为它跟putVal方法逻辑几乎一样，而不同地方可参考以下数字标记的说明：publicVcomputeIfAbsent(Kkey,Function?superK,?extendsVmappingFunction){if(key==null||mappingFunction==null)thrownewNullPointerException()inth=spread(key.hashCode());Vval=null;intbinCount=0;for(NodeK,V[]tab=table;;){//看到这个写法应该很熟悉了：自旋+cas机制，为啥要自旋，因为线程不保证自己一次cas就成功，如果和其他线程竞争失败，则需要重试cas，这就是“自旋+cas机制”的黄金搭配。NodeK,Vf;intn,i,fh;if(tab==null||(n=tab.length)==0)tab=initTable();elseif((f=tabAt(tab,i=(n-1)h))==null){//① 如果key对应的桶位为空，先创建一个保留节点用于接下里的占位逻辑NodeK,Vr=newReservationNodeK,V();// ②当前线程用保留节点占位当然需要借用独占锁对r对象进行加锁synchronized(r){// 在当前桶位放置保留节点用于占位，占位之后就可以给该桶位放入新建的node节点if(casTabAt(tab,i,null,r)){binCount=1;NodeK,Vnode=null;try{/*putVal在桶位为空时的逻辑，可看到非常简单，直接使用cas给当前桶位设置新节点，value是给定的value，不需要通过函数计算出value if (casTabAt(tab, i, null,new NodeK,V(hash, key, value, null))) { break; } */// 对于computeIfAbsent，value是需要用给定的匿名函数计算出的，正如前面场景1的“bar”这个key对应的“value”就是使用computeKeyLength(key)计算处理的值if((val=mappingFunction.apply(key))!=null)node=newNodeK,V(h,key,val,null);}finally{//虽然在②步骤那里已经在桶位i放置了一个ReservationNode用于占位，到了这个步骤才是真正把数据节点node放入桶位i当中setTabAt(tab,i,node);}}}// 显然②步骤一定能成功在桶位i放入node节点（binCount=1），既然已经将key和value放入map，那么任务完成，当前线程退出自旋if(binCount!=0)break;}//③如果key定位到的桶位i恰好是一个ForwardingNode占位节点，那么当前线程要去参与“帮助扩容”的逻辑，这里跟putval一样。elseif((fh=f.hash)==MOVED)tab=helpTransfer(tab,f);else{//代码若执行到这，说明桶位i是一个链表或者一棵红色树booleanadded=false;// 当前线程先给头节点加独占锁，保证当前线程写入节点操作时的独占性synchronized(f){//并发环境，这里当然还要二次检查头节点是不是刚刚加锁前的头节点（也即检查加锁前后的头节点有无被改动过）if(tabAt(tab,i)==f){// ④f节点是链表的情况if(fh=0){binCount=1;for(NodeK,Ve=f;;++binCount){Kek;Vev;// 在链表中遇到相同的key，那么就不做更新value操作，返回旧valueif(e.hash==h((ek=e.key)==key||(ek!=nullkey.equals(ek)))){val=e.val;break;}NodeK,Vpred=e;// 尾插法：来到链表尾部if((e=e.next)==