Java 8 ConcurrentHashMap源码中竟然隐藏着两个Bug

 Java 7的ConcurrenHashMap的源码我建议大家都看看，那个版本的源码就是Java多线程编程的教科书。在Java 7的源码中，作者对悲观锁的使用非常谨慎，大多都转换为自旋锁加volatile获得相同的语义，即使最后迫不得已要用，作者也会通过各种技巧减少锁的临界区。在上一篇文章中我们也有讲到，自旋锁在临界区比较小的时候是一个较优的选择是因为它避免了线程由于阻塞而切换上下文，但本质上它也是个锁，在自旋等待期间只有一个线程能进入临界区，其他线程只会自旋消耗CPU的时间片。Java 8中ConcurrentHashMap的实现通过一些巧妙的设计和技巧，避开了自旋锁的局限，提供了更高的并发性能。如果说Java 7版本的源码是在教我们如何将悲观锁转换为自旋锁，那么在Java 8中我们甚至可以看到如何将自旋锁转换为无锁的方法和技巧。

新区网站制作公司哪家好，找创新互联建站！从网页设计、网站建设、微信开发、APP开发、响应式网站设计等网站项目制作，到程序开发，运营维护。创新互联建站成立于2013年到现在10年的时间，我们拥有了丰富的建站经验和运维经验，来保证我们的工作的顺利进行。专注于网站建设就选创新互联建站。

把书读薄

image

图片来源：https://www.zhenchao.org/2019/01/31/java/cas-based-concurrent-hashmap/

在开始本文之前，大家首先在心里还是要有这样的一张图，如果有同学对HashMap比较熟悉，那这张图也应该不会陌生。事实上在整体的数据结构的设计上Java 8的ConcurrentHashMap和HashMap基本上是一致的。

Java 7中ConcurrentHashMap为了提升性能使用了很多的编程技巧，但是引入Segment的设计还是有很大的改进空间的，Java 7中ConcurrrentHashMap的设计有下面这几个可以改进的点：

    1.  Segment在扩容的时候非扩容线程对本Segment的写操作时都要挂起等待的

    2.  对ConcurrentHashMap的读操作需要做两次哈希寻址，在读多写少的情况下其实是有额外的性能损失的

    3.  尽管size()方法的实现中先尝试无锁读，但是如果在这个过程中有别的线程做写入操作，那调用size()的这个线程就会给整个ConcurrentHashMap加锁，这是整个ConcurrrentHashMap唯一一个全局锁，这点对底层的组件来说还是有性能隐患的

    4.  极端情况下（比如客户端实现了一个性能很差的哈希函数）get()方法的复杂度会退化到O(n)。

针对1和2，在Java 8的设计是废弃了Segment的使用，将悲观锁的粒度降低至桶维度，因此调用get的时候也不需要再做两次哈希了。size()的设计是Java 8版本中最大的亮点，我们在后面的文章中会详细说明。至于红黑树，这篇文章仍然不做过多阐述。接下来的篇幅会深挖细节，把书读厚，涉及到的模块有：初始化，put方法, 扩容方法transfer以及size()方法，而其他模块，比如hash函数等改变较小，故不再深究。

准备知识

ForwardingNode

 
 
 
 
      
  
  
  
  1. static final class ForwardingNode extends Node {  
    2.   
  
  
  
  2.     final Node[] nextTable;  
    3.   
  
  
  
  3.     ForwardingNode(Node[] tab) {  
    4.   
  
  
  
  4.         // MOVED = -1，ForwardingNode的哈希值为-1  
    5.   
  
  
  
  5.         super(MOVED, null, null, null);  
    6.   
  
  
  
  6.         this.nextTable = tab;  
    7.   
  
  
  
  7.     }  
    8.   
  
  
  
  8. } 
    9.

