Java併發5：ConcurrentHashMap
為什麼要使用 ConcurrentHashMap
HashMap 是非執行緒安全的，put操作可能導致死迴圈。其解決方案有 HashTable 和 Collections.synchronizedMap(hashMap) 。這兩種方案都是對讀寫加鎖，獨佔式，效率比較低下。
HashMap 在併發執行put操作時會引起死迴圈，因為多執行緒導致 HashMap 的 Entry 連結串列形成環形資料結構，則 Entry 的 next 節點永遠不為空，會死迴圈獲取 Entry。
HashTable 使用 synchronized 來保證執行緒安全，但是線上程競爭激烈的情況下，效率非常低。其原因是所有訪問該容器的執行緒都必須競爭一把鎖。
針對上述問題，ConcurrentHashMap 使用鎖分段技術，容器裡有多把鎖，每一把鎖用於其中一部分資料，當多執行緒訪問不同資料段的資料時，執行緒間就不會存在鎖的競爭。
ConcurrentHashMap 實現（JDK 1.7）
在JDK 1.7中，ConcurrentHashMap 採用了 Segment 陣列和 HashEntry 陣列的方式進行實現。其中 Segment 是一種可重入鎖（ReentrantLock），扮演鎖的角色。而 HashEntry 則是用於儲存鍵值對的資料。結構如下圖所示：
一個 Segment 包含一個 HashEntry 陣列，每個 HashEntry 是一個連結串列結構的元素。每個 Segment 守護一個 HashEntry 陣列的元素。
初始化
初始化時，計算出 Segment 陣列的大小 ssize 和每個 Segment 中 HashEntry 陣列的大小 cap，並初始化 Segment 陣列的第一個元素。其中 ssize 為2的冪次方，預設為16，cap 大小也是2的冪次方，最小值為2。最終結果根據初始化容量 initialCapacity 計算。
//計算segment陣列長度
if (concurrencyLevel > MAX_SEGMENTS)
concurrencyLevel = MAX_SEGMENTS;
// Find power-of-two sizes best matching arguments 首先，初始化了 segments 陣列，其長度 ssize 是通過 concurrencyLevel 計算得出的。需要保證ssize的長度是2的N次方，segments 陣列的長度最大是65536。
然後初始化了 segmentShift 和 segmentMask 這兩個全域性變數，用於定位 segment 的雜湊演算法。segmentShift 是用於雜湊運算的位數， segmentMask 是雜湊運算的掩碼。
之後根據 initialCapaicity 和 loadfactor 這兩個引數來計算每個 Segment 中 HashEntry 陣列的大小 cap。
最後根據以上確定的引數，初始化了 segment 陣列以及 segment[0]。
get操作
整個 get 操作過程都不需要加鎖，因此非常高效。首先將 key 經過 Hash 之後定位到 Segment，然後再通過一個 Hash 定位到具體元素。不需要加鎖是因為 get 方法將需要使用的共享變數都定義成 volatile 型別，因此能線上程之間保持可見性，在多執行緒同時讀時能保證不會讀到過期的值。
put操作
put 方法需要對共享變數進行寫入操作，為了執行緒安全，必須加鎖。 put 方法首先定位到 Segment，然後在 Segment 裡進行插入操作。插入操作首先要判斷是否需要對 Segment 裡的 HashEntry 陣列進行擴容，然後定位新增元素的位置，將其放入到 HashEntry 陣列。
Segment 的擴容比 HashMap 更恰當，因為後者是插入元素後判斷是否已經到達容量，如果到達了就擴容，但是可能擴容後沒有插入，進行了無效的擴容。Segment 在擴容時，首先建立一個原來容量兩倍的陣列，然後將原數組裡的元素進行再雜湊後插入到新的陣列。同時為了高效， ConcurrentHashMap 不會對整個容器進行擴容，而是隻對某個 segment 進行擴容。
size方法
每個 Segment 都有一個 volatile 修飾的全域性變數 count ,求整個 ConcurrentHashMap 的 size 時很明顯就是將所有的 count 累加即可。但是 volatile 修飾的變數卻不能保證多執行緒的原子性，所有直接累加很容易出現併發問題。但是如果在呼叫 size 方法時鎖住其餘的操作，效率也很低。
ConcurrentHashMap 的做法是先嚐試兩次通過不加鎖的方式進行計算，如果兩次結果相同，說明結果正確。如果計算結果不同，則給每個 Segment 加鎖，進行統計。
ConcurrentHashMap 實現（JDK 1.8）
在JDK 1.8中，改變了分段鎖的思路，採用了 Node陣列 + CAS + Synchronized 來保證併發安全。底層仍然採用了陣列+連結串列+紅黑樹的儲存結構。
Node
在JDK 1.8中，使用 Node 替換 HashEntry，兩者作用相同。在 Node 中， val 和 next 兩個變數都是 volatile 修飾的，保證了可見性。
使用 table 變數存放 Node 節點陣列，預設為 null， 預設大小為16，且每次擴容時大小總是2的冪次方。在擴容時，使用 nextTable 存放新生成的資料，陣列為 table 的兩倍。
ForwardingNode 是一個特殊的 Node 節點，hash 值為-1，儲存了 nextTable 的引用。只有table 發生擴容時，其發生作用，作為佔位符放在 table 中表示當前節點為null或者已經被移動。
TreeNode
在 HashMap 中，其核心的資料結構是連結串列。而在 ConcurrentHashMap 中，如果連結串列的資料過長會轉換為紅黑樹來處理。通過將連結串列的節點包裝成 TreeNode 放在 TreeBin 中，然後經由 TreeBin 完成紅黑樹的轉換。
TreeBin
TreeBin 不負責鍵值對的包裝，用於在連結串列轉換為紅黑樹時包裝 TreeNode 節點，用來構建紅黑樹。
初始化：initTable()
在建構函式中，ConcurrentHashMap 僅僅設定了一些引數。當首次插入元素時，才通過 initTable() 方法進行了初始化。
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
//有其他執行緒在初始化，掛起當前執行緒
if ((sc = sizeCtl) < 0)
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;
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;
//下次擴容的大小，相當於0.75*n
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}
複製程式碼該方法的關鍵為sizeCtl。
sizeCtl：控制識別符號，用來控制table初始化和擴容操作的，在不同的地方有不同的用途，其值也不同，所代表的含義也不同：
- 負數代表正在進行初始化或擴容操作
- -1代表正在初始化
- -N 表示有N-1個執行緒正在進行擴容操作
- 正數或0代表hash表還沒有被初始化，這個數值表示初始化或下一次進行擴容的大小
sizeCtl 預設為0。如果該值小於0，表示有其他執行緒在初始化，需要暫停該執行緒。如果該執行緒獲取了初始化的權利，先將其設定為-1。最後將 sizeCtl 設定為 0.75*n，表示擴容的閾值。
put操作
put操作的核心思想依然是根據 hash 計算節點插入在 table 的位置，如果為空，直接插入，否則插入到連結串列或樹中。
首先計算hash值，然後進入迴圈中遍歷table，嘗試插入。
int hash = spread(key.hashCode());
int binCount = 0;
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
//詳細程式碼接下來分別講述
}
複製程式碼首先判斷 table 是否為空，如果為空或者是 null，則先進行初始化操作。
if (tab == null || (n = tab.length) == 0)
tab = initTable();
複製程式碼如果已經初始化過，且插入的位置沒有節點，直接插入該節點。使用CAS嘗試插入該節點。
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
break; // no lock when adding to empty bin
}
複製程式碼如果有執行緒在擴容，先幫助擴容。
//當前位置的hashcode等於-1，需要擴容
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
複製程式碼如果都不滿足，使用 synchronized 鎖寫入資料。根據資料結構的不同，如果是連結串列則插入尾部；如果是樹節點，使用樹的插入操作。
else {
V oldVal = null;
//對該節點進行加鎖處理（hash值相同的連結串列的頭節點）
synchronized (f) {
if (tabAt(tab, i) == f) {
//fh > 0 表示為連結串列，將該節點插入到連結串列尾部
if (fh >= 0) {
binCount = 1;
for (Node<K,V> e = f;; ++binCount) {
K ek;
//hash 和 key 都一樣，替換value
if (e.hash == hash &&
((ek = e.key) == key || (ek != null && key.equals(ek)))) {
oldVal = e.val;
//putIfAbsent()
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;
}
}
}
//樹節點，按照樹的插入操作進行插入
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;
}
}
}
}
複製程式碼在for迴圈的最後，判斷連結串列的長度是否需要連結串列轉換為樹結構。
if (binCount != 0) {
// 如果連結串列長度已經達到臨界值8,把連結串列轉換為樹結構
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
複製程式碼最後，如果是更新節點，前邊已經返回了 oldVal，否則就是插入新的節點。還需要使用 addCount() 方法，為 size 加一。
總結步驟如下：
- 判斷 key 和 value 是否為 null,如果是則丟擲異常
- 計算 hash
- 遍歷 table，插入節點
- 如果 table 為空，進行初始化
- 插入位置為空，直接插入，無需加鎖
- 如果是 ForwardingNode 節點，表示有其他執行緒正在擴容，幫助執行緒一起進行擴容。
- 如果是連結串列結構，遍歷連結串列，如果存在 key 則更新 value，否則插入到連結串列尾部；如果是 TreeBin 節點，按照紅黑樹的方法更新或增加節點。
- 如果完成後發現連結串列長度大於設定的閾值，將其裝換為紅黑樹
- 如果是更新，返回oddVal；如果是插入，使用 addCount() 方法，增加 size, 返回 null。
get方法
- 計算 hash 值
- 判斷 table 是否為空，為空返回 null
- 根據 hash 獲取 Node 節點，如果是該節點則返回 value
- 根據連結串列或紅黑樹查詢到對應節點，返回 value
- 找不到則返回 null
參考資料