## ConcurrentHashMap [TOC] ### 1\. 概述　　 　　眾所周知，哈希表是中非常高效，復雜度為 O(1) 的數據結構，在 Java 開發中，我們最常見到最頻繁使用的就是 HashMap 和 HashTable，但是在線程競爭激烈的并發場景中使用都不夠合理。 　　**HashMap**：先說 HashMap，HashMap 是**線程不安全**的，在并發環境下，可能會形成**環狀鏈表**（擴容時可能造成），導致 get 操作時，cpu 空轉，所以，在并發環境中使 用HashMap 是非常危險的。 　　**HashTable**： HashTable 和 HashMap的實現原理幾乎一樣，差別無非是：（1）HashTable不允許key和value為null；（2）HashTable是線程安全的。 　　但是 HashTable 線程安全的策略實現代價卻太大了，簡單粗暴，get/put 所有相關操作都是 synchronized 的，這相當于給整個哈希表加了一把大鎖，多線程訪問時候，只要有一個線程訪問或操作該對象，那其他線程只能阻塞，相當于將所有的操作串行化，在競爭激烈的并發場景中性能就會非常差。 :-:  　　HashTable 性能差主要是由于所有操作需要競爭同一把鎖，而如果容器中有多把鎖，每一把鎖鎖一段數據，這樣在多線程訪問時不同段的數據時，就不會存在鎖競爭了，這樣便可以有效地提高并發效率。這就是ConcurrentHashMap 所采用的 "**分段鎖**" 思想。  ### 2\. 存儲結構 ConcurrentHashMap 采用了非常精妙的"分段鎖"策略，ConcurrentHashMap 的主干是個 Segment 數組。 ~~~java final Segment<K,V>[] segments; ~~~ 　　Segment 繼承了 ReentrantLock，所以它就是一種可重入鎖（ReentrantLock)。在 ConcurrentHashMap，一個 Segment 就是一個子哈希表，Segment 里維護了一個 HashEntry 數組，并發環境下，對于不同 Segment 的數據進行操作是不用考慮鎖競爭的。（就按默認的 ConcurrentLeve 為16來講，理論上就允許 16 個線程并發執行，有木有很酷） 　　**所以，對于同一個 Segment 的操作才需考慮線程同步，不同的 Segment 則無需考慮。** Segment 類似于 HashMap，一個 Segment 維護著一個 HashEntry 數組 ~~~java transient volatile HashEntry<K,V>[] table; ~~~ HashEntry 是目前我們提到的最小的邏輯處理單元了。一個 ConcurrentHashMap 維護一個 Segment 數組，一個 Segment 維護一個 HashEntry 數組。 ~~~java static final class HashEntry<K,V> { final int hash; final K key; volatile V value; volatile HashEntry<K,V> next; } ~~~ 　　ConcurrentHashMap 和 HashMap 實現上類似，最主要的差別是 ConcurrentHashMap 采用了分段鎖（Segment），每個分段鎖維護著幾個桶（HashEntry），多個線程可以同時訪問不同分段鎖上的桶，從而使其并發度更高（并發度就是 Segment 的個數）。 Segment 繼承自**ReentrantLock**。 ~~~java static final class Segment<K,V> extends ReentrantLock implements Serializable { private static final long serialVersionUID = 2249069246763182397L; static final int MAX_SCAN_RETRIES = Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1; transient volatile HashEntry<K,V>[] table; transient int count; transient int modCount; transient int threshold; final float loadFactor; } ~~~ ~~~java final Segment<K,V>[] segments; ~~~ 默認的并發級別為 16，也就是說默認創建 16 個 Segment。 ~~~java static final int DEFAULT_CONCURRENCY_LEVEL = 16; ~~~ ### 2\. size 操作 每個 Segment 維護了一個 count 變量來統計該 Segment 中的鍵值對個數。 ~~~java /** * The number of elements. Accessed only either within locks * or among other volatile reads that maintain visibility. */ transient int count; ~~~ 在執行 size 操作時，需要遍歷所有 Segment 然后把 count 累計起來。 ConcurrentHashMap 在執行 size 操作時先嘗試不加鎖，如果連續兩次不加鎖操作得到的結果一致，那么可以認為這個結果是正確的。 嘗試次數使用 RETRIES\_BEFORE\_LOCK 定義，該值為 2，retries 初始值為 -1，因此嘗試次數為 3。 如果嘗試的次數超過 3 次，就需要對每個 Segment 加鎖。 ~~~java /** * Number of unsynchronized retries in size and containsValue * methods before resorting to locking. This is used to avoid * unbounded retries if tables undergo continuous modification * which would make it impossible to obtain an accurate result. */ static final int RETRIES_BEFORE_LOCK = 2; public int size() { // Try a few times to get accurate count. On failure due to // continuous async changes in table, resort to locking. final Segment<K,V>[] segments = this.segments; int size; boolean overflow; // true if size overflows 32 bits long sum; // sum of modCounts long last = 0L; // previous sum int retries = -1; // first iteration isn't retry try { for (;;) { // 超過嘗試次數，則對每個 Segment 加鎖 if (retries++ == RETRIES_BEFORE_LOCK) { for (int j = 0; j < segments.length; ++j) ensureSegment(j).lock(); // force creation } sum = 0L; size = 0; overflow = false; for (int j = 0; j < segments.length; ++j) { Segment<K,V> seg = segmentAt(segments, j); if (seg != null) { sum += seg.modCount; int c = seg.count; if (c < 0 || (size += c) < 0) overflow = true; } } // 連續兩次得到的結果一致，則認為這個結果是正確的 if (sum == last) break; last = sum; } } finally { if (retries > RETRIES_BEFORE_LOCK) { for (int j = 0; j < segments.length; ++j) segmentAt(segments, j).unlock(); } } return overflow ? Integer.MAX_VALUE : size; } ~~~ ### 3\. 同步方式 Segment 繼承自 ReentrantLock，所以我們可以很方便的對每一個 Segment 上鎖。 對于讀操作，獲取 Key 所在的 Segment 時，需要保證可見性。具體實現上可以使用 volatile 關鍵字，也可使用鎖。但使用鎖開銷太大，而使用 volatile 時每次寫操作都會讓所有 CPU 內緩存無效，也有一定開銷。ConcurrentHashMap 使用如下方法保證可見性，取得最新的 Segment。 ~~~java Segment<K,V> s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u) ~~~ 獲取 Segment 中的 HashEntry 時也使用了類似方法 ~~~java HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile (tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE) ~~~ 對于寫操作，并不要求同時獲取所有 Segment 的鎖，因為那樣相當于鎖住了整個 Map。它會先獲取該 Key-Value 對所在的 Segment 的鎖，獲取成功后就可以像操作一個普通的 HashMap 一樣操作該 Segment，并保證該Segment 的安全性。 同時由于其它 Segment 的鎖并未被獲取，因此理論上可支持 concurrencyLevel（等于 Segment 的個數）個線程安全的并發讀寫。 獲取鎖時，并不直接使用 lock 來獲取，因為該方法獲取鎖失敗時會掛起。事實上，它使用了自旋鎖，如果 tryLock 獲取鎖失敗，說明鎖被其它線程占用，此時通過循環再次以 tryLock 的方式申請鎖。如果在循環過程中該 Key 所對應的鏈表頭被修改，則重置 retry 次數。如果 retry 次數超過一定值，則使用 lock 方法申請鎖。 這里使用自旋鎖是因為自旋鎖的效率比較高，但是它消耗 CPU 資源比較多，因此在自旋次數超過閾值時切換為互斥鎖。 ### 4\. JDK 1.8 的改動 * JDK 1.7 使用分段鎖機制來實現并發更新操作，核心類為 Segment，它繼承自重入鎖 ReentrantLock，并發程度與 Segment 數量相等。 * JDK 1.8 使用了 CAS 操作來支持更高的并發度，在 CAS 操作失敗時使用內置鎖 synchronized。 * 并且 JDK 1.8 的實現也在鏈表過長時會轉換為紅黑樹。 參考資料： * [ConcurrentHashMap演進從Java7到Java8](http://www.jasongj.com/java/concurrenthashmap/) * [ConcurrentHashMap實現原理及源碼分析 - dreamcatcher-cx - 博客園](https://www.cnblogs.com/chengxiao/p/6842045.html)