日韩无码专区无码一级三级片|91人人爱网站中日韩无码电影|厨房大战丰满熟妇|AV高清无码在线免费观看|另类AV日韩少妇熟女|中文日本大黄一级黄色片|色情在线视频免费|亚洲成人特黄a片|黄片wwwav色图欧美|欧亚乱色一区二区三区

RELATEED CONSULTING
相關咨詢
選擇下列產(chǎn)品馬上在線溝通
服務時間：8:30-17:00
你可能遇到了下面的問題
關閉右側(cè)工具欄

新聞中心

這里有您想知道的互聯(lián)網(wǎng)營銷解決方案
Java 8 ConcurrentHashMap源碼中竟然隱藏著兩個Bug

 Java 7的ConcurrenHashMap的源碼我建議大家都看看，那個版本的源碼就是Java多線程編程的教科書。在Java 7的源碼中，作者對悲觀鎖的使用非常謹慎，大多都轉(zhuǎn)換為自旋鎖加volatile獲得相同的語義，即使最后迫不得已要用，作者也會通過各種技巧減少鎖的臨界區(qū)。在上一篇文章中我們也有講到，自旋鎖在臨界區(qū)比較小的時候是一個較優(yōu)的選擇是因為它避免了線程由于阻塞而切換上下文，但本質(zhì)上它也是個鎖，在自旋等待期間只有一個線程能進入臨界區(qū)，其他線程只會自旋消耗CPU的時間片。Java 8中ConcurrentHashMap的實現(xiàn)通過一些巧妙的設計和技巧，避開了自旋鎖的局限，提供了更高的并發(fā)性能。如果說Java 7版本的源碼是在教我們?nèi)绾螌⒈^鎖轉(zhuǎn)換為自旋鎖，那么在Java 8中我們甚至可以看到如何將自旋鎖轉(zhuǎn)換為無鎖的方法和技巧。

新區(qū)網(wǎng)站制作公司哪家好，找創(chuàng)新互聯(lián)建站！從網(wǎng)頁設計、網(wǎng)站建設、微信開發(fā)、APP開發(fā)、響應式網(wǎng)站設計等網(wǎng)站項目制作，到程序開發(fā)，運營維護。創(chuàng)新互聯(lián)建站成立于2013年到現(xiàn)在10年的時間，我們擁有了豐富的建站經(jīng)驗和運維經(jīng)驗，來保證我們的工作的順利進行。專注于網(wǎng)站建設就選創(chuàng)新互聯(lián)建站。

把書讀薄

image

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

在開始本文之前，大家首先在心里還是要有這樣的一張圖，如果有同學對HashMap比較熟悉，那這張圖也應該不會陌生。事實上在整體的數(shù)據(jù)結(jié)構(gòu)的設計上Java 8的ConcurrentHashMap和HashMap基本上是一致的。

Java 7中ConcurrentHashMap為了提升性能使用了很多的編程技巧，但是引入Segment的設計還是有很大的改進空間的，Java 7中ConcurrrentHashMap的設計有下面這幾個可以改進的點：

    1.  Segment在擴容的時候非擴容線程對本Segment的寫操作時都要掛起等待的

    2.  對ConcurrentHashMap的讀操作需要做兩次哈希尋址，在讀多寫少的情況下其實是有額外的性能損失的

    3.  盡管size()方法的實現(xiàn)中先嘗試無鎖讀，但是如果在這個過程中有別的線程做寫入操作，那調(diào)用size()的這個線程就會給整個ConcurrentHashMap加鎖，這是整個ConcurrrentHashMap唯一一個全局鎖，這點對底層的組件來說還是有性能隱患的

    4.  極端情況下（比如客戶端實現(xiàn)了一個性能很差的哈希函數(shù)）get()方法的復雜度會退化到O(n)。

針對1和2，在Java 8的設計是廢棄了Segment的使用，將悲觀鎖的粒度降低至桶維度，因此調(diào)用get的時候也不需要再做兩次哈希了。size()的設計是Java 8版本中最大的亮點，我們在后面的文章中會詳細說明。至于紅黑樹，這篇文章仍然不做過多闡述。接下來的篇幅會深挖細節(jié)，把書讀厚，涉及到的模塊有：初始化，put方法, 擴容方法transfer以及size()方法，而其他模塊，比如hash函數(shù)等改變較小，故不再深究。