除了普通的Node和TreeNode之外，ConcurrentHashMap还引入了一个新的数据类型ForwardingNode，我们这里只展示他的构造方法，ForwardingNode的作用有两个：

  •  在动态扩容的过程中标志某个桶已经被复制到了新的桶数组中
  •  如果在动态扩容的时候有get方法的调用，则ForwardingNode将会把请求转发到新的桶数组中，以避免阻塞get方法的调用，ForwardingNode在构造的时候会将扩容后的桶数组nextTable保存下来。

UNSAFE.compareAndSwap***

这是在Java 8版本的ConcurrentHashMap实现CAS的工具，以int类型为例其方法定义如下：

 
 
 
 
      
  
  
  
  1. /**  
    2.   
  
  
  
  2. * Atomically update Java variable to x if it is currently  
    3.   
  
  
  
  3. * holding expected.  
    4.   
  
  
  
  4. * @return true if successful  
    5.   
  
  
  
  5. */  
    6.   
  
  
  
  6. public final native boolean compareAndSwapInt(Object o, long offset,  
    7.   
  
  
  
  7.                                               int expected,  
    8.   
  
  
  
  8.                                               int x); 
    9.

相应的语义为：

如果对象o起始地址偏移量为offset的值等于expected，则将该值设为x，并返回true表明更新成功，否则返回false，表明CAS失败

初始化 

 
 
 
 
      
  
  
  
  1. public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) {  
    2.   
  
  
  
  2.     if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0) // 检查参数  
    3.   
  
  
  
  3.         throw new IllegalArgumentException();  
    4.   
  
  
  
  4.     if (initialCapacity < concurrencyLevel)  
    5.   
  
  
  
  5.         initialCapacity = concurrencyLevel;  
    6.   
  
  
  
  6.     long size = (long)(1.0 + (long)initialCapacity / loadFactor);  
    7.   
  
  
  
  7.     int cap = (size >= (long)MAXIMUM_CAPACITY) ?  
    8.   
  
  
  
  8.         MAXIMUM_CAPACITY : tableSizeFor((int)size); // tableSizeFor，求不小于size的 2^n的算法，jdk1.8的HashMap中说过  
    9.   
  
  
  
  9.     this.sizeCtl = cap;   
    10.   
  
  
  
  10. } 
    11.

即使是最复杂的一个初始化方法代码也是比较简单的，这里我们只需要注意两个点：

  •  concurrencyLevel在Java 7中是Segment数组的长度，由于在Java 8中已经废弃了Segment，因此concurrencyLevel只是一个保留字段，无实际意义
  •  sizeCtl这个值第一次出现，这个值如果等于-1则表明系统正在初始化，如果是其他负数则表明系统正在扩容，在扩容时sizeCtl二进制的低十六位等于扩容的线程数加一，高十六位（除符号位之外）包含桶数组的大小信息

put方法 

 
 
 
 
      
  
  
  
  1. public V put(K key, V value) {  
    2.   
  
  
  
  2.     return putVal(key, value, false);  
    3.   
  
  
  
  3. } 
    4.

