溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
JDK1.8中的ConcurrentHashMap如何使用？相信很多沒有經驗的人對此束手無策，為此本文總結了問題出現的原因和解決方法，通過這篇文章希望你能解決這個問題。
ConcurrentHashMap 的初步使用及場景：
ConcurrentHashMap 是 J.U.C 包里面提供的一個線程安全并且高效的 HashMap，所以ConcurrentHashMap 在并發編程的場景中使用的頻率比較高，那么我們就從ConcurrentHashMap 的使用上以及源碼層面來分析 ConcurrentHashMap 到底是如何實現安全性的
api 使用：
ConcurrentHashMap 是 Map 的派生類，所以 api 基本和 Hashmap 是類似，主要就是 put、get 這些方法，接下來基于 ConcurrentHashMap 的 put 和 get 這兩個方法作為切入點來分析 ConcurrentHashMap 的源碼實現。
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 數
//組中的元素，保證每次拿到的數據都是最新的
//(Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
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) {
//默認初始容量為 16
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
//初始化數組，長度為 16，或者初始化在構造 ConcurrentHashMap 的時候傳入的長度
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;//將這個數組賦值給 table
//計算下次擴容的大小，實際就是當前容量的 0.75倍，這里使用了右移來計算
//sc =12
sc = n - (n >>> 2);
}
} finally {
//設置 sizeCtl 為 sc, 如果默認是 16 的話，那么這個時候sc=16*0.75=12
sizeCtl = sc;
}
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) {
// 通過該值與當前線程 probe 求與，獲得cells 的下標元素，和 hash 表獲取索引是一樣的
if ((a = as[(n - 1) & h]) == null) {
//cellsBusy=0 表示 counterCells 不在初始化或者擴容狀態下
if (cellsBusy == 0) {// Try to attach new Cell
//構造一個 CounterCell 的值，傳入元素個數
CounterCell r = new CounterCell(x); // Optimistic create
//通過 cas 設置 cellsBusy 標識，防止其他線程來對 counterCells 并發處理
if (cellsBusy == 0 &&
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;
//說明指定 cells 下標位置的數據不為空，則進行下一次循環
continue; // Slot is now non-empty
}
}
collide = false;
}
//說明在 addCount 方法中 cas 失敗了，并且獲取 probe 的值不為空
else if (!wasUncontended) // CAS already known to fail
// 設置為未沖突標識，進入下一次自旋
wasUncontended = true; // Continue after rehash
// 由于指定下標位置的 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; // At max size or stale
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; // Retry with expanded table
}//更新隨機數的值
h = ThreadLocalRandom.advanceProbe(h);
}
//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) { //初始化容量為 2
CounterCell[] rs = new CounterCell[2];
//將 x 也就是元素的個數放在指定的數組下標位置
rs[h & 1] = new CounterCell(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() 擴容階段:
回到addCount(long x, int check)。判斷是否需要擴容，也就是當更新后的鍵值對總數 baseCount >= 閾值 sizeCtl 時，進行rehash，這里面會有兩個邏輯。
如果當前正在處于擴容階段，則當前線程會加入并且協助擴容
如果當前沒有在擴容，則直接觸發擴容操作
private final void addCount(long x, int check) {
........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;
//當前線程嘗試幫助此次擴容，如果成功，則調用 transfer
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
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 00000000
然后再+2 =1000 0000 0001 1100 0000 0000 0000 0000+10=1000 0000 0001 1100 0000 00000000 0010
高 RESIZE_STAMP_BITS 16 位代表擴容的標記、低RESIZE_STAMP_BITS 16 位代表并行擴容的線程數。這樣來存儲有什么好處呢？
首先在 CHM 中是支持并發擴容的，也就是說如果當前的數組需要進行擴容操作，可以由多個線程來共同負責，這塊后續會單獨講
可以保證每次擴容都生成唯一的生成戳，每次新的擴容，都有一個不同的 n，這個生成戳就是根據 n 來計算出來的一個數字，n 不同，這個數字也不同
第一個線程嘗試擴容的時候，為什么是+2 ？
因為 1 表示初始化，2 表示一個線程在執行擴容，而且對 sizeCtl 的操作都是基于位運算的，所以不會關心它本身的數值是多少，只關心它在二進制上的數值，而 sc + 1 會在低 16 位上加 1。
擴容是 ConcurrentHashMap 的精華之一，擴容操作的核心在于數據的轉移，在單線程環境下數據的轉移很簡單，無非就是把舊數組中的數據遷移到新的數組。但是這在多線程環境下，在擴容的時候其他線程也可能正在添加元素，這時又觸發了擴容怎么辦？可能大家想到的第一個解決方案是加互斥鎖，把轉移過程鎖住，雖然是可行的解決方案，但是會帶來較大的性能開銷。因為互斥鎖會導致所有訪問臨界區的線程陷入到阻塞狀態，持有鎖的線程耗時越長，其他競爭線程就會一直被阻塞，導致吞吐量較低。而且還可能導致死鎖。而 ConcurrentHashMap 并沒有直接加鎖，而是采用 CAS 實現無鎖的并發同步策略，最精華的部分是它可以利用多線程來進行協同擴容簡單來說，它把 Node 數組當作多個線程之間共享的任務隊列，然后通過維護一個指針來劃分每個線程鎖負責的區間，每個線程通過區間逆向遍歷來實現擴容，一個已經遷移完的bucket會被替換為一個ForwardingNode節點，標記當前bucket已經被其他線程遷移完了。接下來分析一下它的源碼實現
fwd:這個類是個標識類，用于指向新表用的，其他線程遇到這個類會主動跳過這個類，因為這個類要么就是擴容遷移正在進行，要么就是已經完成擴容遷移，也就是這個類要保證線程安全，再進行操作。
advance:這個變量是用于提示代碼是否進行推進處理，也就是當前桶處理完，處理下一個桶的標識
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 使用 int 的最大值
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;//更新成員變量
transferIndex = n;//更新轉移下標，表示轉移時的下標
}//新的 tab 的長度
int nextn = nextTab.length;
// 創建一個 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 (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
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 問題導致的擴容重疊的情況。
數據遷移階段的實現分析
通過分配好遷移的區間之后，開始對數據進行遷移。在看這段代碼之前，先來了解一下原理
高低位原理分析
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 鏈
接著，通過 CAS 操作，把 hn 鏈放在 i+n 也就是 14+16 的位置，ln 鏈保持原來的位置不動。并且設置當前節點為 fwd，表示已經被當前線程遷移完了
setTabAt(nextTab, i, ln); setTabAt(nextTab, i + n, hn); setTabAt(tab, i, fwd);
遷移完成以后的數據分布如下
為什么要做高低位的劃分？
要想了解這么設計的目的，我們需要從 ConcurrentHashMap 的根據下標獲取對象的算法來看，在 putVal 方法中 ：(f = tabAt(tab, i = (n - 1) & hash)) == null。通過(n-1) & hash 來獲得在 table 中的數組下標來獲取節點數據，【&運算是二進制運算符，1& 1=1，其他都為 0】
假設我們的 table 長度是 16， 二進制是【0001 0000】，減一以后的二進制是 【0000 1111】。假如某個 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，極大提升了效率。
接下來回到鏈表的擴容代碼：
else {
synchronized (f) {//對數組該節點位置加鎖，開始處理數組該位置的遷移工作
if (tabAt(tab, i) == f) {//再做一次校驗
//ln 表示低位， hn 表示高位;接下來這段代碼的作用是把鏈表拆分成兩部分，0 在低位，1 在高位
Node<K,V> ln, hn;
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;
}
}//如果最后更新的 runBit 是 0，設置低位節點
if (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);
}
//將低位的鏈表放在 i 位置也就是不動
setTabAt(nextTab, i, ln);
//將高位鏈表放在 i+n 位置
setTabAt(nextTab, i + n, hn);
// 把舊 table 的 hash 桶中放置轉發節點，表明此 hash 桶已經被處理
setTabAt(tab, i, fwd);
advance = true;
}//f 是紅黑樹的根節點
else if (f instanceof TreeBin) {
TreeBin<K,V> t = (TreeBin<K,V>)f;
//創建低位的樹節點
TreeNode<K,V> lo = null, loTail = null;
//創建高位的樹節點
TreeNode<K,V> hi = null, hiTail = null;
int lc = 0, hc = 0;
for (Node<K,V> e = t.first; e != null; e = e.next) {
int h = e.hash;
TreeNode<K,V> p = new TreeNode<K,V>
(h, e.key, e.val, null, null);
if ((h & n) == 0) {
if ((p.prev = loTail) == null)
lo = p;
else
loTail.next = p;
loTail = p;
++lc;
}
else {
if ((p.prev = hiTail) == null)
hi = p;
else
hiTail.next = p;
hiTail = p;
++hc;
}
}
//判斷是否長度< 6 轉成鏈表
ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
(hc != 0) ? new TreeBin<K,V>(lo) : t;
hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
(lc != 0) ? new TreeBin<K,V>(hi) : t;
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
setTabAt(tab, i, fwd);
advance = true;
}
}
}
}擴容結束以后的退出機制：
如果線程擴容結束，那么需要退出，就會執行 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);
Helps transfer if a resize is in progress.
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;
//在低 16 位上增加擴容線程數
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
transfer(tab, nextTab);//幫助擴容
break;
}
}
return nextTab;//返回新的數組
}
return table;
}put 方法第四階段：
這個方法的主要作用是，如果被添加的節點的位置已經存在節點的時候，需要以鏈表的方式加入到節點中。如果當前節點已經是一顆紅黑樹，那么就會按照紅黑樹的規則將當前節點加入到紅黑樹中。
如果是鏈表，添加完以后判斷鏈表的長度是否已經達到臨界值 8. 如果達到了臨界值，這個時候會根據當前數組的長度來決定是擴容還是將鏈表轉化為紅黑樹。也就是說如果當前數組的長度小于 64，就會先擴容。否則，會把當前鏈表轉化為紅黑樹
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;
}
}
}
}//說明上面在做鏈表操作
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD)
//如果鏈表長度已經達到臨界值 8 就需要把鏈表轉換為樹結構
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) {//tab 的長度是不是小于 64，如果是，則執行擴容
if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
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);
}
} 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);
}
}
}就這樣基于CHM的put()方法我們基本上就分析完了，多過幾遍源碼理解應該不難。對于有些計算模糊的，可以通過寫測試類進行佐證。
看完上述內容，你們掌握JDK1.8中的ConcurrentHashMap如何使用的方法了嗎？如果還想學到更多技能或想了解更多相關內容，歡迎關注億速云行業資訊頻道，感謝各位的閱讀！
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
