ConcurrentHashMap，它在技術面試中出現的頻率相當之高，所以我們必須對它深入理解和掌握。

談到 ConcurrentHashMap，就一定會想到 HashMap。HashMap 在我們的代碼中使用頻率更高，不需要考慮線程安全的地方，我們一般都會使用 HashMap。HashMap 的實現非常經典，如果你讀過 HashMap 的源代碼，那么對 ConcurrentHashMap 源代碼的理解會相對輕松，因為兩者采用的數據結構是類似的

這篇文章主要講解ConcurrentHashMap的核心原理，并注釋詳細源碼，文章篇幅較長，可收藏再看

基本結構

ConcurrentHashMap 是一個存儲 key/value 對的容器，并且是線程安全的。我們先看 ConcurrentHashMap 的存儲結構，如下圖：

雖然 ConcurrentHashMap 的底層數據結構，和方法的實現細節和 HashMap 大體一致，但兩者在類結構上卻沒有任何關聯，我們看下 ConcurrentHashMap 的類圖：

看 ConcurrentHashMap 源碼，我們會發現很多方法和代碼和 HashMap 很相似，有的同學可能會問，為什么不繼承 HashMap 呢？

繼承的確是個好辦法，但ConcurrentHashMap 都是在方法中間進行一些加鎖操作，也就是說加鎖把方法切割了，繼承就很難解決這個問題。

ConcurrentHashMap和HashMap兩者的相同之處：

數組、鏈表結構幾乎相同，所以底層對數據結構的操作思路是相同的（只是思路相同，底層實現不同）；

都實現了 Map 接口，繼承了 AbstractMap 抽象類，所以大多數的方法也都是相同的，HashMap 有的方法，ConcurrentHashMap 幾乎都有，所以當我們需要從 HashMap 切換到 ConcurrentHashMap 時，無需關心兩者之間的兼容問題。

不同之處：

紅黑樹結構略有不同，HashMap 的紅黑樹中的節點叫做 TreeNode，TreeNode 不僅僅有屬性，還維護著紅黑樹的結構，比如說查找，新增等等；ConcurrentHashMap 中紅黑樹被拆分成兩塊，TreeNode 僅僅維護的屬性和查找功能，新增了 TreeBin，來維護紅黑樹結構，并負責根節點的加鎖和解鎖；

新增 ForwardingNode （轉移）節點，擴容的時候會使用到，通過使用該節點，來保證擴容時的線程安全。

這些概念名詞文章后面都會依次介紹

基本構成

重要屬性

我們來看看 ConcurrentHashMap 的幾個重要屬性

//這個Node數組就是ConcurrentHashMap用來存儲數據的哈希表。 transient?volatile?Node[]?table //這是默認的初始化哈希表數組大小 private?static?final?int?DEFAULT_CAPACITY?=?16; //轉化為紅黑樹的鏈表長度閾值 static?final?int?TREEIFY_THRESHOLD?=?8 //這個標識位用于識別擴容時正在轉移數據 static?final?int?MOVED?=?-1 //計算哈希值時用到的參數，用來去除符號位 static?final?int?HASH_BITS?=?0x7fffffff; //數據轉移時，新的哈希表數組 private?transient?volatile?Node[]?nextTable;

重要組成元素

Node

“

鏈表中的元素為Node對象。他是鏈表上的一個節點，內部存儲了key、value值，以及他的下一個節點的引用。這樣一系列的Node就串成一串，組成一個鏈表。

”

ForwardingNode

“

當進行擴容時，要把鏈表遷移到新的哈希表，在做這個操作時，會在把數組中的頭節點替換為ForwardingNode對象。ForwardingNode中不保存key和value，只保存了擴容后哈希表（nextTable）的引用。此時查找相應node時，需要去nextTable中查找。

”

TreeBin

“

當鏈表轉為紅黑樹后，數組中保存的引用為 TreeBin，TreeBin 內部不保存 key/value，他保存了 TreeNode的list以及紅黑樹 root。

”

TreeNode

“

紅黑樹的節點。

”

下面依次講解各個核心方法，有詳細注釋

put方法

public?V?put(K?key,?V?value)?{return?putVal(key,?value,?false); }

ConcurrentHashMap 在 put 方法上的整體思路和 HashMap 相同，但在線程安全方面寫了很多保障的代碼，我們先來看下大體思路：

1.如果數組為空，初始化，初始化完成之后，走 2；