put方法将调用转发到putVal方法：

 
 
 
 
      
  
  
  
  1. final V putVal(K key, V value, boolean onlyIfAbsent) {  
    2.   
  
  
  
  2.     if (key == null || value == null) throw new NullPointerException();  
    3.   
  
  
  
  3.     int hash = spread(key.hashCode());  
    4.   
  
  
  
  4.     int binCount = 0;  
    5.   
  
  
  
  5.     for (Node[] tab = table;;) {  
    6.   
  
  
  
  6.         Node f; int n, i, fh;  
    7.   
  
  
  
  7.         // 【A】延迟初始化  
    8.   
  
  
  
  8.         if (tab == null || (n = tab.length) == 0)  
    9.   
  
  
  
  9.             tab = initTable();  
    10.   
  
  
  
  10.         // 【B】当前桶是空的，直接更新  
    11.   
  
  
  
  11.         else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {  
    12.   
  
  
  
  12.             if (casTabAt(tab, i, null, 
    13.   
  
  
  
  13.                              new Node(hash, key, value, null)))  
    14.   
  
  
  
  14.                 break;                   // no lock when adding to empty bin  
    15.   
  
  
  
  15.         } 
    16.   
  
  
  
  16.         // 【C】如果当前的桶的第一个元素是一个ForwardingNode节点，则该线程尝试加入扩容  
    17.   
  
  
  
  17.         else if ((ffh = f.hash) == MOVED)  
    18.   
  
  
  
  18.             tab = helpTransfer(tab, f);  
    19.   
  
  
  
  19.         // 【D】否则遍历桶内的链表或树，并插入 
    20.   
  
  
  
  20.          else {  
    21.   
  
  
  
  21.             // 暂时折叠起来，后面详细看  
    22.   
  
  
  
  22.         }  
    23.   
  
  
  
  23.     }  
    24.   
  
  
  
  24.     // 【F】流程走到此处，说明已经put成功，map的记录总数加一  
    25.   
  
  
  
  25.     addCount(1L, binCount);  
    26.   
  
  
  
  26.     return null;  
    27.   
  
  
  
  27. } 
    28.

从整个代码结构上来看流程还是比较清楚的，我用括号加字母的方式标注了几个非常重要的步骤，put方法依然牵扯出很多的知识点

桶数组的初始化 

 
 
 
 
      
  
  
  
  1. private final Node[] initTable() {  
    2.   
  
  
  
  2.     Node[] tab; int sc;  
    3.   
  
  
  
  3.     while ((tab = table) == null || tab.length == 0) {  
    4.   
  
  
  
  4.         if ((sc = sizeCtl) < 0)  
    5.   
  
  
  
  5.             // 说明已经有线程在初始化了，本线程开始自旋  
    6.   
  
  
  
  6.             Thread.yield(); // lost initialization race; just spin  
    7.   
  
  
  
  7.         else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {  
    8.   
  
  
  
  8.             // CAS保证只有一个线程能走到这个分支  
    9.   
  
  
  
  9.             try {  
    10.   
  
  
  
  10.                 if ((tab = table) == null || tab.length == 0) {  
    11.   
  
  
  
  11.                     int n = (sc > 0) ? sc : DEFAULT_CAPACITY;  
    12.   
  
  
  
  12.                     @SuppressWarnings("unchecked")  
    13.   
  
  
  
  13.                     Node[] nt = (Node[])new Node[n];  
    14.   
  
  
  
  14.                     tabtable = tab = nt;  
    15.   
  
  
  
  15.                     // sc = n - n/4 = 0.75n  
    16.   
  
  
  
  16.                     sc = n - (n >>> 2);  
    17.   
  
  
  
  17.                 }  
    18.   
  
  
  
  18.             } finally {  
    19.   
  
  
  
  19.                 // 恢复sizeCtl > 0相当于释放锁  
    20.   
  
  
  
  20.                 sizeCtl = sc;  
    21.   
  
  
  
  21.             }  
    22.   
  
  
  
  22.             break;  
    23.   
  
  
  
  23.         }  
    24.   
  
  
  
  24.     }  
    25.   
  
  
  
  25.     return tab;  
    26.   
  
  
  
  26. } 
    27.

在初始化桶数组的过程中，系统如何保证不会出现并发问题呢，关键点在于自旋锁的使用，当有多个线程都执行initTable方法的时候，CAS可以保证只有一个线程能够进入到真正的初始化分支，其他线程都是自旋等待。这段代码中我们关注三点即可：

  •  依照前文所述，当有线程开始初始化桶数组时，会通过CAS将sizeCtl置为-1，其他线程以此为标志开始自旋等待
  •  当桶数组初始化结束后将sizeCtl的值恢复为正数，其值等于0.75倍的桶数组长度，这个值的含义和之前HashMap中的THRESHOLD一致，是系统触发扩容的临界点
  •  在finally语句中对sizeCtl的操作并没有使用CAS是因为CAS保证只有一个线程能够执行到这个地方

