ConcurrentHashMap的初步使用及場景
CHM的使用
ConcurrentHashMap是J.U.C包里面提供的一個線程安全並且高效的HashMap，所以ConcurrentHashMap在並發編程的場景中使用的頻率比較高，那么這一節課我們就從ConcurrentHashMap的使用上以及源碼層面來分析ConcurrentHashMap到底是如何實現安全性的
api使用
ConcurrentHashMap是Map的派生類，所以api基本和Hashmap是類似，主要就是put、get這些方法，接下來基於ConcurrentHashMap的put和get這兩個方法作為切入點來分析ConcurrentHashMap的源碼實現
ConcurrentHashMap的源碼分析
先要做一個說明，這節課分析的ConcurrentHashMap是基於Jdk1.8的版本。 JDK1.7和Jdk1.8版本的變化 ConcurrentHashMap和HashMap的實現原理是差不多的，但是因為ConcurrentHashMap需要支持並發操作，所以在實現上要比hashmap稍微復雜一些。 在JDK1.7的實現上，ConrruentHashMap由一個個Segment組成，簡單來說，ConcurrentHashMap是一個Segment數組，它通過繼承ReentrantLock來進行加鎖，通過每次鎖住一個segment來保證每個segment內的操作的線程安全性從而實現全局線程安全。整個結構圖如下
當每個操作分布在不同的segment上的時候，默認情況下，理論上可以同時支持16個線程的並發寫入。 相比於1.7版本，它做了兩個改進
-
取消了segment分段設計，直接使用Node數組來保存數據，並且采用Node數組元素作為鎖來實現每一行數據進行加鎖來進一步減少並發沖突的概率
-
將原本數組+單向鏈表的數據結構變更為了數組+單向鏈表+紅黑樹的結構。為什么要引入紅黑樹呢？在正常情況下，key hash之后如果能夠很均勻的分散在數組中，那么table數組中的每個隊列的長度主要為0或者1.但是實際情況下，還是會存在一些隊列長度過長的情況。如果還采用單向列表方式，那么查詢某個節點的時間復雜度就變為O(n); 因此對於隊列長度超過8的列表，JDK1.8采用了紅黑樹的結構，那么查詢的時間復雜度就會降低到O(logN),可以提升查找的性能；
這個結構和JDK1.8版本中的Hashmap的實現結構基本一致，但是為了保證線程安全性，ConcurrentHashMap的實現會稍微復雜一下。接下來我們從源碼層面來了解一下它的原理. 我們基於put和get方法來分析它的實現即可
put方法第一階段
public V put(K key, V value) { return putVal(key, value, false); } /** Implementation for put and putIfAbsent */ final V putVal(K key, V value, boolean onlyIfAbsent) { if (key == null || value == null) throw new NullPointerException(); int hash = spread(key.hashCode()); //計算hash值 int binCount = 0; //用來記錄鏈表的長度 for (Node<K,V>[] tab = table;;) { //這里其實就是自旋操作，當出現線程競爭時不斷自旋 Node<K,V> f; int n, i, fh; if (tab == null || (n = tab.length) == 0)//如果數組為空，則進行數組初始化 tab = initTable(); //初始化數組 //通過hash值對應的數組下標得到第一個節點; 以volatile讀的方式來讀取table數組中的元素，保證每次拿到的數據都是最新的 else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { //如果該下標返回的節點為空，則直接通過cas將新的值封裝成node插入即可；如果cas失敗，說明存在競爭，則進入下一次循環 if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null))) break; // no lock when adding to empty bin } //... ｝
假如在上面這段代碼中存在兩個線程，在不加鎖的情況下：線程A成功執行casTabAt操作后，隨后的線程B可以通過tabAt方法立刻看到table[i]的改變。原因如下：線程A的casTabAt操作，具有volatile讀寫相同的內存語義，根據volatile的happens-before規則：線程A的casTabAt操作，一定對線程B的tabAt操作可見
initTable
數組初始化方法，這個方法比較簡單，就是初始化一個合適大小的數組 sizeCtl這個要單獨說一下，如果沒搞懂這個屬性的意義，可能會被搞暈 這個標志是在Node數組初始化或者擴容的時候的一個控制位標識，負數代表正在進行初始化或者擴容操作 -1 代表正在初始化 -N 代表有N-1有二個線程正在進行擴容操作，這里不是簡單的理解成n個線程，sizeCtl就是-N,這塊后續在講擴容的時候會說明 0標識Node數組還沒有被初始化，正數代表初始化或者下一次擴容的大小
private final Node<K,V>[] initTable() { Node<K,V>[] tab; int sc; while ((tab = table) == null || tab.length == 0) { if ((sc = sizeCtl) < 0)//被其他線程搶占了初始化的操作,則直接讓出自己的CPU時間片 Thread.yield(); // lost initialization race; just spin //通過cas操作，將sizeCtl替換為-1，標識當前線程搶占到了初始化資格 else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { try { if ((tab = table) == null || tab.length == 0) { int n = (sc > 0) ? sc : DEFAULT_CAPACITY;//默認初始容量為16 @SuppressWarnings("unchecked") //初始化數組，長度為16，或者初始化在構造ConcurrentHashMap的時候傳入的長度 Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n]; table = tab = nt;//將這個數組賦值給table sc = n - (n >>> 2); //計算下次擴容的大小，實際就是當前容量的0.75倍，這里使用了右移來計算 } } finally { sizeCtl = sc; //設置sizeCtl為sc, 如果默認是16的話，那么這個時候sc=16*0.75=12 } break; } } return tab; }
tabAt
該方法獲取對象中offset偏移地址對應的對象field的值。實際上這段代碼的含義等價於tab[i], 但是為什么不直接使用tab[i]來計算呢？ getObjectVolatile，一旦看到volatile關鍵字，就表示可見性。因為對volatile寫操作happen-before於volatile讀操作，因此其他線程對table的修改均對get讀取可見； 雖然table數組本身是增加了volatile屬性，但是“volatile的數組只針對數組的引用具有volatile的語義，而不是它的元素”。 所以如果有其他線程對這個數組的元素進行寫操作，那么當前線程來讀的時候不一定能讀到最新的值。 出於性能考慮，Doug Lea直接通過Unsafe類來對table進行操作。
static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) { return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE); }
圖解分析
put方法第二階段
在putVal方法執行完成以后，會通過addCount來增加ConcurrentHashMap中的元素個數，並且還會可能觸發擴容操作。這里會有兩個非常經典的設計
高並發下的擴容 如何保證addCount的數據安全性以及性能 //將當前ConcurrentHashMap的元素數量加1，有可能觸發transfer操作(擴容) addCount(1L, binCount); return null; }
addCount
在putVal最后調用addCount的時候，傳遞了兩個參數，分別是1和binCount(鏈表長度)，看看addCount方法里面做了什么操作 x表示這次需要在表中增加的元素個數，check參數表示是否需要進行擴容檢查，大於等於0都需要進行檢查
private final void addCount(long x, int check) { CounterCell[] as; long b, s; 判斷counterCells是否為空， 1. 如果為空，就通過cas操作嘗試修改baseCount變量，對這個變量進行原子累加操作(做這個操作的意義是：如果在沒有競爭的情況下，仍然采用baseCount來記錄元素個數) 2. 如果cas失敗說明存在競爭，這個時候不能再采用baseCount來累加，而是通過CounterCell來記錄 if ((as = counterCells) != null || !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) { CounterCell a; long v; int m; boolean uncontended = true;//是否沖突標識，默認為沒有沖突 這里有幾個判斷 1. 計數表為空則直接調用fullAddCount 2. 從計數表中隨機取出一個數組的位置為空，直接調用fullAddCount 3. 通過CAS修改CounterCell隨機位置的值，如果修改失敗說明出現並發情況（這里又用到了一種巧妙的方法），調用fullAndCount Random在線程並發的時候會有性能問題以及可能會產生相同的隨機數,ThreadLocalRandom.getProbe可以解決這個問題，並且性能要比Random高 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);//執行fullAddCount方法 return; } if (check <= 1)//鏈表長度小於等於1，不需要考慮擴容 return; s = sumCount();//統計ConcurrentHashMap元素個數 } //…. }
CounterCells解釋
ConcurrentHashMap是采用CounterCell數組來記錄元素個數的，像一般的集合記錄集合大小，直接定義一個size的成員變量即可，當出現改變的時候只要更新這個變量就行。為什么ConcurrentHashMap要用這種形式來處理呢？ 問題還是處在並發上，ConcurrentHashMap是並發集合，如果用一個成員變量來統計元素個數的話，為了保證並發情況下共享變量的的難全興，勢必會需要通過加鎖或者自旋來實現，如果競爭比較激烈的情況下，size的設置上會出現比較大的沖突反而影響了性能，所以在ConcurrentHashMap采用了分片的方法來記錄大小，具體什么意思，我們來分析下
private transient volatile int cellsBusy;// 標識當前cell數組是否在初始化或擴容中的CAS標志位 /** * Table of counter cells. When non-null, size is a power of 2. */ private transient volatile CounterCell[] counterCells;// counterCells數組，總數值的分值分別存在每個cell中 @sun.misc.Contended static final class CounterCell { volatile long value; CounterCell(long x) { value = x; } } //看到這段代碼就能夠明白了，CounterCell數組的每個元素，都存儲一個元素個數，而實際我們調用size方法就是通過這個循環累加來得到的 //又是一個設計精華，大家可以借鑒； 有了這個前提，再會過去看addCount這個方法，就容易理解一些了 final long sumCount() { CounterCell[] as = counterCells; CounterCell a; long sum = baseCount; if (as != null) { for (int i = 0; i < as.length; ++i) { if ((a = as[i]) != null) sum += a.value; } } return sum; }
fullAddCount源碼分析
fullAddCount主要是用來初始化CounterCell，來記錄元素個數，里面包含擴容，初始化等操作
private final void fullAddCount(long x, boolean wasUncontended) { int h; //獲取當前線程的probe的值，如果值為0，則初始化當前線程的probe的值,probe就是隨機數 if ((h = ThreadLocalRandom.getProbe()) == 0) { ThreadLocalRandom.localInit(); // force initialization h = ThreadLocalRandom.getProbe(); wasUncontended = true; // 由於重新生成了probe，未沖突標志位設置為true } boolean collide = false; // True if last slot nonempty for (;;) {//自旋 CounterCell[] as; CounterCell a; int n; long v; //說明counterCells已經被初始化過了，我們先跳過這個代碼，先看初始化部分 if ((as = counterCells) != null && (n = as.length) > 0) { if ((a = as[(n - 1) & h]) == null) {// 通過該值與當前線程probe求與，獲得cells的下標元素，和hash 表獲取索引是一樣的 if (cellsBusy == 0) { //cellsBusy=0表示counterCells不在初始化或者擴容狀態下 CounterCell r = new CounterCell(x); //構造一個CounterCell的值，傳入元素個數 if (cellsBusy == 0 &&//通過cas設置cellsBusy標識，防止其他線程來對counterCells並發處理 U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) { boolean created = false; try { // Recheck under lock CounterCell[] rs; int m, j; //將初始化的r對象的元素個數放在對應下標的位置 if ((rs = counterCells) != null && (m = rs.length) > 0 && rs[j = (m - 1) & h] == null) { rs[j] = r; created = true; } } finally {//恢復標志位 cellsBusy = 0; } if (created)//創建成功，退出循環 break; continue;//說明指定cells下標位置的數據不為空，則進行下一次循環 } } collide = false; } //說明在addCount方法中cas失敗了，並且獲取probe的值不為空 else if (!wasUncontended) // CAS already known to fail wasUncontended = true; //設置為未沖突標識，進入下一次自旋 //由於指定下標位置的cell值不為空，則直接通過cas進行原子累加，如果成功，則直接退出 else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))// break; //如果已經有其他線程建立了新的counterCells或者CounterCells大於CPU核心數（很巧妙，線程的並發數不會超過cpu核心數） else if (counterCells != as || n >= NCPU) collide = false; //設置當前線程的循環失敗不進行擴容 else if (!collide)//恢復collide狀態，標識下次循環會進行擴容 collide = true; //進入這個步驟，說明CounterCell數組容量不夠，線程競爭較大，所以先設置一個標識表示為正在擴容 else if (cellsBusy == 0 && U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) { try { if (counterCells == as) {// Expand table unless stale //擴容一倍 2變成4，這個擴容比較簡單 CounterCell[] rs = new CounterCell[n << 1]; for (int i = 0; i < n; ++i) rs[i] = as[i]; counterCells = rs; } } finally { cellsBusy = 0;//恢復標識 } collide = false; continue;//繼續下一次自旋 } h = ThreadLocalRandom.advanceProbe(h);//更新隨機數的值 }
初始化CounterCells數組
//cellsBusy=0表示沒有在做初始化，通過cas更新cellsbusy的值標注當前線程正在做初始化操作 else if (cellsBusy == 0 && counterCells == as && U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) { boolean init = false; try { // Initialize table if (counterCells == as) { CounterCell[] rs = new CounterCell[2]; //初始化容量為2 rs[h & 1] = new CounterCell(x);//將x也就是元素的個數放在指定的數組下標位置 counterCells = rs;//賦值給counterCells init = true;//設置初始化完成標識 } } finally { cellsBusy = 0;//恢復標識 } if (init) break; } //競爭激烈，其它線程占據cell 數組，直接累加在base變量中 else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x)) break; // Fall back on using base } }
CounterCells初始化圖解
初始化長度為2的數組，然后隨機得到指定的一個數組下標，將需要新增的值加入到對應下標位置處
transfer擴容階段
判斷是否需要擴容，也就是當更新后的鍵值對總數baseCount >= 閾值sizeCtl時，進行rehash，這里面會有兩個邏輯。
如果當前正在處於擴容階段，則當前線程會加入並且協助擴容 如果當前沒有在擴容，則直接觸發擴容操作 if (check >= 0) {//如果binCount>=0，標識需要檢查擴容 Node<K,V>[] tab, nt; int n, sc; //s標識集合大小，如果集合大小大於或等於擴容閾值（默認值的0.75） //並且table不為空並且table的長度小於最大容量 while (s >= (long)(sc = sizeCtl) && (tab = table) != null && (n = tab.length) < MAXIMUM_CAPACITY) { int rs = resizeStamp(n);//這里是生成一個唯一的擴容戳，這個是干嘛用的呢？且聽我慢慢分析 if (sc < 0) {//sc<0，也就是sizeCtl<0，說明已經有別的線程正在擴容了 //這5個條件只要有一個條件為true，說明當前線程不能幫助進行此次的擴容，直接跳出循環 //sc >>> RESIZE_STAMP_SHIFT!=rs 表示比較高RESIZE_STAMP_BITS位生成戳和rs是否相等，相同 //sc=rs+1 表示擴容結束 //sc==rs+MAX_RESIZERS 表示幫助線程線程已經達到最大值了 //nt=nextTable -> 表示擴容已經結束 //transferIndex<=0 表示所有的transfer任務都被領取完了，沒有剩余的hash桶來給自己自己好這個線程來做transfer if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || sc == rs + MAX_RESIZERS || (nt = nextTable) == null || transferIndex <= 0) break; if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))//當前線程嘗試幫助此次擴容，如果成功，則調用transfer transfer(tab, nt); } // 如果當前沒有在擴容，那么rs肯定是一個正數，通過rs<<RESIZE_STAMP_SHIFT 將sc設置為一個負數，+2 表示有一個線程在執行擴容 else if (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2)) transfer(tab, null); s = sumCount();// 重新計數，判斷是否需要開啟下一輪擴容 } }
resizeStamp
這塊邏輯要理解起來，也有一點復雜。 resizeStamp用來生成一個和擴容有關的擴容戳，具體有什么作用呢？我們基於它的實現來做一個分析
-
static final int resizeStamp(int n) { -
return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1)); -
}
-
Integer.numberOfLeadingZeros 這個方法是返回無符號整數n最高位非0位前面的0的個數 -
比如10的二進制是 0000 0000 0000 0000 0000 0000 0000 1010 -
那么這個方法返回的值就是28 -
根據resizeStamp的運算邏輯，我們來推演一下，假如n=16，那么resizeStamp(16)=32796轉化為二進制是 -
[0000 0000 0000 0000 1000 0000 0001 1100] -
接着再來看,當第一個線程嘗試進行擴容的時候，會執行下面這段代碼 -
U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2) -
rs左移16位，相當於原本的二進制低位變成了高位1000 0000 0001 1100 0000 0000 0000 0000 -
然后再+2 =1000 0000 0001 1100 0000 0000 0000 0000+10=1000 0000 0001 1100 0000 0000 0000 0010 -
高16位代表擴容的標記、低16位代表並行擴容的線程數 -
-
這樣來存儲有什么好處呢？ -
1. 首先在CHM中是支持並發擴容的，也就是說如果當前的數組需要進行擴容操作，可以由多個線程來共同負責，這塊后續會單獨講 -
2. 可以保證每次擴容都生成唯一的生成戳，每次新的擴容，都有一個不同的n，這個生成戳就是根據n來計算出來的一個數字，n不同，這個數字也不同 -
第一個線程嘗試擴容的時候，為什么是+2 -
因為1表示初始化，2表示一個線程在執行擴容，而且對sizeCtl的操作都是基於位運算的，所以不會關心它本身的數值是多少，只關心它在二進制上的數值，而sc + 1會在 -
低16位上加1。
transfer
擴容是ConcurrentHashMap的精華之一，擴容操作的核心在於數據的轉移，在單線程環境下數據的轉移很簡單，無非就是把舊數組中的數據遷移到新的數組。但是這在多線程環境下，在擴容的時候其他線程也可能正在添加元素，這時又觸發了擴容怎么辦？可能大家想到的第一個解決方案是加互斥鎖，把轉移過程鎖住，雖然是可行的解決方案，但是會帶來較大的性能開銷。因為互斥鎖會導致所有訪問臨界區的線程陷入到阻塞狀態，持有鎖的線程耗時越長，其他競爭線程就會一直被阻塞，導致吞吐量較低。而且還可能導致死鎖。 而ConcurrentHashMap並沒有直接加鎖，而是采用CAS實現無鎖的並發同步策略，最精華的部分是它可以利用多線程來進行協同擴容 簡單來說，它把Node數組當作多個線程之間共享的任務隊列，然后通過維護一個指針來划分每個線程鎖負責的區間，每個線程通過區間逆向遍歷來實現擴容，一個已經遷移完的bucket會被替換為一個ForwardingNode節點，標記當前bucket已經被其他線程遷移完了。接下來分析一下它的源碼實現 1、fwd:這個類是個標識類，用於指向新表用的，其他線程遇到這個類會主動跳過這個類，因為這個類要么就是擴容遷移正在進行，要么就是已經完成擴容遷移，也就是這個類要保證線程安全，再進行操作。 2、advance:這個變量是用於提示代碼是否進行推進處理，也就是當前桶處理完，處理下一個桶的標識 3、finishing:這個變量用於提示擴容是否結束用的
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) { int n = tab.length, stride; //將 (n>>>3相當於 n/8) 然后除以 CPU核心數。如果得到的結果小於 16，那么就使用 16 // 這里的目的是讓每個 CPU 處理的桶一樣多，避免出現轉移任務不均勻的現象，如果桶較少的話，默認一個 CPU（一個線程）處理 16 個桶，也就是長度為16的時候，擴容的時候只會有一個線程來擴容 if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE) stride = MIN_TRANSFER_STRIDE; // subdivide range //nextTab未初始化，nextTab是用來擴容的node數組 if (nextTab == null) { // initiating try { @SuppressWarnings("unchecked") //新建一個n<<1原始table大小的nextTab,也就是32 Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1]; nextTab = nt;//賦值給nextTab } catch (Throwable ex) { // try to cope with OOME sizeCtl = Integer.MAX_VALUE; //擴容失敗，sizeCtl使用int的最大值 return; } nextTable = nextTab; //更新成員變量 transferIndex = n;//更新轉移下標，表示轉移時的下標 } int nextn = nextTab.length;//新的tab的長度 // 創建一個 fwd 節點，表示一個正在被遷移的Node，並且它的hash值為-1(MOVED)，也就是前面我們在講putval方法的時候，會有一個判斷MOVED的邏輯。它的作用是用來占位，表示原數組中位置i處的節點完成遷移以后，就會在i位置設置一個fwd來告訴其他線程這個位置已經處理過了，具體后續還會在講 ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab); // 首次推進為 true，如果等於 true，說明需要再次推進一個下標（i--），反之，如果是 false，那么就不能推進下標，需要將當前的下標處理完畢才能繼續推進 boolean advance = true; //判斷是否已經擴容完成，完成就return，退出循環 boolean finishing = false; // to ensure sweep before committing nextTab 通過for自循環處理每個槽位中的鏈表元素，默認advace為真，通過CAS設置transferIndex屬性值，並初始化i和bound值，i指當前處理的槽位序號，bound指需要處理的槽位邊界，先處理槽位15的節點； for (int i = 0, bound = 0;;) { // 這個循環使用CAS不斷嘗試為當前線程分配任務 // 直到分配成功或任務隊列已經被全部分配完畢 // 如果當前線程已經被分配過bucket區域 // 那么會通過--i指向下一個待處理bucket然后退出該循環 Node<K,V> f; int fh; while (advance) { int nextIndex, nextBound; //--i表示下一個待處理的bucket，如果它>=bound,表示當前線程已經分配過bucket區域 if (--i >= bound || finishing) advance = false; else if ((nextIndex = transferIndex) <= 0) {//表示所有bucket已經被分配完畢 i = -1; advance = false; } //通過cas來修改TRANSFERINDEX,為當前線程分配任務，處理的節點區間為(nextBound,nextIndex)->(0,15) else if (U.compareAndSwapInt (this, TRANSFERINDEX, nextIndex, nextBound = (nextIndex > stride ? nextIndex - stride : 0))) { bound = nextBound;//0 i = nextIndex - 1;//15 advance = false; } } //i<0說明已經遍歷完舊的數組，也就是當前線程已經處理完所有負責的bucket if (i < 0 || i >= n || i + n >= nextn) { int sc; if (finishing) {//如果完成了擴容 nextTable = null;//刪除成員變量 table = nextTab;//更新table數組 sizeCtl = (n << 1) - (n >>> 1);//更新閾值(32*0.75=24) return; } // sizeCtl 在遷移前會設置為 (rs << RESIZE_STAMP_SHIFT) + 2 (詳細介紹點擊這里) // 然后，每增加一個線程參與遷移就會將 sizeCtl 加 1， // 這里使用 CAS 操作對 sizeCtl 的低16位進行減 1，代表做完了屬於自己的任務 if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) { 第一個擴容的線程，執行transfer方法之前，會設置 sizeCtl = (resizeStamp(n) << RESIZE_STAMP_SHIFT) + 2) 后續幫其擴容的線程，執行transfer方法之前，會設置 sizeCtl = sizeCtl+1 每一個退出transfer的方法的線程，退出之前，會設置 sizeCtl = sizeCtl-1 那么最后一個線程退出時：必然有 sc == (resizeStamp(n) << RESIZE_STAMP_SHIFT) + 2)，即 (sc - 2) == resizeStamp(n) << RESIZE_STAMP_SHIFT // 如果 sc - 2 不等於標識符左移 16 位。如果他們相等了，說明沒有線程在幫助他們擴容了。也就是說，擴容結束了。 if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT) return; // 如果相等，擴容結束了，更新 finising 變量 finishing = advance = true; // 再次循環檢查一下整張表 i = n; // recheck before commit } } // 如果位置 i 處是空的，沒有任何節點，那么放入剛剛初始化的 ForwardingNode ”空節點“ else if ((f = tabAt(tab, i)) == null) advance = casTabAt(tab, i, null, fwd); //表示該位置已經完成了遷移，也就是如果線程A已經處理過這個節點，那么線程B處理這個節點時，hash值一定為MOVED else if ((fh = f.hash) == MOVED) advance = true; // already processed } }
擴容過程圖解
ConcurrentHashMap支持並發擴容，實現方式是，把Node數組進行拆分，讓每個線程處理自己的區域，假設table數組總長度是64，默認情況下，那么每個線程可以分到16個bucket。 然后每個線程處理的范圍，按照倒序來做遷移 通過for自循環處理每個槽位中的鏈表元素，默認advace為真，通過CAS設置transferIndex屬性值，並初始化i和bound值，i指當前處理的槽位序號，bound指需要處理的槽位邊界，先處理槽位31的節點； （bound,i） =(16,31) 從31的位置往前推動。
假設這個時候ThreadA在進行transfer，那么邏輯圖表示如下
在當前假設條件下，槽位15中沒有節點，則通過CAS插入在第二步中初始化的ForwardingNode節點，用於告訴其它線程該槽位已經處理過了；
sizeCtl擴容退出機制
在擴容操作transfer的第2414行，代碼如下
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) { if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT) 每存在一個線程執行完擴容操作，就通過cas執行sc-1。 接着判斷(sc-2) !=resizeStamp(n) << RESIZE_STAMP_SHIFT ; 如果相等，表示當前為整個擴容操作的 最后一個線程，那么意味着整個擴容操作就結束了；如果不想等，說明還得繼續 這么做的目的，一方面是防止不同擴容之間出現相同的sizeCtl，另外一方面，還可以避免sizeCtl的ABA問題導致的擴容重疊的情況 數據遷移階段的實現分析 通過分配好遷移的區間之后，開始對數據進行遷移。在看這段代碼之前，先來了解一下原理 synchronized (f) {//對數組該節點位置加鎖，開始處理數組該位置的遷移工作 if (tabAt(tab, i) == f) {//再做一次校驗 Node<K,V> ln, hn;//ln表示低位， hn表示高位;接下來這段代碼的作用是把鏈表拆分成兩部分，0在低位，1在高位 if (fh >= 0) {//下面部分代碼原理點擊這里 int runBit = fh & n; Node<K,V> lastRun = f; //遍歷當前bucket的鏈表，目的是盡量重用Node鏈表尾部的一部分 for (Node<K,V> p = f.next; p != null; p = p.next) { int b = p.hash & n; if (b != runBit) { runBit = b; lastRun = p; } } if (runBit == 0) {如果最后更新的runBit是0，設置低位節點 ln = lastRun; hn = null; } else {//否則，設置高位節點 hn = lastRun; ln = null; } //構造高位以及低位的鏈表 for (Node<K,V> p = f; p != lastRun; p = p.next) { int ph = p.hash; K pk = p.key; V pv = p.val; if ((ph & n) == 0) ln = new Node<K,V>(ph, pk, pv, ln); else hn = new Node<K,V>(ph, pk, pv, hn); } setTabAt(nextTab, i, ln);//將低位的鏈表放在i位置也就是不動 setTabAt(nextTab, i + n, hn);//將高位鏈表放在i+n位置 setTabAt(tab, i, fwd); // 把舊table的hash桶中放置轉發節點，表明此hash桶已經被處理 advance = true; } //紅黑樹的擴容部分暫時忽略 }
高低位原理分析
ConcurrentHashMap在做鏈表遷移時，會用高低位來實現，這里有兩個問題要分析一下
如何實現高低位鏈表的區分 假如我們有這樣一個隊列
第14個槽位插入新節點之后，鏈表元素個數已經達到了8，且數組長度為16，優先通過擴容來緩解鏈表過長的問題，擴容這塊的圖解稍后再分析，先分析高低位擴容的原理 假如當前線程正在處理槽位為14的節點，它是一個鏈表結構，在代碼中，首先定義兩個變量節點ln和hn，實際就是lowNode和HighNode，分別保存hash值的第x位為0和不等於0的節點 通過fn&n可以把這個鏈表中的元素分為兩類，A類是hash值的第X位為0，B類是hash值的第x位為不等於0（至於為什么要這么區分，稍后分析），並且通過lastRun記錄最后要處理的節點。最終要達到的目的是，A類的鏈表保持位置不動，B類的鏈表為14+16(擴容增加的長度)=30 我們把14槽位的鏈表單獨伶出來，我們用藍色表示 fn&n=0的節點，假如鏈表的分類是這樣
for (Node<K,V> p = f.next; p != null; p = p.next) { int b = p.hash & n; if (b != runBit) { runBit = b; lastRun = p; } } 通過上面這段代碼遍歷，會記錄runBit以及lastRun，按照上面這個結構，那么runBit應該是藍色節點，lastRun應該是第6個節點 接着，再通過這段代碼進行遍歷，生成ln鏈以及hn鏈 for (Node<K,V> p = f; p != lastRun; p = p.next) { int ph = p.hash; K pk = p.key; V pv = p.val; if ((ph & n) == 0) ln = new Node<K,V>(ph, pk, pv, ln); else hn = new Node<K,V>(ph, pk, pv, hn); }
接着，通過CAS操作，把hn鏈放在i+n也就是14+16的位置，ln鏈保持原來的位置不動。並且設置當前節點為fwd，表示已經被當前線程遷移完了
-
setTabAt(nextTab, i, ln); -
setTabAt(nextTab, i + n, hn); -
setTabAt(tab, i, fwd);
遷移完成以后的數據分布如下
為什么要做高低位的划分
要想了解這么設計的目的，我們需要從ConcurrentHashMap的根據下標獲取對象的算法來看，在putVal方法中1018行
-
(f = tabAt(tab, i = (n - 1) & hash)) == null
-
通過(n-1) & hash來獲得在table中的數組下標來獲取節點數據，【&運算是二進制運算符，1 & 1=1，其他都為0】 -
-
假設我們的table長度是16， 二進制是【0001 0000】，減一以后的二進制是 【0000 1111】 -
假如某個key的hash值=9，對應的二進制是【0000 1001】，那么按照（n-1） & hash的算法 -
0000 1111 & 0000 1001 =0000 1001 ， 運算結果是9 -
當我們擴容以后，16變成了32，那么(n-1)的二進制是 【0001 1111】 -
仍然以hash值=9的二進制計算為例 -
0001 1111 & 0000 1001 =0000 1001 ，運算結果仍然是9 -
-
我們換一個數字，假如某個key的hash值是20，對應的二進制是【0001 0100】，仍然按照(n-1) & hash算法，分別在16為長度和32位長度下的計算結果 -
16位： 0000 1111 & 0001 0100=0000 0100 -
32位： 0001 1111 & 0001 0100 =0001 0100 -
從結果來看，同樣一個hash值，在擴容前和擴容之后，得到的下標位置是不一樣的，這種情況當然是不允許出現的，所以在擴容的時候就需要考慮， -
而使用高低位的遷移方式，就是解決這個問題. -
大家可以看到，16位的結果到32位的結果，正好增加了16. -
比如 20 & 15=4 、20 & 31=20 ； 4-20 =16 -
比如 60 & 15=12 、60 & 31=28； 12-28=16 -
所以對於高位，直接增加擴容的長度，當下次hash獲取數組位置的時候，可以直接定位到對應的位置。 -
這個地方又是一個很巧妙的設計，直接通過高低位分類以后，就使得不需要在每次擴容的時候來重新計算hash，極大提升了效率。
擴容結束以后的退出機制
如果線程擴容結束，那么需要退出，就會執行transfer方法的如下代碼
//i<0說明已經遍歷完舊的數組，也就是當前線程已經處理完所有負責的bucket if (i < 0 || i >= n || i + n >= nextn) { int sc; if (finishing) {//如果完成了擴容 nextTable = null;//刪除成員變量 table = nextTab;//更新table數組 sizeCtl = (n << 1) - (n >>> 1);//更新閾值(32*0.75=24) return; } // sizeCtl 在遷移前會設置為 (rs << RESIZE_STAMP_SHIFT) + 2 // 然后，每增加一個線程參與遷移就會將 sizeCtl 加 1， // 這里使用 CAS 操作對 sizeCtl 的低16位進行減 1，代表做完了屬於自己的任務 if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) { 第一個擴容的線程，執行transfer方法之前，會設置 sizeCtl = (resizeStamp(n) << RESIZE_STAMP_SHIFT) + 2) 后續幫其擴容的線程，執行transfer方法之前，會設置 sizeCtl = sizeCtl+1 每一個退出transfer的方法的線程，退出之前，會設置 sizeCtl = sizeCtl-1 那么最后一個線程退出時：必然有 sc == (resizeStamp(n) << RESIZE_STAMP_SHIFT) + 2)，即 (sc - 2) == resizeStamp(n) << RESIZE_STAMP_SHIFT // 如果 sc - 2 不等於標識符左移 16 位。如果他們相等了，說明沒有線程在幫助他們擴容了。也就是說，擴容結束了。 if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT) return; // 如果相等，擴容結束了，更新 finising 變量 finishing = advance = true; // 再次循環檢查一下整張表 i = n; // recheck before commit } }
put方法第三階段
如果對應的節點存在，判斷這個節點的hash是不是等於MOVED(-1)，說明當前節點是ForwardingNode節點， 意味着有其他線程正在進行擴容，那么當前現在直接幫助它進行擴容，因此調用helpTransfer方法
else if ((fh = f.hash) == MOVED) tab = helpTransfer(tab, f); helpTransfer 從名字上來看，代表當前是去協助擴容 final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) { Node<K,V>[] nextTab; int sc; // 判斷此時是否仍然在執行擴容,nextTab=null的時候說明擴容已經結束了 if (tab != null && (f instanceof ForwardingNode) && (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) { int rs = resizeStamp(tab.length);//生成擴容戳 while (nextTab == nextTable && table == tab && (sc = sizeCtl) < 0) {//說明擴容還未完成的情況下不斷循環來嘗試將當前線程加入到擴容操作中 //下面部分的整個代碼表示擴容結束，直接退出循環 //transferIndex<=0表示所有的Node都已經分配了線程 //sc=rs+MAX_RESIZERS 表示擴容線程數達到最大擴容線程數 //sc >>> RESIZE_STAMP_SHIFT !=rs， 如果在同一輪擴容中，那么sc無符號右移比較高位和rs的值，那么應該是相等的。如果不相等，說明擴容結束了 //sc==rs+1 表示擴容結束 if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || sc == rs + MAX_RESIZERS || transferIndex <= 0) break;//跳出循環 if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {//在低16位上增加擴容線程數 transfer(tab, nextTab);//幫助擴容 break; } } return nextTab; } return table;//返回新的數組 }
put方法第四階段
這個方法的主要作用是，如果被添加的節點的位置已經存在節點的時候，需要以鏈表的方式加入到節點中 如果當前節點已經是一顆紅黑樹，那么就會按照紅黑樹的規則將當前節點加入到紅黑樹中
else { //進入到這個分支，說明f是當前nodes數組對應位置節點的頭節點，並且不為空 V oldVal = null; synchronized (f) { //給對應的頭結點加鎖 if (tabAt(tab, i) == f) {//再次判斷對應下標位置是否為f節點 if (fh >= 0) { //頭結點的hash值大於0，說明是鏈表 binCount = 1; //用來記錄鏈表的長度 for (Node<K,V> e = f;; ++binCount) {//遍歷鏈表 K ek; //如果發現相同的key，則判斷是否需要進行值的覆蓋 if (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; if ((e = e.next) == null) { pred.next = new Node<K,V>(hash, key, value, null); break; } } } //如果當前的f節點是一顆紅黑樹 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; //同樣，如果值已經存在，則直接替換 if (!onlyIfAbsent) p.val = value; } } } }
put方法第四個階段
判斷鏈表的長度是否已經達到臨界值8. 如果達到了臨界值，這個時候會根據當前數組的長度來決定是擴容還是將鏈表轉化為紅黑樹。也就是說如果當前數組的長度小於64，就會先擴容。否則，會把當前鏈表轉化為紅黑樹
if (binCount != 0) {//說明上面在做鏈表操作 //如果鏈表長度已經達到臨界值8 就需要把鏈表轉換為樹結構 if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); if (oldVal != null)//如果val是被替換的，則返回替換之前的值 return oldVal; break; }
treeifyBin
在putVal的最后部分，有一個判斷，如果鏈表長度大於8，那么就會觸發擴容或者紅黑樹的轉化操作。
private final void treeifyBin(Node<K,V>[] tab, int index) { Node<K,V> b; int n, sc; if (tab != null) { if ((n = tab.length) < MIN_TREEIFY_CAPACITY) //tab的長度是不是小於64，如果是，則執行擴容 tryPresize(n << 1); else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {//否則，將當前鏈表轉化為紅黑樹結構存儲 synchronized (b) {// 將鏈表轉換成紅黑樹 if (tabAt(tab, index) == b) { TreeNode<K,V> hd = null, tl = null; for (Node<K,V> e = b; e != null; e = e.next) { TreeNode<K,V> p = new TreeNode<K,V>(e.hash, e.key, e.val, null, null); if ((p.prev = tl) == null) hd = p; else tl.next = p; tl = p; } setTabAt(tab, index, new TreeBin<K,V>(hd)); } } } } }
tryPresize
tryPresize里面部分代碼和addCount的部分代碼類似，看起來會稍微簡單一些
private final void tryPresize(int size) { //對size進行修復,主要目的是防止傳入的值不是一個2次冪的整數，然后通過tableSizeFor來講入參轉化為離該整數最近的2次冪 int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY : tableSizeFor(size + (size >>> 1) + 1); int sc; while ((sc = sizeCtl) >= 0) { Node<K,V>[] tab = table; int n; //下面這段代碼和initTable是一樣的，如果table沒有初始化，則開始初始化 if (tab == null || (n = tab.length) == 0) { n = (sc > c) ? sc : c; if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { try { if (table == tab) { @SuppressWarnings("unchecked") Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n]; table = nt; sc = n - (n >>> 2);//0.75 } } finally { sizeCtl = sc; } } } // else if (c <= sc || n >= MAXIMUM_CAPACITY) break; else if (tab == table) {//這段代碼和addCount后部分代碼是一樣的，做輔助擴容操作 int rs = resizeStamp(n); if (sc < 0) { Node<K,V>[] nt; if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || sc == rs + MAX_RESIZERS || (nt = nextTable) == null || transferIndex <= 0) break; if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) transfer(tab, nt); } else if (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2)) transfer(tab, null); } } }