2.計算當前槽點有沒有值，沒有值的話，cas 創建，失敗繼續自旋（for 死循環），直到成功，槽點有值的話，走 3；

3.如果槽點是轉移節點(正在擴容)，就會一直自旋等待擴容完成之后再新增，不是轉移節點走 4；

4.槽點有值的，先鎖定當前槽點，保證其余線程不能操作，如果是鏈表，新增值到鏈表的尾部，如果是紅黑樹，使用紅黑樹新增的方法新增；

5.新增完成之后 check 需不需要擴容，需要的話去擴容。

ConcurrentHashMap在put過程中，采用了哪些手段來保證線程安全呢？

數組初始化時的線程安全

數組初始化時，首先通過自旋來保證一定可以初始化成功，然后通過 CAS 設置 SIZECTL 變量的值，來保證同一時刻只能有一個線程對數組進行初始化，CAS 成功之后，還會再次判斷當前數組是否已經初始化完成，如果已經初始化完成，就不會再次初始化，通過自旋 + CAS + 雙重 check 等手段保證了數組初始化時的線程安全

那么接下來我們就來看看 initTable 方法。

注意里面有個關鍵的值 sizeCtl，這個值有多個含義。

1、-1 代表有線程正在創建 table；

2、-N 代表有 N-1 個線程正在復制 table；

3、在 table 被初始化前，代表根據構造函數傳入的值計算出的應被初始化的大小；

4、在 table 被初始化后，則被設置為 table 大小 的 75%，代表 table 的容量（數組容量）。

新增槽點值時的線程安全

此時為了保證線程安全，做了四處優化：

1.通過自旋死循環保證一定可以新增成功。

在新增之前，通過 for (Node<K,V>[] tab = table;;)這樣的死循環來保證新增一定可以成功，一旦新增成功，就可以退出當前死循環，新增失敗的話，會重復新增的步驟，直到新增成功為止。

2.當前槽點為空時，通過 CAS 新增。

Java 這里的寫法非常嚴謹，沒有在判斷槽點為空的情況下直接賦值，因為在判斷槽點為空和賦值的瞬間，很有可能槽點已經被其他線程賦值了，所以我們采用 CAS 算法，能夠保證槽點為空的情況下賦值成功，如果恰好槽點已經被其他線程賦值，當前 CAS 操作失敗，會再次執行 for 自旋，再走槽點有值的 put 流程，這里就是自旋 + CAS 的結合。

3.當前槽點有值，鎖住當前槽點。

put 時，如果當前槽點有值，就是 key 的 hash 沖突的情況，此時槽點上可能是鏈表或紅黑樹，我們通過鎖住槽點，來保證同一時刻只會有一個線程能對槽點進行修改

V?oldVal?=?null; //鎖定當前槽點，其余線程不能操作，保證了安全 synchronized?(f)?{

4.紅黑樹旋轉時，鎖住紅黑樹的根節點，保證同一時刻，當前紅黑樹只能被一個線程旋轉

Hash算法

spread方法源碼分析

哈希算法的邏輯，決定 ConcurrentHashMap 保存和讀取速度。

static?final?int?spread(int?h)?{return?(h?^?(h?>>>?16))?&?HASH_BITS; }

傳入的參數h為 key 對象的 hashCode，spreed 方法對 hashCode 進行了加工。重新計算出 hash。

hash 值是用來映射該 key 值在哈希表中的位置。取出哈希表中該 hash 值對應位置的代碼如下。

tabAt(tab,?i?=?(n?-?1)?&?hash);

我們先看這一行代碼的邏輯，第一個參數為哈希表，第二個參數是哈希表中的數組下標。通過 (n - 1) & hash 計算下標。n 為數組長度，我們以默認大小 16 為例，那么 n-1 = 15，我們可以假設 hash 值為 100

n的值15轉為二進制： 0000?0000?0000?0000?0000?0000?0000?1111 hash的值100轉為二進制： 0000?0000?0000?0000?0000?0000?0110?0100。 計算結果： 0000?0000?0000?0000?0000?0000?0000?0100 對應的十進制值為?4

15的二進制高位都為0，低位都是1。那么經過&計算后，hash值100的高位全部被清零，低位則保持不變，并且一定是小于（n-1）的。也就是說經過如此計算，通過hash值得到的數組下標絕對不會越界。

這里提出幾個問題：

1、數組大小可以為 17，或者 18 嗎？

2、如果為了保證不越界為什么不直接用 % 計算取余數？