添加桶数组第一个元素 

 
 
 
 
      
  
  
  
  1. static final  Node tabAt(Node[] tab, int i) {  
    2.   
  
  
  
  2.     return (Node)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);  
    3.   
  
  
  
  3. }  
    4.   
  
  
  
  4. static final  boolean casTabAt(Node[] tab, int i,  
    5.   
  
  
  
  5.                                     Node c, Node v) {  
    6.   
  
  
  
  6.     return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);  
    7.   
  
  
  
  7. } 
    8.

put方法的第二个分支会用tabAt判断当前桶是否是空的，如果是则会通过CAS写入，tabAt通过UNSAFE接口会拿到桶中的最新元素，casTabAt通过CAS保证不会有并发问题，如果CAS失败，则通过循环再进入其他分支

判断是否需要新增线程扩容 

 
 
 
 
      
  
  
  
  1. final Node[] helpTransfer(Node[] tab, Node f) {  
    2.   
  
  
  
  2.     Node[] nextTab; int sc;  
    3.   
  
  
  
  3.     if (tab != null && (f instanceof ForwardingNode) &&  
    4.   
  
  
  
  4.         (nextTab = ((ForwardingNode)f).nextTable) != null) {  
    5.   
  
  
  
  5.         int rs = resizeStamp(tab.length);  
    6.   
  
  
  
  6.         while (nextTab == nextTable && table == tab &&  
    7.   
  
  
  
  7.                 (sc = sizeCtl) < 0) {  
    8.   
  
  
  
  8.             // RESIZE_STAMP_SHIFT = 16  
    9.   
  
  
  
  9.             if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||  
    10.   
  
  
  
  10.                 sc == rs + MAX_RESIZERS || transferIndex <= 0)  
    11.   
  
  
  
  11.                 break;  
    12.   
  
  
  
  12.             // 这里将sizeCtl的值自增1，表明参与扩容的线程数量+1  
    13.   
  
  
  
  13.             if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {  
    14.   
  
  
  
  14.                 transfer(tab, nextTab);  
    15.   
  
  
  
  15.                 break;  
    16.   
  
  
  
  16.             }  
    17.   
  
  
  
  17.         }  
    18.   
  
  
  
  18.         return nextTab;  
    19.   
  
  
  
  19.     }  
    20.   
  
  
  
  20.     return table;  
    21.   
  
  
  
  21. } 
    22.

在这个地方我们就要详细说下sizeCtl这个标志位了，临时变量rs由resizeStamp这个方法返回

 
 
 
 
      
  
  
  
  1. static final int resizeStamp(int n) {  
    2.   
  
  
  
  2.     // RESIZE_STAMP_BITS = 16  
    3.   
  
  
  
  3.     return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1));  
    4.   
  
  
  
  4. } 
    5.

因为入参n是一个int类型的值，所有Integer.numberOfLeadingZeros(n)的返回值介于0到32之间，如果转换成二进制

  •  Integer.numberOfLeadingZeros(n)的最大值是：00000000 00000000 00000000 00100000
  •  Integer.numberOfLeadingZeros(n)的最小值是：00000000 00000000 00000000 00000000

因此resizeStampd的返回值也就介于00000000 00000000 10000000 00000000到00000000 00000000 10000000 00100000之间，从这个返回值的范围可以看出来resizeStamp的返回值高16位全都是0，是不包含任何信息的。因此在ConcurrrentHashMap中，会把resizeStamp的返回值左移16位拼到sizeCtl中，这就是为什么sizeCtl的高16位包含整个Map大小的原理。有了这个分析，这段代码中比较长的if判断也就能看懂了

 
 
 
 
      
  
  
  
  1. if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||  
    2.   
  
  
  
  2.     sc == rs + MAX_RESIZERS || transferIndex <= 0)  
    3.   
  
  
  
  3.     break;  
    4.   
  
  
  
  4.     (sc >>> RESIZE_STAMP_SHIFT) != rs保证所有线程要基于同一个旧的桶数组扩容  
    5.   
  
  
  
  5.     transferIndex <= 0已经有线程完成扩容任务了 
    6.

