經典面試題之ConcurrentHashMap

一 ConcurrentHashMap 與 HashMap的區別？

• ConcurrentHashMap線程安全，而HashMap非線程安全
• HashMap允許Key和Value為null，而ConcurrentHashMap不允許
• HashMap不允許通過Iterator遍歷的同時通過HashMap修改，而ConcurrentHashMap允許該行為，並且該更新對後續的遍歷可見

以上說的比較籠統，我們具體看一下ConcurrentHashMap：

先來看下ConcurrentHashMap的數據結構

1.8之前的 ConcurrentHashMap是在1.7HashMap的基礎上實現了線程安全的版本。採用分段鎖的概念，使鎖更加細化。它默認將Hash表分為16個分段，segments數組的長度最大為65536，最大容量 1 << 30。

JDK1.8 的實現已經摒棄了 Segment 的概念，而是直接用 Node 數組 + 鏈表 + 紅黑樹的數據結構來實現，併發控制使用 Synchronized 和 CAS 來操作，整個看起來就像是優化過且線程安全的 HashMap，雖然在 JDK1.8 中還能看到 Segment 的數據結構，但是已經簡化了屬性，只是為了兼容舊版本。

二 concurrentHashMap最大容量？

<code>
/**
     * The largest possible table capacity.  This value must be
     * exactly 1<<30 to stay within Java array allocation and indexing
     * bounds for power of two table sizes, and is further required
     * because the top two bits of 32bit hash fields are used for
     * control purposes.
     */
    private static final int MAXIMUM_CAPACITY = 1 << 30;/<code>

注意這是 The largest possible table capacity，它是否代表最多能存儲到map中的元素數量？答案是否定的。至於為什麼，作為思考題，留給你。（關於這個問題在前一個系列關於HashMap的文章中也提到過相似的問題）

提示 看一下size方法，為什麼n要設計為long？實際元素數量和返回值一樣嗎？