3、為什么不直接用 key 的 hashCode，而是使用經 spreed 方法加工后的 hash 值？

數組大小必須為 2 的 n 次方

第一個問題的答案是數組大小必須為 2 的 n 次方，也就是 16、32、64….不能為其他值。因為如果不是 2 的 n 次方，那么經過計算的數組下標會增大碰撞的幾率

如果hash值的二進制是 10000（十進制16）、10010（十進制18）、10001（十進制17），和10100做&計算后，都是10000，也就是都被映射到數組16這個下標上。這三個值會以鏈表的形式存儲在數組16下標的位置。這顯然不是我們想要的結果。

但如果數組長度n為2的n次方，2進制的數值為10，100，1000，10000……n-1后對應二進制為1，11，111，1111……這樣和hash值低位&后，會保留原來hash值的低位數值，那么只要hash值的低位不一樣，就不會發生碰撞。

同時(n - 1) & hash等價于 hash%n。那么為什么不直接用hash%n呢？

這是因為按位的操作效率會更高。

為什么不直接用 key 的 hashCode？

其實說到底還是為了減少碰撞的概率。我們先看看 spreed 方法中的代碼做了什么事情：

h?^?(h?>>>?16)

這個意思是把 h 的二進制數值向右移動 16 位。我們知道整形為 32 位，那么右移 16 位后，就是把高 16 位移到了低 16 位。而高 16 位清0了。

^為異或操作，二進制按位比較，如果相同則為 0，不同則為 1。這行代碼的意思就是把高低16位做異或。如果兩個hashCode值的低16位相同，但是高位不同，經過如此計算，低16位會變得不一樣了。

為什么要把低位變得不一樣呢？

這是由于哈希表數組長度n會是偏小的數值，那么進行(n - 1) & hash運算時，一直使用的是hash較低位的值。那么即使hash值不同，但如果低位相當，也會發生碰撞。而進行h ^ (h >>> 16)加工后的hash值，讓hashCode高位的值也參與了哈希運算，因此減少了碰撞的概率。

(h?^?(h?>>>?16))?&?HASH_BITS

為何高位移到低位和原來低位做異或操作后，還需要和HASH_BITS這個常量做 & 計算呢？HASH_BITS 這個常量的值為 0x7fffffff，轉化為二進制為 0111 1111 1111 1111 1111 1111 1111 1111。這個操作后會把最高位轉為 0，其實就是消除了符號位，得到的都是正數。這是因為負的 hashCode 在ConcurrentHashMap 中有特殊的含義，因此我們需要得到一個正的 hashCode。

擴容源碼分析

我們大致了解了ConcurrentHashMap 的存儲結構，那么我們思考一個問題，當數組中保存的鏈表越來越多，那么再存儲進來的元素大概率會插入到現有的鏈表中，而不是使用數組中剩下的空位。這樣會造成數組中保存的鏈表越來越長，由此導致哈希表查找速度下降，從 O(1) 慢慢趨近于鏈表的時間復雜度 O(n/2)，這顯然違背了哈希表的初衷。

所以ConcurrentHashMap 會做一個操作，稱為擴容。也就是把數組長度變大，增加更多的空位出來，最終目的就是預防鏈表過長，這樣查找的時間復雜度才會趨向于 O(1)。

擴容的操作并不會在數組沒有空位時才進行，因為在桶位快滿時，新保存元素更大的概率會命中已經使用的位置，那么可能最后幾個桶位很難被使用，而鏈表卻越來越長了。

另外 ConcurrentHashMap 還會有鏈表轉紅黑樹的操作，以提高查找的速度，紅黑樹時間復雜度為 O(logn)，而鏈表是 O(n/2)，因此只在 O(logn)<O(n/2) 時才會進行轉換，也就是以 8 作為分界點。

接下來我們分析 treeifyBin 方法代碼，這個代碼中會選擇是把此時保存數據所在的鏈表轉為紅黑樹，還是對整個哈希表擴容

我們再重點看一下 tryPresize，此方法中實現了對數組的擴容，傳入的參數 size 是原來哈希表大小的一倍。我們假定原來哈希表大小為 16，那么傳入的 size 值為 32

ConcurrentHashMap 的擴容時機和 HashMap 相同，都是在 put 方法的最后一步檢查是否需要擴容，如果需要則進行擴容，但兩者擴容的過程完全不同，ConcurrentHashMap 擴容的方法叫做 transfer，從 put 方法的 addCount 方法進去，就能找到 transfer 方法，transfer 方法的主要思路是：