至于sc == rs + 1 || sc == rs + MAX_RESIZERS这两个判断条件如果是细心的同学一定会觉得难以理解，这个地方确实是JDK的一个BUG，这个BUG已经在JDK 12中修复，详细情况可以参考一下Oracle的官网：https://bugs.java.com/bugdatabase/view_bug.do?bug_id=JDK-8214427，这两个判断条件应该写成这样：sc == (rs << RESIZE_STAMP_SHIFT) + 1 || sc == (rs << RESIZE_STAMP_SHIFT) + MAX_RESIZERS,因为直接比较rs和sc是没有意义的，必须要有移位操作。它表达的含义是

  •  sc == (rs << RESIZE_STAMP_SHIFT) + 1当前扩容的线程数为0，即已经扩容完成了，就不需要再新增线程扩容
  •  sc == (rs << RESIZE_STAMP_SHIFT) + MAX_RESIZERS参与扩容的线程数已经到了最大，就不需要再新增线程扩容

真正扩容的逻辑在transfer方法中，我们后面会详细看，不过有个小细节可以提前注意，如果nextTable已经初始化了，transfer会返回nextTable的的引用，后续可以直接操作新的桶数组。

插入新值

如果桶数组已经初始化好了，该扩容的也扩容了，并且根据哈希定位到的桶中已经有元素了，那流程就跟普通的HashMap一样了，唯一一点不同的就是，这时候要给当前的桶加锁，且看代码：

 
 
 
 
      
  
  
  
  1. final V putVal(K key, V value, boolean onlyIfAbsent) {  
    2.   
  
  
  
  2.     if (key == null || value == null) throw new NullPointerException();  
    3.   
  
  
  
  3.     int hash = spread(key.hashCode());  
    4.   
  
  
  
  4.     int binCount = 0;  
    5.   
  
  
  
  5.     for (Node[] tab = table;;) {  
    6.   
  
  
  
  6.         Node f; int n, i, fh;  
    7.   
  
  
  
  7.         if (tab == null || (n = tab.length) == 0)// 折叠  
    8.   
  
  
  
  8.         else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {// 折叠}  
    9.   
  
  
  
  9.         else if ((ffh = f.hash) == MOVED)// 折叠  
    10.   
  
  
  
  10.         else {  
    11.   
  
  
  
  11.             V oldVal = null;  
    12.   
  
  
  
  12.             synchronized (f) {  
    13.   
  
  
  
  13.                 // 要注意这里这个不起眼的判断条件  
    14.   
  
  
  
  14.                 if (tabAt(tab, i) == f) {  
    15.   
  
  
  
  15.                     if (fh >= 0) { // fh>=0的节点是链表，否则是树节点或者ForwardingNode  
    16.   
  
  
  
  16.                         binCount = 1;  
    17.   
  
  
  
  17.                         for (Node e = f;; ++binCount) {  
    18.   
  
  
  
  18.                             K ek;  
    19.   
  
  
  
  19.                             if (e.hash == hash &&  
    20.   
  
  
  
  20.                                 ((eek = e.key) == key ||  
    21.   
  
  
  
  21.                                     (ek != null && key.equals(ek)))) {  
    22.   
  
  
  
  22.                                 oldVal = e.val; // 如果链表中有值了，直接更新  
    23.   
  
  
  
  23.                                 if (!onlyIfAbsent)  
    24.   
  
  
  
  24.                                     e.val = value;  
    25.   
  
  
  
  25.                                 break;  
    26.   
  
  
  
  26.                             }  
    27.   
  
  
  
  27.                             Node pred = e;  
    28.   
  
  
  
  28.                             if ((ee = e.next) == null) {  
    29.   
  
  
  
  29.                                 // 如果流程走到这里，则说明链表中还没值，直接连接到链表尾部  
    30.   
  
  
  
  30.                                 pred.next = new Node(hash, key, value, null);  
    31.   
  
  
  
  31.                                 break;  
    32.   
  
  
  
  32.                             }  
    33.   
  
  
  
  33.                         }  
    34.   
  
  
  
  34.                     }  
    35.   
  
  
  
  35.                     // 红黑树的操作先略过  
    36.   
  
  
  
  36.                 }  
    37.   
  
  
  
  37.             }  
    38.   
  
  
  
  38.         }  
    39.   
  
  
  
  39.     }  
    40.   
  
  
  
  40.     // put成功，map的元素个数+1  
    41.   
  
  
  
  41.     addCount(1L, binCount);  
    42.   
  
  
  
  42.     return null;  
    43.   
  
  
  
  43. } 
    44.

这段代码中要特备注意一个不起眼的判断条件（上下文在源码上已经标注出来了）：tabAt(tab, i) == f，这个判断的目的是为了处理调用put方法的线程和扩容线程的竞争。因为synchronized是阻塞锁，如果调用put方法的线程恰好和扩容线程同时操作同一个桶，且调用put方法的线程竞争锁失败，等到该线程重新获取到锁的时候，当前桶中的元素就会变成一个ForwardingNode，那就会出现tabAt(tab, i) != f的情况。

多线程动态扩容 

 
 
 
 
      
  
  
  
  1. private final void transfer(Node[] tab, Node[] nextTab) {  
    2.   
  
  
  
  2.     int n = tab.length, stride;  
    3.   
  
  
  
  3.     if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)  
    4.   
  
  
  
  4.         stride = MIN_TRANSFER_STRIDE; // subdivide range  
    5.   
  
  
  
  5.     if (nextTab == null) {            // 初始化新的桶数组  
    6.   
  
  
  
  6.         try {  
    7.   
  
  
  
  7.             @SuppressWarnings("unchecked")  
    8.   
  
  
  
  8.             Node[] nt = (Node[])new Node[n << 1];  
    9.   
  
  
  
  9.             nextTab = nt;  
    10.   
  
  
  
  10.         } catch (Throwable ex) {      // try to cope with OOME  
    11.   
  
  
  
  11.             sizeCtl = Integer.MAX_VALUE;  
    12.   
  
  
  
  12.             return; 
    13.   
  
  
  
  13.         }  
    14.   
  
  
  
  14.         nextTabnextTable = nextTab;  
    15.   
  
  
  
  15.         transferIndex = n;  
    16.   
  
  
  
  16.     }  
    17.   
  
  
  
  17.     int nextn = nextTab.length;  
    18.   
  
  
  
  18.     ForwardingNode fwd = new ForwardingNode(nextTab);  
    19.   
  
  
  
  19.     boolean advance = true;  
    20.   
  
  
  
  20.     boolean finishing = false; // to ensure sweep before committing nextTab  
    21.   
  
  
  
  21.     for (int i = 0, bound = 0;;) {  
    22.   
  
  
  
  22.         Node f; int fh;  
    23.   
  
  
  
  23.         while (advance) {  
    24.   
  
  
  
  24.             int nextIndex, nextBound; 
    25.   
  
  
  
  25.             if (--i >= bound || finishing)  
    26.   
  
  
  
  26.                 advance = false;  
    27.   
  
  
  
  27.             else if ((nextIndex = transferIndex) <= 0) {  
    28.   
  
  
  
  28.                 i = -1;  
    29.   
  
  
  
  29.                 advance = false;  
    30.   
  
  
  
  30.             }  
    31.   
  
  
  
  31.             else if (U.compareAndSwapInt  
    32.   
  
  
  
  32.                         (this, TRANSFERINDEX, nextIndex,  
    33.   
  
  
  
  33.                         nextBound = (nextIndex > stride ?  
    34.   
  
  
  
  34.                                     nextIndex - stride : 0))) {  
    35.   
  
  
  
  35.                 bound = nextBound;  
    36.   
  
  
  
  36.                 i = nextIndex - 1;  
    37.   
  
  
  
  37.                 advance = false;  
    38.   
  
  
  
  38.             }  
    39.   
  
  
  
  39.         }  
    40.   
  
  
  
  40.         if (i < 0 || i >= n || i + n >= nextn) {  
    41.   
  
  
  
  41.             int sc;  
    42.   
  
  
  
  42.             if (finishing) {  
    43.   
  
  
  
  43.                 nextTable = null;  
    44.   
  
  
  
  44.                 table = nextTab;  
    45.   
  
  
  
  45.                 sizeCtl = (n << 1) - (n >>> 1);  
    46.   
  
  
  
  46.                 return;  
    47.   
  
  
  
  47.             }  
    48.   
  
  
  
  48.             if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {  
    49.   
  
  
  
  49.                 // 判断是会否是最后一个扩容线程  
    50.   
  
  
  
  50.                 if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)  
    51.   
  
  
  
  51.                     return;  
    52.   
  
  
  
  52.                 finishing = advance = true;  
    53.   
  
  
  
  53.                 i = n; // recheck before commit  
    54.   
  
  
  
  54.             }  
    55.   
  
  
  
  55.         }  
    56.   
  
  
  
  56.         else if ((f = tabAt(tab, i)) == null)  
    57.   
  
  
  
  57.             advance = casTabAt(tab, i, null, fwd);  
    58.   
  
  
  
  58.         else if ((ffh = f.hash) == MOVED) // 只有最后一个扩容线程才有机会执行这个分支  
    59.   
  
  
  
  59.             advance = true; // already processed  
    60.   
  
  
  
  60.         else { // 复制过程与HashMap类似，这里不再赘述  
    61.   
  
  
  
  61.             synchronized (f) {  
    62.   
  
  
  
  62.                // 折叠  
    63.   
  
  
  
  63.             }  
    64.   
  
  
  
  64.         }  
    65.   
  
  
  
  65.     }  
    66.   
  
  
  
  66. } 
    67.

