如何進行ConcurrentHashMap內部實現

如何進行ConcurrentHashMap內部實現，針對這個問題，這篇文章詳細介紹了相對應的分析和解答，希望可以幫助更多想解決這個問題的小伙伴找到更簡單易行的方法。

ConcurrentHashMap可以說是目前使用最多的并發數據結構之一，作為如此核心的基本組件，不僅僅要滿足我們功能的需求，更要滿足性能的需求。而實現一個高性能的線程安全的HashMap也絕非易事。

ConcurrentHashMap作為JDK8的內部實現，一個成功的典范，有著諸多可以讓我們學習和致敬的地方。

我全局在項目中搜索這個類的時候，發現大量項目代碼和源碼都用到了，為什么他會這么吃香呢?到底是道德的....呸。

下面我們就來扒一扒，ConcurrentHashMap的內部實現，來體會一下它的精妙之處吧!

ConcurrentHashMap的內部數據結構

在JDK8中，  ConcurrentHashMap的內部實現發生了天翻地覆的變化。這里依據JDK8，來介紹一下ConcurrentHashMap的內部實現。

從靜態數據結構上說，ConcurrentHashMap包含以下內容：

int sizeCtl

這是一個多功能的字段，可以用來記錄參與Map擴展的線程數量，也用來記錄新的table的擴容閾值

CounterCell[] counterCells

用來記錄元素的個數，這是一個數組，使用數組來記錄，是因為避免多線程競爭時，可能產生的沖突。使用了數組，那么多個線程同時修改數量時，極有可能實際操作數組中不同的單元，從而減少競爭。

Node<K,V>[] table

實際存放Map內容的地方，一個map實際上就是一個Node數組，每個Node里包含了key和value的信息。

Node<K,V>[] nextTable

當table需要擴充時，會把新的數據填充到nextTable中，也就是說nextTable是擴充后的Map。

以上就是ConcurrentHashMap的核心元素，其中最值得注意的便是Node，Node并非想象中如此簡單，下面的圖展示了Node的類族結構：

可以看到，在Map中的Node并非簡單的Node對象，實際上，它有可能是Node對象，也有可能是一個Treebin或者ForwardingNode。

那什么時候是Node，什么時候是TreeBin，什么時候又是一個ForwardingNode呢?

其實在絕大部分場景中，使用的依然是Node，從Node數據結構中，不難看出，Node其實是一個鏈表，也就是說，一個正常的Map可能是長這樣的:

上圖中，綠色部分表示Node數組，里面的元素是Node，也就是鏈表的頭部，當兩個元素在數據中的位置發生沖突時，就將它們通過鏈表的形式，放在一個槽位中。

當數組槽位對應的是一個鏈表時，在一個鏈表中查找key只能使用簡單的遍歷，這在數據不多時，還是可以接受的，當沖突數據比較多少，這種簡單的遍歷就有點慢了。

因此，在具體實現中，當鏈表的長度大于等于8時，會將鏈表樹狀化，也就是變成一顆紅黑樹。如下圖所示，其中一個槽位就變成了一顆樹，這就是TreeBin(在TreeBin中使用TreeNode構造整科樹)。

當數組容量快滿時，即超過75%的容量時，數組還需要進行擴容，在擴容過程中，如果老的數組已經完成了復制，那么就會將老數組中的元素使用ForwardingNode對象替代，表示當前槽位的數據已經處理了，不需要再處理了，這樣，當有多個線程同時參與擴容時，就不會沖突。

put()方法的實現

現在來看一下作為一個HashMap最為重要的方法put()：

• public V put(K key, V value)

它負責將給定的key和value對存入HashMap，它的工作主要有以下幾個步驟：

1. 鴻蒙官方戰略合作共建——HarmonyOS技術社區

2. 如果沒有初始化數組，則嘗試初始化數組

3. 如果當前正在擴容，則參與幫助擴容(調用helpTransfer()方法)

4. 將給定的key,value 放入對應的槽位

5. 統計元素總數

6. 觸發擴容操作

根據以上主要4個步驟，來依次詳細說明一下：

如果沒有初始化數組，則嘗試初始化數組

初始化數據會生成一個Node數組:

Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];

默認情況下，n為16。同時設置sizeCtl為&middot;n - (n >>> 2);  這意味著sizeCtl為n的75%，表示Map的size，也就是說ConcurrentHashMap的負載因子是0.75。(為了避免沖突，Map的容量是數組的75%，超過這個閾值，就會擴容)

如果當前正在擴容，則參與幫助擴容

else if ((fh = f.hash) == MOVED)     tab = helpTransfer(tab, f);

