## 24 經典并發容器，多線程面試必備。—深入解析ConcurrentHashMap > 勤能補拙是良訓，一分辛勞一分才。 > ——華羅庚 本小節我們將學習一個經典的并發容器 ConcurrentHashMap，它在技術面試中出現的頻率相當之高，所以我們必須對它深入理解和掌握。 談到 ConcurrentHashMap，就一定會想到 HashMap。HashMap 在我們的代碼中使用頻率更高，不需要考慮線程安全的地方，我們一般都會使用 HashMap。HashMap 的實現非常經典，如果你讀過 HashMap 的源代碼，那么對 ConcurrentHashMap 源代碼的理解會相對輕松，因為兩者采用的數據結構是類似的。不過即使沒讀過HashMap 源代碼，也不影響本節的學習。 ## 1、ConcurrentHashMap 原理概述 ConcurrentHashMap 是一個存儲 key/value 對的容器，并且是線程安全的。我們先看 ConcurrentHashMap 的存儲結構，如下圖：  這是經典的數組加鏈表的形式。并且在鏈表長度過長時轉化為紅黑樹存儲（ Java 8 的優化），加快查找速度。 存儲結構定義了容器的“形狀”，那容器內的東西按照什么規則來放呢？換句話講，某個 key 是按照什么邏輯放入容器的對應位置呢？ **我們假設要存入的 key 為對象 x，這個過程如下：** 1、通過對象 x 的 hashCode() 方法獲取其 hashCode； 2、將 hashCode 映射到數組的某個位置上； 3、把該元素存儲到該位置的鏈表中。 **從容器取數的邏輯如下：** 1、通過對象 x 的 hashCode() 方法獲取其 hashCode； 2、將 hashCode 映射到數組的某個位置上； 3、遍歷該位置的鏈表或者從紅黑樹中找到匹配的對象返回。 這個數組+鏈表的存儲結構其實是一個哈希表。把對象 x 映射到數組的某個位置的函數，叫做 hash函數。這個函數的好壞決定元素在哈希表中分布是否均勻，如果元素都堆積在一個位置上，那么在取值時需要遍歷很長的鏈表。但元素如果是均勻的分布在數組中，那么鏈表就會較短，通過哈希函數定位位置后，能夠快速找到對應的元素。具體 ConcurrentHashMap 中的哈希函數如何實現我們后面會詳細講到。 **擴容** 我們大致了解了ConcurrentHashMap 的存儲結構，那么我們思考一個問題，當數組中保存的鏈表越來越多，那么再存儲進來的元素大概率會插入到現有的鏈表中，而不是使用數組中剩下的空位。這樣會造成數組中保存的鏈表越來越長，由此導致哈希表查找速度下降，從 O(1) 慢慢趨近于鏈表的時間復雜度 O(n/2)，這顯然違背了哈希表的初衷。所以 ConcurrentHashMap 會做一個操作，稱為擴容。也就是把數組長度變大，增加更多的空位出來，最終目的就是預防鏈表過長，這樣查找的時間復雜度才會趨向于 O(1)。擴容的操作并不會在數組沒有空位時才進行，因為在桶位快滿時，新保存元素更大的概率會命中已經使用的位置，那么可能最后幾個桶位很難被使用，而鏈表卻越來越長了。ConcurrentHashMap 會在更合適的時機進行擴容，通常是在數組中 75% 的位置被使用時。 另外 ConcurrentHashMap 還會有鏈表轉紅黑樹的操作，以提高查找的速度，紅黑樹時間復雜度為 O(logn)，而鏈表是 O(n/2)，因此只在 O(logn)<O(n/2) 時才會進行轉換，也就是以 8 作為分界點。 其實以上內容和 HashMap 類似，ConcurrentHashMap 此外提供了線程安全的保證，它主要是通過 CAS 和Synchronized 關鍵字來實現，我們在源碼分析中再詳細來看。 我們做一下總結： 1、ConcurrentHashMap 采用數組+鏈表+紅黑樹的存儲結構； 2、存入的Key值通過自己的 hashCode 映射到數組的相應位置； 3、ConcurrentHashMap 為保障查詢效率，在特定的時候會對數據增加長度，這個操作叫做擴容； 4、當鏈表長度增加到 8 時，可能會觸發鏈表轉為紅黑樹（數組長度如果小于 64，優先擴容，具體看后面源碼分析）。 接下來，我們的源碼分析就從 ConcurrentHashMap 的構成、保存元素、哈希算法、擴容、查找數據這幾個方面來進行。 ## 2、ConcurrentHashMap 的構成 ### 2.1 重要屬性 我們來看看 ConcurrentHashMap 的幾個重要屬性 **1、transient volatile Node\[\] table** 這個Node數組就是ConcurrentHashMap用來存儲數據的哈希表。 **2、private static final int DEFAULT\_CAPACITY = 16;** 這是默認的初始化哈希表數組大小 **3、static final int TREEIFY\_THRESHOLD = 8** 轉化為紅黑樹的鏈表長度閾值 **4、static final int MOVED = -1** 這個標識位用于識別擴容時正在轉移數據 **5、static final int HASH\_BITS = 0x7fffffff;** 計算哈希值時用到的參數，用來去除符號位 **6、private transient volatile Node\[\] nextTable;** 數據轉移時，新的哈希表數組 可能有些屬性看完解釋你還摸不到頭腦。沒關系，我們在后面源碼分析時，具體使用的地方還會做相應講解。 ### 2.2 重要組成元素 #### Node 鏈表中的元素為Node對象。他是鏈表上的一個節點，內部存儲了key、value值，以及他的下一個節點的引用。這樣一系列的Node就串成一串，組成一個鏈表。 #### ForwardingNode 當進行擴容時，要把鏈表遷移到新的哈希表，在做這個操作時，會在把數組中的頭節點替換為ForwardingNode對象。ForwardingNode中不保存key和value，只保存了擴容后哈希表（nextTable）的引用。此時查找相應node時，需要去nextTable中查找。 #### TreeBin 當鏈表轉為紅黑樹后，數組中保存的引用為 TreeBin，TreeBin 內部不保存 key/value，他保存了 TreeNode的list以及紅黑樹 root。 #### TreeNode 紅黑樹的節點。 ## 3、put 方法源碼分析 put 方法用來把一個鍵值對存儲到map中。代碼如下： ~~~java public V put(K key, V value) { return putVal(key, value, false); } ~~~ 實際調用的是 putVal 方法，第三個參數傳入 false，控制 key 存在時覆蓋原來的值。 我們接下來看 putVal 的代碼，代碼比較多，我把解釋直接放到代碼中： ~~~java final V putVal(K key, V value, boolean onlyIfAbsent) { //key和value不能為空 if (key == null || value == null) throw new NullPointerException(); //計算key的hash值，后面我們會看spread方法的實現 int hash = spread(key.hashCode()); int binCount = 0; //開始自旋，table屬性采取懶加載，第一次put的時候進行初始化 for (Node<K,V>[] tab = table;;) { Node<K,V> f; int n, i, fh; //如果table未被初始化，則初始化table if (tab == null || (n = tab.length) == 0) tab = initTable(); //通過key的hash值映射table位置，如果該位置的值為空，那么生成新的node來存儲該key、value，放入此位置 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 } //如果該位置節點元素的hash值為MOVED，也就是-1，代表正在做擴容的復制。那么該線程參與復制工作。 else if ((fh = f.hash) == MOVED) tab = helpTransfer(tab, f); //下面分支處理table映射的位置已經存在node的情況 else { V oldVal = null; synchronized (f) { //再次確認該位置的值是否已經發生了變化 if (tabAt(tab, i) == f) { //fh大于0，表示該位置存儲的還是鏈表 if (fh >= 0) { binCount = 1; //遍歷鏈表 for (Node<K,V> e = f;; ++binCount) { K ek; //如果存在一樣hash值的node，那么根據onlyIfAbsent的值選擇覆蓋value或者不覆蓋 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; //如果找到最后一個元素，也沒有找到相同hash的node，那么生成新的node存儲key/value，作為尾節點放入鏈表。 if ((e = e.next) == null) { pred.next = new Node<K,V>(hash, key, value, null); break; } } } //下面的邏輯處理鏈表已經轉為紅黑樹時的key/value保存 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; } } } } //node保存完成后，判斷鏈表長度是否已經超出閾值，則進行哈希表擴容或者將鏈表轉化為紅黑樹 if (binCount != 0) { if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); if (oldVal != null) return oldVal; break; } } } //計數，并且判斷哈希表中使用的桶位是否超出閾值，超出的話進行擴容 addCount(1L, binCount); return null; } ~~~ 主線流程梳理如下圖：  其實 put 的核心思想都在這里了。接下來我們分別看一下關鍵節點的方法源碼。 ## 4、spread 方法源碼分析 哈希算法的邏輯，決定 ConcurrentHashMap 保存和讀取速度。hash 算法是 hashmap 的核心算法，JDK 的實現十分巧妙，值得我們學習。 spreed 方法源代碼如下： ~~~java static final int spread(int h) { return (h ^ (h >>> 16)) & HASH_BITS; } ~~~ 傳入的參數h為 key 對象的 hashCode，spreed 方法對 hashCode 進行了加工。重新計算出 hash。我們先暫不分析這一行代碼的邏輯，先繼續往下看如何使用此 hash 值。 hash 值是用來映射該 key 值在哈希表中的位置。取出哈希表中該 hash 值對應位置的代碼如下。 ~~~java tabAt(tab, i = (n - 1) & hash) ~~~ 我們先看這一行代碼的邏輯，第一個參數為哈希表，第二個參數是哈希表中的數組下標。通過 (n - 1) & hash 計算下標。n 為數組長度，我們以默認大小 16 為例，那么 n-1 = 15，我們可以假設 hash 值為 100，那么 15 & 100為多少呢？& 把它左右數值轉化為二進制，按位進行與操作，只有兩個值都為 1 才為 1，有一個為 0 則為 0。那么我們把 15 和 100 轉化為二進制來計算，java中 int 類型為 8 個字節，一共 32 個bit位。 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 值？ 這幾個問題是面試經常會問到的相關問題。我們一個個來解答。 ### 4.1 數組大小必須為 2 的 n 次方 第一個問題的答案是數組大小必須為 2 的 n 次方，也就是 16、32、64….不能為其他值。因為如果不是 2 的 n 次方，那么經過計算的數組下標會增大碰撞的幾率，例如數組長度為 21，那么 n-1=20，對應的二進制為： 10100 那么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值的低位不一樣，就不會發生碰撞。 其實如果數組長度為 2 的 n 次方，那么 (n - 1) & hash 等價于 hash%n。那么為什么不直接用hash%n呢？這是因為按位的操作效率會更高，經過我本地測試，& 計算速度大概是 % 操作的 50 倍左右。 所以 JDK 為了性能，而使用這種巧妙的算法，在確保元素均勻分布的同時，還保證了效率。 ### 4.2 為什么不直接用 key 的 hashCode？ 本來我們要分析 spreed 方法的代碼，但是現在看起來這個方法好像并沒有什么用處，直接用 key 的 hashCode來定位哈希表的位置就可以了啊，為什么還要經過 spreed 方法的加工呢？其實說到底還是為了減少碰撞的概率。我們先看看 spreed 方法中的代碼做了什么事情： **1、h ^ (h >>> 16)** 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高位的值也參與了哈希運算，因此減少了碰撞的概率。 **2、(h ^ (h >>> 16)) & HASH\_BITS** 我們再看完整的代碼，為何高位移到低位和原來低位做異或操作后，還需要和HASH\_BITS這個常量做 & 計算呢？HASH\_BITS 這個常量的值為 0x7fffffff，轉化為二進制為 0111 1111 1111 1111 1111 1111 1111 1111。這個操作后會把最高位轉為 0，其實就是消除了符號位，得到的都是正數。這是因為負的 hashCode 在ConcurrentHashMap 中有特殊的含義，因此我們需要得到一個正的 hashCode。 ## 總結 通過以上分析我們已經清楚 ConcurrentHashMap 中是如何通過 Hash 值來映射數組位置的，這里面的算法設計確實十分的巧妙。可能平時我們編碼用不到，但也要熟記于心，相信在面試中一定用得到。下面一節我們再來看看table 初始化以及擴容的相關內容，put 方法的主要內容就是這些，下節最后再看一下 get 方法。