在深入到源码细节之前我们先根据下图看一下在Java 8中ConcurrentHashMap扩容的几个特点：

  •  新的桶数组nextTable是原先桶数组长度的2倍，这与之前HashMap一致
  •  参与扩容的线程也是分段将table中的元素复制到新的桶数组nextTable中
  •  桶一个桶数组中的元素在新的桶数组中均匀的分布在两个桶中，桶下标相差n(旧的桶数组的长度)，这一点依然与HashMap保持一致

image-20210424202636495

各个线程之间如何通力协作

先看一个关键的变量transferIndex，这是一个被volatile修饰的变量，这一点可以保证所有线程读到的一定是最新的值。

 
 
 
 
      
  
  
  
  1. private transient volatile int transferIndex; 
    2.

这个值会被第一个参与扩容的线程初始化，因为只有第一个参与扩容的线程才满足条件nextTab == null

 
 
 
 
      
  
  
  
  1. if (nextTab == null) {            // initiating  
    2.   
  
  
  
  2.     try {  
    3.   
  
  
  
  3.         @SuppressWarnings("unchecked")  
    4.   
  
  
  
  4.         Node[] nt = (Node[])new Node[n << 1];  
    5.   
  
  
  
  5.         nextTab = nt;  
    6.   
  
  
  
  6.     } catch (Throwable ex) {      // try to cope with OOME  
    7.   
  
  
  
  7.         sizeCtl = Integer.MAX_VALUE;  
    8.   
  
  
  
  8.         return; 
    9.   
  
  
  
  9.     }  
    10.   
  
  
  
  10.     nextTabnextTable = nextTab;  
    11.   
  
  
  
  11.     transferIndex = n;  
    12.   
  
  
  
  12. } 
    13.