如果一個節點的hash是MOVE，則表示這是一個ForwardingNode，也就是當前正在擴容中，為了盡快完成擴容，當前線程就會參與到擴容的工作中，而不是等待擴容操作完成，如此緊密細致的操作，恰恰是ConcurrentHashMap高性能的原因。

而代碼中的f.hash==MOVE語義上等同于f instanceof  ForwardingNode，但是使用整數相等的判斷的效率要遠遠高于instanceof，所以，這里也是一處對性能的極限優化。

將給定的key,value 放入對應的槽位

在大部分情況下，應該會走到這一步，也就是將key和value放入數組中。在這個操作中會使用大概如下操作：

Node<K,V> f; synchronized (f) {      if(所在槽位是一個鏈表)          插入鏈表      else if(所在槽位是紅黑樹)          插入樹      if(鏈表長度大于8[TREEIFY_THRESHOLD])          將鏈表樹狀化 }

可以看到，這使用了synchronized關鍵字，鎖住了Node對象。由于在絕大部分情況下，不同線程大概率會操作不同的Node，因此這里的競爭應該不會太大。

并且隨著數組規模越來越大，競爭的概率會越來越小，因此ConcurrentHashMap有了極好的并行性。

統計元素總數

為了有一個高性能的size()方法，ConcurrentHashMap使用了單獨的方法來統計元素總數，元素數量統計在CounterCell數組中：

CounterCell[] counterCells; @sun.misc.Contended static final class CounterCell {     volatile long value;     CounterCell(long x) { value = x; } }

CounterCell使用偽共享優化，具有很高的讀寫性能。counterCells中所有的成員的value相加，就是整個Map的大小。這里使用數組，也是為了防止沖突。

如果簡單使用一個變量，那么多線程累加一個計數器時，難免要有競爭，現在分散到一個數組中，這種競爭就小了很多，對并發就更加友好了。

累加的主要邏輯如下：

if (as == null || (m = as.length - 1) < 0 ||     //不同線程映射到不同的數組元素，防止沖突     (a = as[ThreadLocalRandom.getProbe() & m]) == null ||     //使用CAS直接增加對應的數據     !(uncontended =       U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x)))     //如果有競爭，在這里會重試，如果競爭嚴重還會將CounterCell[]數組擴容，以減少競爭

觸發擴容操作

最后，ConcurrentHashMap還會檢查是否需要擴容，它會檢查當前Map的大小是否超過了閾值，如果超過了，還會進行擴容。

ConcurrentHashMap的擴容過程非常巧妙，它并沒有完全打亂當前已有的元素位置，而是在數組擴容2倍后，將一半的元素移動到新的空間中。

所有的元素根據高位是否為1分為low節點和high節點：

//n是數組長度，數組長度是2的冪次方，因此一定是100 1000 10000 100000這種二進制數字 //這里將low節點串一起， high節點串一起 if ((ph & n) == 0)     ln = new Node<K,V>(ph, pk, pv, ln); else     hn = new Node<K,V>(ph, pk, pv, hn);

接著，重新放置這些元素的位置：

//low節點留在當前位置 setTabAt(nextTab, i, ln); //high節點放到擴容后的新位置，新位置距離老位置n setTabAt(nextTab, i + n, hn); //擴容完成，用ForwardingNode填充 setTabAt(tab, i, fwd);

下圖顯示了 從8擴充到16時的可能得一種擴容情況，注意，新的位置總是在老位置的后面n個槽位(n為原數組大小)

這樣做的好處是，每個元素的位置不需要重新計算，進行查找時，由于總是會對n-1(一定是一個類似于1111 11111  111111這樣的二進制數)按位與，因此，high類的節點自然就會出現在+n的位置上。

get()方法的實現

與put()方法相比，get()方法就比較簡單了。步驟如下：

1. 鴻蒙官方戰略合作共建——HarmonyOS技術社區

2. 根據hash值 得到對應的槽位 (n - 1) & h

3. 如果當前槽位第一個元素key就和請求的一樣，直接返回

4. 否則調用Node的find()方法查找

5. 對于ForwardingNode 使用的是 ForwardingNode.find()

6. 對于紅黑樹 使用的是TreeBin.find()

7. 對于鏈表型的槽位，依次順序查找對應的key

ConcurrentHashMap可以說是并發設計的典范，在JDK8中，ConcurrentHashMap可以說是再一次脫胎換骨，全新的架構和實現帶來了飛一般的體驗(JDK7中的ConcurrentHashMap還是采用比較骨板的segment實現的)，細細品讀，還是有不少的收獲。

關于如何進行ConcurrentHashMap內部實現問題的解答就分享到這里了，希望以上內容可以對大家有一定的幫助，如果你還有很多疑惑沒有解開，可以關注億速云行業資訊頻道了解更多相關知識。