準備知識

ForwardingNode

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

除了普通的Node和TreeNode之外，ConcurrentHashMap還引入了一個新的數(shù)據(jù)類型ForwardingNode，我們這里只展示他的構(gòu)造方法，F(xiàn)orwardingNode的作用有兩個：

  •  在動態(tài)擴容的過程中標志某個桶已經(jīng)被復制到了新的桶數(shù)組中
  •  如果在動態(tài)擴容的時候有g(shù)et方法的調(diào)用，則ForwardingNode將會把請求轉(zhuǎn)發(fā)到新的桶數(shù)組中，以避免阻塞get方法的調(diào)用，F(xiàn)orwardingNode在構(gòu)造的時候會將擴容后的桶數(shù)組nextTable保存下來。

UNSAFE.compareAndSwap***

這是在Java 8版本的ConcurrentHashMap實現(xiàn)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) // 檢查參數(shù)  
    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數(shù)組的長度，由于在Java 8中已經(jīng)廢棄了Segment，因此concurrencyLevel只是一個保留字段，無實際意義
  •  sizeCtl這個值第一次出現(xiàn)，這個值如果等于-1則表明系統(tǒng)正在初始化，如果是其他負數(shù)則表明系統(tǒng)正在擴容，在擴容時sizeCtl二進制的低十六位等于擴容的線程數(shù)加一，高十六位（除符號位之外）包含桶數(shù)組的大小信息

put方法 

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

put方法將調(diào)用轉(zhuǎn)發(fā)到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節(jié)點，則該線程嘗試加入擴容  
    17.   
  
  
  
  17.         else if ((ffh = f.hash) == MOVED)  
    18.   
  
  
  
  18.             tab = helpTransfer(tab, f);  
    19.   
  
  
  
  19.         // 【D】否則遍歷桶內(nèi)的鏈表或樹，并插入 
    20.   
  
  
  
  20.          else {  
    21.   
  
  
  
  21.             // 暫時折疊起來，后面詳細看  
    22.   
  
  
  
  22.         }  
    23.   
  
  
  
  23.     }  
    24.   
  
  
  
  24.     // 【F】流程走到此處，說明已經(jīng)put成功，map的記錄總數(shù)加一  
    25.   
  
  
  
  25.     addCount(1L, binCount);  
    26.   
  
  
  
  26.     return null;  
    27.   
  
  
  
  27. } 
    28.

從整個代碼結(jié)構(gòu)上來看流程還是比較清楚的，我用括號加字母的方式標注了幾個非常重要的步驟，put方法依然牽扯出很多的知識點

桶數(shù)組的初始化 

 
 
 
 
      
  
  
  
  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.             // 說明已經(jīng)有線程在初始化了，本線程開始自旋  
    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.

在初始化桶數(shù)組的過程中，系統(tǒng)如何保證不會出現(xiàn)并發(fā)問題呢，關鍵點在于自旋鎖的使用，當有多個線程都執(zhí)行initTable方法的時候，CAS可以保證只有一個線程能夠進入到真正的初始化分支，其他線程都是自旋等待。這段代碼中我們關注三點即可：

  •  依照前文所述，當有線程開始初始化桶數(shù)組時，會通過CAS將sizeCtl置為-1，其他線程以此為標志開始自旋等待
  •  當桶數(shù)組初始化結(jié)束后將sizeCtl的值恢復為正數(shù)，其值等于0.75倍的桶數(shù)組長度，這個值的含義和之前HashMap中的THRESHOLD一致，是系統(tǒng)觸發(fā)擴容的臨界點
  •  在finally語句中對sizeCtl的操作并沒有使用CAS是因為CAS保證只有一個線程能夠執(zhí)行到這個地方

添加桶數(shù)組第一個元素 

 
 
 
 
      
  
  
  
  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保證不會有并發(fā)問題，如果CAS失敗，則通過循環(huán)再進入其他分支

判斷是否需要新增線程擴容 

 
 
 
 
      
  
  
  
  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，表明參與擴容的線程數(shù)量+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.