在了解了transferIndex属性的基础上，上面的这个循环就好理解了

 
 
 
 
      
  
  
  
  1. while (advance) {  
    2.   
  
  
  
  2.     int nextIndex, nextBound;  
    3.   
  
  
  
  3.       // 当bound <= i <= transferIndex的时候i自减跳出这个循环继续干活  
    4.   
  
  
  
  4.     if (--i >= bound || finishing)  
    5.   
  
  
  
  5.         advance = false;  
    6.   
  
  
  
  6.     // 扩容的所有任务已经被认领完毕，本线程结束干活  
    7.   
  
  
  
  7.     else if ((nextIndex = transferIndex) <= 0) {  
    8.   
  
  
  
  8.         i = -1;  
    9.   
  
  
  
  9.         advance = false; 
    10.   
  
  
  
  10.      }  
    11.   
  
  
  
  11.     // 否则认领新的一段复制任务，并通过`CAS`更新transferIndex的值  
    12.   
  
  
  
  12.     else if (U.compareAndSwapInt 
    13.   
  
  
  
  13.                  (this, TRANSFERINDEX, nextIndex,  
    14.   
  
  
  
  14.                 nextBound = (nextIndex > stride ?  
    15.   
  
  
  
  15.                             nextIndex - stride : 0))) {  
    16.   
  
  
  
  16.         bound = nextBound;  
    17.   
  
  
  
  17.         i = nextIndex - 1;  
    18.   
  
  
  
  18.         advance = false;  
    19.   
  
  
  
  19.     }  
    20.   
  
  
  
  20. } 
    21.

