溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
今天小編給大家分享一下java核心基礎ConcurrentHashMap怎么掌握的相關知識點，內容詳細，邏輯清晰，相信大部分人都還太了解這方面的知識，所以分享這篇文章給大家參考一下，希望大家閱讀完這篇文章后有所收獲，下面我們一起來了解一下吧。
ConcurrentHashMap是基于Hash表的Map接口實現，鍵與值均不允許為NULL，他是一個線程安全的Map。同時他也是一個無序的Map，不同時間進行遍歷可能會得到不同的順序。在JDK1.8之前，ConcurrentHashMap使用分段鎖以在保證線程安全的同時獲得更大的效率。JDK1.8開始舍棄了分段鎖，使用自旋+CAS+sync關鍵字來實現同步。本文所述便是基于JDK1.8。
ConcurrentHashMap與HashMap有共同之處，一些HashMap的基本概念與實現，本文不再贅述。
public class ConcurrentHashMap<K,V> extends AbstractMap<K,V>
implements ConcurrentMap<K,V>, Serializable可以看到ConcurrentHashMap繼承了AbstractMap及ConcurrentMap抽象類，并實現了Serializable接口，這說明ConcurrentHashMap是一個線程安全的標準Map，且允許序列化。與HashMap不同是ConcurrentHashMap不允許Clone。
ConcurrentHashMap同樣是采用懶初始化的方式，有實際元素時才進行容器的初始化。因此其構造方法與HashMap相差無幾。
// 無參構造
public ConcurrentHashMap() {
}
// 帶有初始容量的構造
public ConcurrentHashMap(int initialCapacity) {
if (initialCapacity < 0)
throw new IllegalArgumentException();
int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
MAXIMUM_CAPACITY :
tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1)); //求最小二次冪
this.sizeCtl = cap; // sizeCtl 暫存容量
}
// 帶有初始集合的構造
public ConcurrentHashMap(Map<? extends K, ? extends V> m) {
this.sizeCtl = DEFAULT_CAPACITY;
putAll(m);
}
// 指定初始容量和裝載因子的構造
public ConcurrentHashMap(int initialCapacity, float loadFactor) {
this(initialCapacity, loadFactor, 1);
}以上構造并沒有什么特別的，邏輯也相對簡單，不再詳細解析，感興趣的話可以到前言提到的HashMap篇了解。
除此之外，ConcurrentHashMap擁有另外一個值得注意的構造方法：
指定初始容量，裝載因子以及并發級別的構造方法：
源碼：
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {
if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0) // 基礎校驗
throw new IllegalArgumentException();
if (initialCapacity < concurrencyLevel) // 取初始容量和并發級別的最大值
initialCapacity = concurrencyLevel;
long size = (long)(1.0 + (long)initialCapacity / loadFactor); // 指定了裝載因子，因此就用初始容量和裝載因子計算出實際要的初始容量
int cap = (size >= (long)MAXIMUM_CAPACITY) ?
MAXIMUM_CAPACITY : tableSizeFor((int)size); // 確定計算出初始容量,
this.sizeCtl = cap;
}解析： 除了初始容量與裝載因子外，此構造方法還有一個并發級別concurrencyLevel的參數。在jdk1.7時，并發級別作為分段鎖的標準進行分段。但是jdk1.8開始舍棄了分段鎖，為了版本兼容，此構造方法依然存在，但是concurrencyLevel也不再具有其分段依據的意義，而是作為初始容量的定義依據。
ConcurrentHashMap采用懶初始化的方式，在第一次putAvl時如果容器為空，則會調用initTable()進行容器的初始化：
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) { // tab為空時就一直循環
if ((sc = sizeCtl) < 0) // sc小于0說明正在初始化或者正在復制，放棄CPU等下一次
Thread.yield(); // lost initialization race; just spin
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { // 否則的話就用CAS把sizectl設置成-1,標識正在初始化，且成功的話
try {
// CAS之后再判斷一下，這是為了避免并發，比如上面判斷了tab為空，然后另一個線程做了初始化操作，結束時sc被設置為擴容閾值，然后繼續(sc = sizeCtl) < 0,這時cs還是大于0，所以還是能走進來的，所以這里再判斷一下
if ((tab = table) == null || tab.length == 0) {
int n = (sc > 0) ? sc : DEFAULT_CAPACITY; // sc最初始是存在的初始容量的
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n]; // 建立一個容器
table = tab = nt; // 然后復制到table屬性上
sc = n - (n >>> 2); // sc等于 n>>>2就是n/4就是0.25n sc = 0.75n,相當于設置擴容閾值
}
} finally {
sizeCtl = sc; // sizeCtl賦值成擴容閾值
}
break; // 只要走到這里面,那么說明一定已經初始化結束了，無論初始化是自己做的還是其他線程做的，然后就跳出
}
}
return tab; // 返回新的tab
}解析： 容器初始化完成之前不斷的進行循環。這是因為ConcurrentHashMap是一個支持并發的Map，可能同時會有多個線程進入initTable方法，但是只有一個線程執行初始化操作，那么剩下的線程就需要等待初始化完成再跳出initTable方法，以滿足接下來的putVal操作。
為了保證只有一個線程執行初始化操作，使用sizeCtl來作為標識，sizeCtl為-1時即說明當前已有線程正在初始化，則放棄CPU繼續循環。sizeCtl不為負數時，則使用CAS(通過Unsafe+偏移量的方式實現)將sizeCtl置為-1,如果當前線程成功的話則進入實際的擴容操作。
通過CAS鎖定成功后再次判斷容器是否為空,這是為了避免并發，比如上面判斷了tab為空，然后另一個線程做了初始化操作，結束時sc被設置為擴容閾值，然后繼續(sc = sizeCtl) < 0,這時cs還是大于0，所以還是能走進來的，所以這里再判斷一下。
如果sc(sizeCtl原值)大于0，則以sc做為初始容量。這個在構造方法篇提到過，對于有初始容量要求的構造，會以sizeCtl暫存初始容量。否則的話就取默認的初始容量DEFAULT_CAPACITY（16）。
接下來就建立目標容量的容器并賦值給容器屬性table,計算該容量下的擴容閾值賦值給sizeCtl 。sizeCtl的賦值放到finally里面的原因是因為無論容器創建成功還是失敗,都需要放開以sizeCtl為負值作為判斷條件的鎖，以保證在本線程創建失敗的情況其它線程能繼續競爭鎖繼續進行容器創建的工作。
至此，容器的初始化便完成了。
ConcurrentHashMap新數據載入主要通過putVal實現：
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode());
int binCount = 0;
for (Node<K,V>[] tab = table;;) { // 會一直循環，直到break
Node<K,V> f; int n, i, fh;
if (tab == null || (n = tab.length) == 0)
tab = initTable(); // 初始化table
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { // unsafe獲取i處的數據，如果為null
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null))) // cas操作，如果i處為null就構造新元素，成功就break 失敗就繼續循環
break; // no lock when adding to empty bin
}
else if ((fh = f.hash) == MOVED) // f處的元素hash是-1 -1意味著這是一個遷移節點
tab = helpTransfer(tab, f); // 輔助遷移開始
else {
V oldVal = null;
synchronized (f) { // 用f進行同步
if (tabAt(tab, i) == f) { // 這里要保證i處還是f,而不是已經被其它元素并發占據了
if (fh >= 0) { // f處的hash要大于等于0,這是為什么呢？除了MOVED還會有其它負值嗎
binCount = 1;
for (Node<K,V> e = f;; ++binCount) { // 從e往后找，并且記錄元素數量
K ek;
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) { // 一系列判斷Key是否相等，如果相等
oldVal = e.val; // 記錄舊值
if (!onlyIfAbsent) // 不是僅允許新增的話
e.val = value; // 記錄新值
break; // 成功的話就跳出大循環
}
Node<K,V> pred = e;
if ((e = e.next) == null) { // e向后移動, 如果移動之后是nul,就是在末尾的話
pred.next = new Node<K,V>(hash, key,
value, null); // 建立一個新節點
break; // 成功的話就跳出大循環
}
}
}
else if (f instanceof TreeBin) { // 如果是一個樹化節點
Node<K,V> p;
binCount = 2; // 樹化的節點的話固定元素數量為2，這是一個相對特殊的值，即會擴容又不會重復樹化
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) { // 往樹里面新增元素，如果存在就返回那個節點，不存在就是新增節點
oldVal = p.val; // 記錄舊值
if (!onlyIfAbsent) // 不是僅允許新增的話
p.val = value; // 記錄新值
}
}
}
}
if (binCount != 0) { // 如果元素數不為0的話
if (binCount >= TREEIFY_THRESHOLD) // 判斷是否到了樹化閾值
treeifyBin(tab, i); // 到了的話進行樹化
if (oldVal != null) // 覆蓋操作直接返回舊值不會執行addCount
return oldVal;
break; // binCount != 0才跳出大循環，為0是tabAt(tab, i) != f導致的，即i處已經被其他線程覆蓋了
}
}
}
addCount(1L, binCount); //
return null;
}解析： 完成為空校驗后，通過spread方法來計算出hash以確定下標位置，spread的計算方式為(h ^ (h >>> 16)) & HASH_BITS,HASH_BITS的值為0x7fffffff，描述出來就是將hashCode的高16位和低16位做異或操作，并保證最高位符號位為0（結果是一個正數），注解表示這樣做的原因是為了使數據更加分散，盡可能的避免hash沖突。
如果容器tab為空或者長度為0說明容器未初始化，那么就調用initTable進行容器初始化。
否則的話對hash與現容量進行與運算得出數據應處的下標，判斷此下標所在處節點是否為空，為空說明沒有元素，直接創建一個新節點作為頭元素放進去，這一步通過CAS操作實現，避免并發情況下另外的線程先一步完成頭節點創建操作。
如果下標所在處節點不為空，說明該處已經有元素了，此時判斷頭節點的hash是否為MOVED，MOVED的值為-1。上面提到過，正常元素通過spread方法計算出來的hash值都會使正數。此處-1為一個特殊值，意味著此節點正在進行擴容遷移工作，那么此時就調用8 輔助擴容-helpTransfer方法進行輔助遷移工作。
如果節點hash不為MOVED，意味著這是一個正常節點，就執行元素載入工作，使用synchronized關鍵字實現同步， 同步塊內開始還要使用tabAt(tab, i) == f判斷進入同步后i處的節點還是原節點，接下來就是元素的載入工作了，整體和HashMap的流程是相差無幾的，感興趣的話可以去文首提到的解析HashMap的文章了解一下。synchronized塊將元素載入節點后，接著會對binCount進行判斷，binCount在節點還處于鏈表模式下的情況下記錄節點在新元素載入后的元素總數量，此處判斷binCount達到TREEIFY_THRESHOLD(8) 樹化閾值后，就會對該節點進行樹化操作。樹化操作依然是通過synchronized來完成的，這里不做過多延伸。
以上整個新元素載入操作都是一個自旋的過程，一直到新元素載入成功后通過break跳出自旋過程。
新元素載入節點后，需要對count實時維護，count通過 6 維護與啟動擴容-addCount 方法完成同步維護工作。
ConcurrentHashMap通過addCount方法實時維護內部元素數量，并在達到擴容閾值的情況下啟動擴容操作：
private final void addCount(long x, int check) {
CounterCell[] as; long b, s;
if ((as = counterCells) != null || // counterCells不為空,如果沒經歷過并發肯定是空的，這個判斷意思是維護count的過程中已經經歷過并發,已經處于不使用baseCount維護的階段了
!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) { //這個比較唯一會出現false的情況就是b = baseCount執行之后，還沒進行cas時,baseCount又被改了。意思是如果沒有并發就用baseCount維護，并發導致baseCount維護失敗就進入counterCells維護
CounterCell a; long v; int m;
boolean uncontended = true; // 記錄是否遭遇了并發
if (as == null || (m = as.length - 1) < 0 || // as等于null意味著baseCount維護遭遇并發了,且是初次(非初次counterCells不會為null) (m = as.length - 1) < 0說明counterCells長度為0，即沒有實際內容
(a = as[ThreadLocalRandom.getProbe() & m]) == null || // counterCells是一個分段計算的概念，這里就是隨機找一個槽，如果這個槽位空的話
!(uncontended =
U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) { // 槽不為空的話就用cas把槽里面的值賦值給v+a.val 如果失敗的話
fullAddCount(x, uncontended); // 用fullAddCount 就是LongAdder的模式將x放進去，初始化counterCells
return;
}
if (check <= 1) // check小于等于1(1表示是當前節點的第一個元素,-1標識是replace方法或者clear過來的，這些都不需要擴容)
return;
s = sumCount(); // 計算一下現在的總量
}
if (check >= 0) { //baseCount成功的情況且 check大于0，即不是replace或者clear方法進來的
Node<K,V>[] tab, nt; int n, sc;
/**
1.s >= (long)(sc = sizeCtl) s是總數量 s大于等于擴容閾值sizeCtl時，或者sizeCtl為負數，正在擴容時
2.(tab = table) != null 并且容器tab不為空，意思就是表已經初始化完畢了,是執行擴容而不是執行初始化
3. (n = tab.length) < MAXIMUM_CAPACITY) 并且容量還沒有達到最大容量時
*/
while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
(n = tab.length) < MAXIMUM_CAPACITY) {
int rs = resizeStamp(n); // 獲得容量特征值：容量n的左0數量并將高位置1
if (sc < 0) { // 如果sc小于0 意味這個正在進行并發擴容
/**
sc >>> RESIZE_STAMP_SHIFT != rs是判斷現在的容量與sc中記錄的長度是不是不一樣長,不一樣長的話說明不是同一長度的擴容,就要break-保證是同等長度的擴容
sc == rs + 1 這個地方在看來是一個bug,如果已經處于擴容中,sc目前處于高16位是容量特征值,低16位標識擴容線程量特征值的狀態，而此時的rs是容量特征值的狀態,也就是目前sc的高16位,這里原意應該是判斷目前是不是只剩下一個非初始擴容線程
,為什么這么說呢，因為初始擴容線程會把sc設置稱高16位是容量特征值即rs,低16位標識擴容線程量特征值(初始擴容線程+2,非初始擴容線程+1),這一步判斷的就是如過只剩下一個非初始擴容線程了,說明初始擴容線程已經結束，現在已經進入擴容的最后階段了，就不需要再繼續進行輔助擴容了
,正確的方式應該是sc == (rs << RESIZE_STAMP_SHIFT) + 1
sc == rs + MAX_RESIZERS 和上面+1一樣，不過這個是判斷參與resize的線程是否已經達到所允許的最大值 ,看起來也是一個bug,正確的應是sc == (rs << RESIZE_STAMP_SHIFT) + MAX_RESIZERS
(nt = nextTable) == null nextTable為空說明已經擴容完畢了,不再需要輔助擴容了
transferIndex <= 0 // 判斷下邊界是不是已經到0了,擴容是根據CPU算出步長,一個擴容線程擴容一段，從高到低分配，transferIndex記錄的就行目前分配到哪個下標了,如果這個下標小于等于0說明整個容器已經分配完了,就不需要在輔助擴容了
*/
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)) // 如果是輔助擴容的sc已經處于高16位是容量特征值,低16位標識擴容線程量特征值的狀態了+1讓地位的擴容線程量特征值再+1
transfer(tab, nt); // 進行擴容
}
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2)) // 初始擴容現在是+2 執行完這一步sc的值高16位是容量特征值,低16位標識擴容線程量特征值,因為resizeStamp時將高位置為1是的其又是一個負數，表示正在擴容
transfer(tab, null);
s = sumCount(); // 計算總數
} // 一直循環擴容
}
}解析： count的維護有兩種方式：未產生并發場景時通過baseCount維護，經歷過并發場景后轉變為通過counterCells維護。baseCount模式為直接使用int進行加減運算，cas保證同步，counterCells模式類似于1.8之前的ConcurrentHashMap，采用一個分段的概念，運算時隨機找一個槽，通過cas保證一個槽的值加算同步，count即為所以槽的加和(使用LongAdder實現)。
addCount方法初始即通過counterCells是否為空(為空說明未經歷初始化，使用的baseCount模式)來判斷當前是否為counterCells模式，如果處于counterCells模式則進入counterCells模式計算邏輯，否則的話使用baseCount模式來維護count并記錄為s，如果遭遇并發導致維護失敗，則轉換為counterCells模式進入counterCells模式計算邏輯。
進入counterCells模式，首先看counterCells是否已完成初始化，若為初始化進入fullAddCount，已初始化的話則隨機找到counterCells中的一個槽位，如果這個槽位未初始化則進入fullAddCount，如果已初始化則對這個操作進行CAS的加操作來維護count，CAS成功則維護完成，并發導致CAS維護失敗則進入fullAddCount。fullAddCount有點類似于CMS的full GC退化機制，會完成counterCells的初始化以及并發沖突場景下同步完成count維護。counterCells最后則判斷check是否小于等于1(1表示是當前節點的第一個元素,-1標識是replace方法或者clear過來的，這些都不需要擴容)則直接結束，不進行擴容判定，否則的話通過sumCount獲得當前count并記錄為s以準備后續的擴容判定。
check大于等于0(即不是replace或者clear方法引起的count改變)時進行擴容判定，執行擴容需要滿足三個條件：1.count大于擴容閾值或者當前正在執行擴容操作2.容器已初始化完畢3.容量未達到最大容量。滿足以上條件則需判定當前是否正在擴容，因為擴容時會將SIZECTL設置為一個特征值，這個特征值為負值，因此通過sc是否小于0來判定當前是否處于擴容狀態。接下來使用resizeStamp獲得容量特征值(一個表示容量n的左0數量并將高位置1的值)，如果當前未處于擴容過程，則通過容量特征值獲得擴容特征值(高16位為容量特征值，低16位為擴容線程量特征值)并通過CAS修改SIZECTL為擴容特征值，成功則當前線程作為第一個擴容線程啟動擴容，失敗則重新計算count并進行擴容判定。
如果已處于擴容過程，則判斷是否還要寫輔助擴容線程，滿足以下幾個條件則：1.此次擴容的容量與當前正在進行的容量一致2.擴容線程未達到最大額定值3.不是處于擴容收尾階段(只剩下一個輔助擴容線程，nextTable已經為空，擴容下標已為0無法繼續分配)則使用CAS修改擴容特征值成功后當前線程作為輔助擴容線程加入擴容任務，不滿足條件則重新計算count并自旋進行擴容判定。
ConcurrentHashMap的實際擴容操作通過transfer來完成：
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
// CPU大于1的話 步長就是0.25n/CPU數 否則就是n 算出來的步長小于最小步長的話 就按最小步長來
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
if (nextTab == null) { // initiating // nextTab == null 說明是第一個來擴容的線程
try {
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1]; // 建立一個二倍長度的數組，然后賦值給nextTab
nextTab = nt;
} catch (Throwable ex) { // try to cope with OOME
sizeCtl = Integer.MAX_VALUE; // 異常的話就是OOM? 唯一的異常也就是n過大？
return;
}
nextTable = nextTab; // 這個賦值為什么不做同步呢？不怕并發問題嗎,哦 不會 因為調用這個方法的地方已經確定了，只會有一個初始線程過來
transferIndex = n; // n就是新的長度
}
int nextn = nextTab.length; // 新容量
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab); // 創建一個中繼節點
boolean advance = true; //一個終止的標記
boolean finishing = false; // to ensure sweep before committing nextTab // 是否已經處于收尾掃描提交階段
for (int i = 0, bound = 0;;) { // 一直循環著按步長擴容
Node<K,V> f; int fh;
while (advance) { // 計算出bound和i 賦值TRANSFERINDEX
int nextIndex, nextBound;
if (--i >= bound || finishing) // i大于bound，說明什么呢，說明i已經在區間內了
advance = false;
else if ((nextIndex = transferIndex) <= 0) { 如果下一次的上邊界已經到0了 也停止
i = -1;
advance = false;
}
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ? //nextBound作為新的上邊界
nextIndex - stride : 0))) {
bound = nextBound; //nextBound作為這次的下邊界
i = nextIndex - 1; // 從nextIndex(上邊界) - 1開始依次遞減到bound區域
advance = false;
}
}
if (i < 0 || i >= n || i + n >= nextn) { // 如果i不在0-n范圍內,已經越界
int sc;
if (finishing) { // 如果已經在收尾階段
nextTable = null; // 重置nextTable
table = nextTab; // 容器確認
sizeCtl = (n << 1) - (n >>> 1); // 2n-0.5n = 0.75*2n 就是新的擴容閾值
return;
}
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) { // sc減1 意味當前擴容線程結束
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT) // 如果sc-2和容量特征值一樣時，就說明現在是最后的一個擴容線程了
return; // 直接返回
finishing = advance = true; // 不是最后的話 設置為收尾階段？，確實，進入這里面已經越界了...
i = n; // recheck before commit // 提交前的重復檢查？
}
}
else if ((f = tabAt(tab, i)) == null) // 如果i處還沒有使用
advance = casTabAt(tab, i, null, fwd); // cas設置i處的新fwdnode,成功還是失敗會記錄在advance上面
else if ((fh = f.hash) == MOVED) // 如果這節點已經是一個中繼節點 說明什么呢？說明已經有人處理過了上一步
advance = true; // already processed
else { // 如果有值，并且不是中繼節點的話
synchronized (f) { // 同步處理f，所以只會有一個線程作為參與者
if (tabAt(tab, i) == f) { // 同步之后還要再判斷一下 萬一f被并發處理掉了呢
Node<K,V> ln, hn;
if (fh >= 0) { // fh>=0說明是一個還沒樹化的鏈表？
int runBit = fh & n;
Node<K,V> lastRun = f;
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) { // 最后一段的開頭節點是低位的話
ln = lastRun; // 開頭節點賦值到ln 也就是說從lastRun開始，之后的都沒有變化了，都是在同一位
hn = null;
}
else {
hn = lastRun;
ln = null;
}
for (Node<K,V> p = f; p != lastRun; p = p.next) { // 只需遍歷到lastRun即可
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); // 賦值新容器
setTabAt(nextTab, i + n, hn);
setTabAt(tab, i, fwd); // 舊容器節點變為中繼節點，但是如果在這一步之前就有新元素來了怎么辦，不會來，因為新增也是sync(f)的
advance = true;
}
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;
}
}
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采用分段擴容的方式，從n-1->0,每個線程每次會占用一個步長的區間，然后針對這個區間進行擴容，擴容完畢再去占用下一個區間，直到無法占用新的區間則結束擴容流程。
步長的計算與應用的CPU數有關，如果當前應用CPU數為1則步長為n,即單線程庫擴容。如果當前應用CPU數不為1，則為0.25n/CPU，但是如果算出的步長小于最小MIN_TRANSFER_STRIDE(16),則以MIN_TRANSFER_STRIDE為步長。
接下來通過nextTable 是否為空判斷是否為初始擴容線程，如果是的話就創建一個長度為2n的新容器并記錄到nextTable屬性，所有擴容線程共同使用nextTable作為新容器，同時記錄當前擴容進度下標transferIndex為n（擴容占用是從n到0的，因此初始未占用時就是n）。這里雖然沒進行加鎖但是依然不會有并發問題，因為nextTable為空的只有初始線程，而初始線程的創建則是線程安全的。
步長確認完畢，新容器創建完畢，擴容進度下標初始化完成，接下來就開始進行區間占用了：使用自旋+CAS來進行線程安全的區間占用，advance作為自旋結束的條件：如果i大于bound意味著本次占用的區間還沒處理完則結束自旋。finishing意味著擴容已經進入尾聲了，也無需再加入擴容，結束自旋。擴容進度下標transferIndex小于等于0說明所有的區間都已經被占用了也無需再加入擴容，結束自旋。如果以上條件不成立，則使用CAS修改擴容進度下標transferIndex嘗試占用一個步長的區間，如果失敗則自旋繼續判定區間占用，修改成功則意味著區間占用成功，賦值本次區間的擴容游標i以及下標邊界bound。
區間占用成功后開始從i到bound的擴容階段，首先需要判斷i的合法性(需要在i和n之間，即判斷是不是已擴容完畢，如–i的循環會使得i為負數表示擴容完畢)，如果沒占據到區間則判斷是不是處于于finishing階段，處于的話說明自己以及是最后一個擴容線程了，將新容器nextTable賦值為table，并設置新的擴容閾值。如果不是處于finishing階段則因為當前線程沒占據到區間需要退出擴容，那么響應的擴容特征值也要相應的-1來完成擴容線程量特征值的維護，如果擴容特征值維護失敗則需要繼續自旋嘗試退出，如果維護成功則判斷自己退出之后是不是只能先一個擴容線程,因為起始擴容線程的線程擴容量特征值為+2，所以通過-2來判斷，如果自己退出之后僅剩一個線程在擴容，那么就把finishing設置為true，以使其在擴容工作結束后進行新容器的賦值以及新擴容閾值的維護。
如果i處于合法范圍內，說明處于正常擴容中，那么獲取i處的節點，節點為空則利用一個CAS為其賦值一個ForwardingNode節點，這個節點是一個中繼節點，意味著這個節點正處于擴容狀態，其hash值為特殊值MOVED(-1)，在5 新數據載入-putVal 的過程中如果發現目標節點hash值為MOVED，那么putVal線程就會暫停put操作，作為輔助擴容線程先行擴容容器，擴容完畢后再進行put操作。
如果i處節點不會空但是hash值已經是MOVED，那么說明節點已經處于遷移狀態則跳過。
如果以上均不符合，說明i合法，且i處擁有實際的數據節點，那么使用i處的節點f作為鎖通過synchronized來達成線程安全的擴容，因為putVal過程中也是用i處的節點f作為鎖進行synchronized的，意味著對i處的操作擴容和賦值只有一個過程能操作，以此來保證putVal和transfer的并發安全性。synchronized內部的擴容過程因為已經保證了線程同步，因此和HashMap的擴容過程區別并不大，這里不再描述，感興趣的話可以查看9: 從源碼看HashMap:一文看懂HashMap了解。
在5 新數據載入-putVal 中以及7 進行擴容-transfer中都提到過如果putVal過程中發現目標節點是一個中繼節點ForwardingNode(hash值為特殊值MOVED)那么putVal會暫停put操作，通過helpTransfer方法進行輔助擴容：
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
Node<K,V>[] nextTab; int sc;
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) {
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || transferIndex <= 0)
break;
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
transfer(tab, nextTab);
break;
}
}
return nextTab;
}
return table;
}解析： 這個方法是不是看起來很熟悉，是的，除了ForwardingNode判斷，其主要邏輯和6 維護與啟動擴容-addCount判定擴容時的邏輯大致相當。因此也不在額外表述。這里貼出來的目的僅僅是為了完整的體現putVal遭遇擴容并發的處理過程。
以上就是“java核心基礎ConcurrentHashMap怎么掌握”這篇文章的所有內容，感謝各位的閱讀！相信大家閱讀完這篇文章都有很大的收獲，小編每天都會為大家更新不同的知識，如果還想學習更多的知識，請關注億速云行業資訊頻道。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