因為入?yún)是一個int類型的值，所有Integer.numberOfLeadingZeros(n)的返回值介于0到32之間，如果轉(zhuǎn)換成二進制

  •  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保證所有線程要基于同一個舊的桶數(shù)組擴容  
    5.   
  
  
  
  5.     transferIndex <= 0已經(jīng)有線程完成擴容任務了 
    6.

至于sc == rs + 1 || sc == rs + MAX_RESIZERS這兩個判斷條件如果是細心的同學一定會覺得難以理解，這個地方確實是JDK的一個BUG，這個BUG已經(jīng)在JDK 12中修復，詳細情況可以參考一下Oracle的官網(wǎng)：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當前擴容的線程數(shù)為0，即已經(jīng)擴容完成了，就不需要再新增線程擴容
  •  sc == (rs << RESIZE_STAMP_SHIFT) + MAX_RESIZERS參與擴容的線程數(shù)已經(jīng)到了最大，就不需要再新增線程擴容

真正擴容的邏輯在transfer方法中，我們后面會詳細看，不過有個小細節(jié)可以提前注意，如果nextTable已經(jīng)初始化了，transfer會返回nextTable的的引用，后續(xù)可以直接操作新的桶數(shù)組。

插入新值

如果桶數(shù)組已經(jīng)初始化好了，該擴容的也擴容了，并且根據(jù)哈希定位到的桶中已經(jīng)有元素了，那流程就跟普通的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的節(jié)點是鏈表，否則是樹節(jié)點或者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的元素個數(shù)+1  
    41.   
  
  
  
  41.     addCount(1L, binCount);  
    42.   
  
  
  
  42.     return null;  
    43.   
  
  
  
  43. } 
    44.

這段代碼中要特備注意一個不起眼的判斷條件（上下文在源碼上已經(jīng)標注出來了）：tabAt(tab, i) == f，這個判斷的目的是為了處理調(diào)用put方法的線程和擴容線程的競爭。因為synchronized是阻塞鎖，如果調(diào)用put方法的線程恰好和擴容線程同時操作同一個桶，且調(diào)用put方法的線程競爭鎖失敗，等到該線程重新獲取到鎖的時候，當前桶中的元素就會變成一個ForwardingNode，那就會出現(xiàn)tabAt(tab, i) != f的情況。

多線程動態(tài)擴容 

 
 
 
 
      
  
  
  
  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) {            // 初始化新的桶數(shù)組  
    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) // 只有最后一個擴容線程才有機會執(zhí)行這個分支  
    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.

在深入到源碼細節(jié)之前我們先根據(jù)下圖看一下在Java 8中ConcurrentHashMap擴容的幾個特點：

  •  新的桶數(shù)組nextTable是原先桶數(shù)組長度的2倍，這與之前HashMap一致
  •  參與擴容的線程也是分段將table中的元素復制到新的桶數(shù)組nextTable中
  •  桶一個桶數(shù)組中的元素在新的桶數(shù)組中均勻的分布在兩個桶中，桶下標相差n(舊的桶數(shù)組的長度)，這一點依然與HashMap保持一致

image-20210424202636495

各個線程之間如何通力協(xié)作

先看一個關鍵的變量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屬性的基礎上，上面的這個循環(huán)就好理解了

 
 
 
 
      
  
  
  
  1. while (advance) {  
    2.   
  
  
  
  2.     int nextIndex, nextBound;  
    3.   
  
  
  
  3.       // 當bound <= i <= transferIndex的時候i自減跳出這個循環(huán)繼續(xù)干活  
    4.   
  
  
  
  4.     if (--i >= bound || finishing)  
    5.   
  
  
  
  5.         advance = false;  
    6.   
  
  
  
  6.     // 擴容的所有任務已經(jīng)被認領完畢，本線程結(jié)束干活  
    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循環(huán)的變量i代表要復制的桶的在桶數(shù)組中的下標，這個值的上限和下限通過游標transferIndex和步長stride計算得來，當i減小為負數(shù)，則說明當前擴容線程完成了擴容任務，這時候流程會走到這個分支：

 
 
 
 
      
  
  
  
  1. // i >= n || i + n >= nextn現(xiàn)在看來取不到  
    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://m.5511xx.com/article/djoppch.html