transferIndex就像是一个游标，每个线程认领一段复制任务的时候都会通过CAS将其更新为transferIndex - stride， CAS可以保证transferIndex可以按照stride这个步长降到0。

最后一个扩容线程需要二次确认？

对于每一个扩容线程，for循环的变量i代表要复制的桶的在桶数组中的下标，这个值的上限和下限通过游标transferIndex和步长stride计算得来，当i减小为负数，则说明当前扩容线程完成了扩容任务，这时候流程会走到这个分支：

 
 
 
 
      
  
  
  
  1. // i >= n || i + n >= nextn现在看来取不到  
    2.   
  
  
  
  2. if (i < 0 || i >= n || i + n >= nextn) {  
    3.   
  
  
  
  3.     int sc;  
    4.   
  
  
  
  4.     if (finishing) { // 【A】完成整个扩容过程  
    5.   
  
  
  
  5.         nextTable = null;  
    6.   
  
  
  
  6.         table = nextTab;  
    7.   
  
  
  
  7.                  
    
                当前名称：Java 8 ConcurrentHashMap源码中竟然隐藏着两个Bug
                
    
                文章路径：http://www.csdahua.cn/qtweb/news18/153068.html
            
    
            
    
                网站建设、网络推广公司-快上网，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等
                                
    
            
    
                广告
                
            
    
            
    
                声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源：
                快上网

成都快上网为您推荐相关内容

软件开发知识

同城分类信息

深圳纱窗安装　　 　网站制作　　 　绵阳服务器托管　　 　成都IDC机房托管　　 　营销型网站建设　　 　高端网站定制　　 　防护网厂　　 　雪茄烟品牌　　 　网站设计　　 　成都木托盘　　 　发光字制作　　 　蓬安网站建设　　 　seo关键词优化　　 　营销推广　　 　广安网站制作公司　　 　叙永网站建设　　 　成都微信公众号开发　　 　做网站　　 　户外广告设计　　 　重庆服务器托管