<code>public int size() {
        long n = sumCount();
        return ((n < 0L) ? 0 :
                (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
                (int)n);
    }/<code>

三 ConcurrentHashMap 也會出現死循環？

是的，當你不當地使用computeIfAbsent 方法時

<code>/**
     * If the specified key is not already associated with a value,
     * attempts to compute its value using the given mapping function
     * and enters it into this map unless {@code null}.  The entire
     * method invocation is performed atomically, so the function is
     * applied at most once per key.  Some attempted update operations 

     * on this map by other threads may be blocked while computation
     * is in progress, so the computation should be short and simple,
     * and must not attempt to update any other mappings of this map.

/<code>

上面的computeIfAbsent 方法註釋也得很清楚了，應該絕對避免在computeIfAbsent中有遞歸，或者修改map的任何操作。所以如果你在調用此方法並有上述操作時就會出現死循環問題。至於為什麼會出現這種問題，有興趣的可以讀讀其他資料或源代碼，本文就不詳述了。好在這個問題在java 1.9中已經基本修復了。

問題如何規避？既然官方給出這麼強烈的提示了，不作死就不會死。或者升級到JDK1.9

四 ConcurrentHashMap 在 JDK 1.8 中，為什麼要使用內置鎖 synchronized 來代替重入鎖 ReentrantLock？

• 粒度降低了(看下圖感覺下鎖粒度的變化)
• JVM 開發團隊沒有放棄 synchronized，而且基於 JVM 的 synchronized 優化空間更大，更加自然。
• 在大量的數據操作下，對於 JVM 的內存壓力，基於 API 的 ReentrantLock 會開銷更多的內存。

JDK1.8的ConcurrentHashMap（TreeBin: 紅黑二叉樹節點

Node: 鏈表節點）

五 put() 方法流程？

<code>/**     * The largest possible table capacity.  This value must be     * exactly 1<<30 to stay within Java array allocation and indexing     * bounds for power of two table sizes, and is further required     * because the top two bits of 32bit hash fields are used for     * control purposes.     */    private static final int MAXIMUM_CAPACITY = 1 << 30;/<code>

1. 如果沒有初始化，就調用 initTable() 方法來進行初始化；
2. 如果沒有 hash 衝突就直接 CAS 無鎖插入；
3. 如果需要擴容，就先進行擴容；
4. 如果存在 hash 衝突，就加鎖來保證線程安全，兩種情況：一種是鏈表形式就直接遍歷到尾端插入，一種是紅黑樹就按照紅黑樹結構插入；
5. 如果該鏈表的數量大於閥值 8，就要先轉換成紅黑樹的結構，break 再一次進入循環
6. 如果添加成功就調用 addCount() 方法統計 size，並且檢查是否需要擴容。

• 擴容方法 transfer()：默認容量為 16，擴容時，容量變為原來的兩倍。
• helpTransfer()：調用多個工作線程一起幫助進行擴容，這樣的效率就會更高。

六 ConcurrentHashMap 的併發度是什麼？

程序運行時能夠同時更新 ConccurentHashMap 且不產生鎖競爭的最大線程數。默認為 16，且可以在構造函數中設置。當用戶設置併發度時，ConcurrentHashMap 會使用大於等於該值的最小2冪指數作為實際併發度（假如用戶設置併發度為17，實際併發度則為32）

七 ConcurrentHashMap的get方法是否要加鎖，為什麼？

不需要。get沒有加鎖的話，ConcurrentHashMap是如何保證讀到的數據不是髒數據的呢？

get操作全程不需要加鎖是因為Node的成員val是用volatile修飾的。

八 ConcurrentHashMap 如何計算 size

size()方法返回的是一個不精確的值

我們先來看一下jdk1.8的代碼註釋：

大致的意思是：返回容器的大小。這個方法應該被用來代替size()方法，因為 ConcurrentHashMap的容量大小可能會大於int的最大值。返回的值是一個估計值;如果有併發插入或者刪除操作，則實際的數量可能有所不同。

<code>/**
     * Returns the number of mappings. This method should be used
     * instead of {@link #size} because a ConcurrentHashMap may
     * contain more mappings than can be represented as an int. The
     * value returned is an estimate; the actual count may differ if
     * there are concurrent insertions or removals.
     *（大致的意思是：返回容器的大小。這個方法應該被用來代替size()方法，因為
     * ConcurrentHashMap的容量大小可能會大於int的最大值。
     * 返回的值是一個估計值;如果有併發插入或者刪除操作，則實際的數量可能有所不同。）
     * @return the number of mappings
     * @since 1.8
     */
    public long mappingCount() {
        long n = sumCount();
        return (n < 0L) ? 0L : n; // ignore transient negative values
    }/<code>

1.7中 Segment繼承ReentrantLock，這樣就很容易對每個Segment加鎖了。類似於get或remove這些操作，都只需要在操作前對一個Segment加鎖。但是有些操作需要跨段，比如size()、containsValue()和isEmpty()方法，因此為了保證併發效率，允許size返回的是一個近似值而不是精確值。

1.7的 put、remove和get操作只需要關心一個Segment，而size操作需要遍歷所有的Segment才能算出整個Map的大小。一個簡單的方案是，先鎖住所有Sgment，計算完後再解鎖。但這樣做，在做size操作時，不僅無法對Map進行寫操作，同時也無法進行讀操作，不利於對Map的並行操作。為更好支持併發操作，

jdk 1.8 put方法和remove方法都會通過addCount方法維護Map的size。size方法通過sumCount獲取由addCount方法維護的Map的size。

<code>final long sumCount() {
        CounterCell[] as = counterCells; CounterCell a;
        long sum = baseCount;
        if (as != null) {
            for (int i = 0; i < as.length; ++i) {
                if ((a = as[i]) != null)
                    sum += a.value;
            }
        }
        return sum;
    }


 private final void addCount(long x, int check) {
        CounterCell[] as; long b, s;
        if ((as = counterCells) != null ||
            !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
            CounterCell a; long v; int m;
            boolean uncontended = true;
            if (as == null || (m = as.length - 1) < 0 ||
                (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
                !(uncontended =
                  U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
                fullAddCount(x, uncontended);
                return;
            }
            if (check <= 1)
                return; 

            s = sumCount();
        }/<code>

注意兩個屬性 ：baseCount 和 counterCells。

• baseCount 一個 volatile 的變量，在 addCount 方法中會使用它，而 addCount 方法在 put 結束後會調用。在 addCount 方法中，會對這個變量做 CAS 加法。
• counterCells 一種用於分配計數的填充單元。改編自LongAdder和Striped64

總結

• JDK1.7 和 JDK1.8 對 size 的計算是不一樣的。1.7 中是先不加鎖計算三次，如果三次結果不一樣在加鎖
• JDK1.8 size 是通過對 baseCount 和 counterCell 進行 CAS 計算，最終通過 baseCount 和 遍歷 CounterCell 數組得出 size。
• JDK 8 推薦使用mappingCount 方法，因為這個方法的返回值是 long 類型，不會因為 size 方法是 int 類型限制最大值。

九 用了ConcurrentHashMap 就一定是線程安全的嗎？

不一定，ConcurrenetHashMap 只能保證提供的原子性讀寫操作是線程安全的，換句話說，如果你的讀寫操作不是原子性的，那麼無法保證絕對的線程安全。如果你希望在一整段業務邏輯中，對容器的操作都保持整體一致性的話，需要另外加鎖處理。

ConcurrentHashMap 對外提供的方法或能力的限制：

• 使用了 ConcurrentHashMap，不代表對它的多個操作之間的狀態是一致的，是沒有其他線程在操作它的，如果需要確保需要手動加鎖。
• 諸如 size、isEmpty 和 containsValue 等聚合方法，在併發情況下可能會反映 ConcurrentHashMap 的中間狀態。因此在併發情況下，這些方法的返回值只能用作參考，而不能用於流程控制。
• 諸如 putAll 這樣的聚合方法也不能確保原子性，在 putAll 的過程中去獲取數據可能會獲取到部分數據。

閱讀更多 小盒子的技術分享 的文章

關鍵字: 跳槽那些事兒 數據結構 面試題