1.首先需要把老數組的值全部拷貝到擴容之后的新數組上，先從數組的隊尾開始拷貝；

2.拷貝數組的槽點時，先把原數組槽點鎖住，保證原數組槽點不能操作，成功拷貝到新數組時，把原數組槽點賦值為轉移節點；

3.這時如果有新數據正好需要 put 到此槽點時，發現槽點為轉移節點，就會一直等待，所以在擴容完成之前，該槽點對應的數據是不會發生變化的；

4.從數組的尾部拷貝到頭部，每拷貝成功一次，就把原數組中的節點設置成轉移節點；

5.直到所有數組數據都拷貝到新數組時，直接把新數組整個賦值給數組容器，拷貝完成。

擴容方法主要是通過在原數組上設置轉移節點，put 時碰到轉移節點時會等待擴容成功之后才能 put 的策略，來保證了整個擴容過程中肯定是線程安全的，因為數組的槽點一旦被設置成轉移節點，在沒有擴容完成之前，是無法進行操作的

get方法

ConcurrentHashMap 讀的話，就比較簡單，先獲取數組的下標，然后通過判斷數組下標的 key 是否和我們的 key 相等，相等的話直接返回，如果下標的槽點是鏈表或紅黑樹的話，分別調用相應的查找數據的方法，整體思路和 HashMap 很像

構造函數源碼

public?ConcurrentHashMap(int?initialCapacity)?{if?(initialCapacity?<?0)throw?new?IllegalArgumentException();//如果傳入的初始化容量值超過最大容量的一半，那么sizeCtl會被設置為最大容量。//否則通過tableSizeFor方法就算出一個2的n次方數值作為sizeint?cap?=?((initialCapacity?>=?(MAXIMUM_CAPACITY?>>>?1))??MAXIMUM_CAPACITY?:tableSizeFor(initialCapacity?+?(initialCapacity?>>>?1)?+?1));this.sizeCtl?=?cap; }

這是一個有參數的構造方法。如果你對未來存儲的數據量有預估，我們可以指定哈希表的大小，避免頻繁的擴容操作。tableSizeFor 這個方法確保了哈希表的大小永遠都是 2 的 n 次方。

注意這里傳入的參數不是 initialCapacity，而是 initialCapacity 的 1.5 倍 + 1。這樣做是為了保證在默認 75% 的負載因子下，能夠足夠容納 initialCapacity 數量的元素。

ConcurrentHashMap (int initialCapacity) 構造函數總結下：

1、構造函數中并不會初始化哈希表；

2、構造函數中僅設置哈希表大小的變量 sizeCtl；

3、initialCapacity 并不是哈希表大小；

4、哈希表大小為 initialCapacity*1.5+1 后，向上取最小的 2 的 n 次方。如果超過最大容量一半，那么就是最大容量。

tableSizeFor 是如何實現向上取得最接近入參 2 的 n 次方的。下面我們來看 tableSizeFor 源代碼：

private?static?final?int?tableSizeFor(int?c)?{int?n?=?c?-?1;n?|=?n?>>>?1;n?|=?n?>>>?2;n?|=?n?>>>?4;n?|=?n?>>>?8;n?|=?n?>>>?16;return?(n?<?0)???1?:?(n?>=?MAXIMUM_CAPACITY)???MAXIMUM_CAPACITY?:?n?+?1; }

依舊是二進制按位操作，這樣一頓操作后，得到的數值就是大于 c 的最小 2 的 n 次。我們推演下過程，假設 c 是 9：

1、int?n?=?9?-?1 n=8 2、n?|=?n?>>>?1 n=1000 n?>>>?1=0100 兩個值按位或后 n=1100 3、n?|=?n?>>>?2 n=1100 n?>>>?2=0011 n=1111

到這里可以看出規律來了。如果 c 足夠大，使得 n 很大，那么運算到 n |= n >>> 16 時，n 的 32 位都為 1。

總結一下這一段邏輯，其實就是把 n 有數值的 bit 位全部置為 1。這樣就得到了一個肯定大于等于 n 的值。我們再看最后一行代碼，最終返回的是 n+1，那么一個所有位都是 1 的二進制數字，+1 后得到的就是一個 2 的 n 次方數值。

有道無術，術可成；有術無道，止于術

歡迎大家關注Java之道公眾號

好文章，我在看??

總結

以上是生活随笔為你收集整理的ConcurrentHashMap核心原理，这次彻底给整明白了的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。