xylz,imxylz

在上一篇中介紹了HashMap的原理，這一節是ConcurrentMap的最后一節，所以會完整的介紹ConcurrentHashMap的實現。

ConcurrentHashMap原理

在讀寫鎖章節部分介紹過一種是用讀寫鎖實現Map的方法。此種方法看起來可以實現Map響應的功能，而且吞吐量也應該不錯。但是通過前面對讀寫鎖原理的分析后知道，讀寫鎖的適合場景是讀操作>>寫操作，也就是讀操作應該占據大部分操作，另外讀寫鎖存在一個很嚴重的問題是讀寫操作不能同時發生。要想解決讀寫同時進行問題（至少不同元素的讀寫分離），那么就只能將鎖拆分，不同的元素擁有不同的鎖，這種技術就是“鎖分離”技術。

默認情況下ConcurrentHashMap是用了16個類似HashMap 的結構，其中每一個HashMap擁有一個獨占鎖。也就是說最終的效果就是通過某種Hash算法，將任何一個元素均勻的映射到某個HashMap的Map.Entry上面，而對某個一個元素的操作就集中在其分布的HashMap上，與其它HashMap無關。這樣就支持最多16個并發的寫操作。

上圖就是ConcurrentHashMap的類圖。參考上面的說明和HashMap的原理分析，可以看到ConcurrentHashMap將整個對象列表分為segmentMask+1個片段（Segment）。其中每一個片段是一個類似于HashMap的結構，它有一個HashEntry的數組，數組的每一項又是一個鏈表，通過HashEntry的next引用串聯起來。

這個類圖上面的數據結構的定義非常有學問，接下來會一個個有針對性的分析。

首先如何從ConcurrentHashMap定位到HashEntry。在HashMap的原理分析部分說過，對于一個Hash的數據結構來說，為了減少浪費的空間和快速定位數據，那么就需要數據在Hash上的分布比較均勻。對于一次Map的查找來說，首先就需要定位到Segment，然后從過Segment定位到HashEntry鏈表，最后才是通過遍歷鏈表得到需要的元素。

在不討論并發的前提下先來討論如何定位到HashEntry的。在ConcurrentHashMap中是通過hash(key.hashCode())和segmentFor(hash)來得到Segment的。清單1 描述了如何定位Segment的過程。其中hash(int)是將key的hashCode進行二次編碼，使之能夠在segmentMask+1個Segment上均勻分布（默認是16個）。可以看到的是這里和HashMap還是有點不同的，這里采用的算法叫Wang/Jenkins hash，有興趣的可以參考資料1和參考資料2。總之它的目的就是使元素能夠均勻的分布在不同的Segment上，這樣才能夠支持最多segmentMask+1個并發，這里segmentMask+1是segments的大小。

清單1 定位Segment

private static int hash(int h) {
    // Spread bits to regularize both segment and index locations,
    // using variant of single-word Wang/Jenkins hash.
    h += (h <<  15) ^ 0xffffcd7d;
    h ^= (h >>> 10);
    h += (h <<   3);
    h ^= (h >>>  6);
    h += (h <<   2) + (h << 14);
    return h ^ (h >>> 16);
}
final Segment<K,V> segmentFor(int hash) {
    return segments[(hash >>> segmentShift) & segmentMask];
}

顯然在不能夠對Segment擴容的情況下，segments的大小就應該是固定的。所以在ConcurrentHashMap中segments/segmentMask/segmentShift都是常量，一旦初始化后就不能被再次修改，其中segmentShift是查找Segment的一個常量偏移量。

有了Segment以后再定位HashEntry就和HashMap中定位HashEntry一樣了，先將hash值與Segment中HashEntry的大小減1進行與操作定位到HashEntry鏈表，然后遍歷鏈表就可以完成相應的操作了。

能夠定位元素以后ConcurrentHashMap就已經具有了HashMap的功能了，現在要解決的就是如何并發的問題。要解決并發問題，加鎖是必不可免的。再回頭看Segment的類圖，可以看到Segment除了有一個volatile類型的元素大小count外，Segment還是集成自ReentrantLock的。另外在前面的原子操作和鎖機制中介紹過，要想最大限度的支持并發，那么能夠利用的思路就是盡量讀操作不加鎖，寫操作不加鎖。如果是讀操作不加鎖，寫操作加鎖，對于競爭資源來說就需要定義為volatile類型的。volatile類型能夠保證happens-before法則，所以volatile能夠近似保證正確性的情況下最大程度的降低加鎖帶來的影響，同時還與寫操作的鎖不產生沖突。

同時為了防止在遍歷HashEntry的時候被破壞，那么對于HashEntry的數據結構來說，除了value之外其他屬性就應該是常量，否則不可避免的會得到ConcurrentModificationException。這就是為什么HashEntry數據結構中key,hash,next是常量的原因(final類型）。

有了上面的分析和條件后再來看Segment的get/put/remove就容易多了。

get操作

清單2 Segment定位元素

V get(Object key, int hash) {
    if (count != 0) { // read-volatile
        HashEntry<K,V> e = getFirst(hash);
        while (e != null) {
            if (e.hash == hash && key.equals(e.key)) {
                V v = e.value;
                if (v != null)
                    return v;
                return readValueUnderLock(e); // recheck
            }
            e = e.next;
        }
    }
    return null;
}
HashEntry<K,V> getFirst(int hash) {
    HashEntry<K,V>[] tab = table;
    return tab[hash & (tab.length - 1)];
}

V readValueUnderLock(HashEntry<K,V> e) {
    lock();
    try {
        return e.value;
    } finally {
        unlock();
    }
}

清單2 描述的是Segment如何定位元素。首先判斷Segment的大小count>0，Segment的大小描述的是HashEntry不為空(key不為空)的個數。如果Segment中存在元素那么就通過getFirst定位到指定的HashEntry鏈表的頭節點上，然后遍歷此節點，一旦找到key對應的元素后就返回其對應的值。但是在清單2 中可以看到拿到HashEntry的value后還進行了一次判斷操作，如果為空還需要加鎖再讀取一次（readValueUnderLock）。為什么會有這樣的操作？盡管ConcurrentHashMap不允許將value為null的值加入，但現在仍然能夠讀到一個為空的value就意味著此值對當前線程還不可見（這是因為HashEntry還沒有完全構造完成就賦值導致的，后面還會談到此機制）。

put操作

清單3 描述的是Segment的put操作。首先就需要加鎖了，修改一個競爭資源肯定是要加鎖的，這個毫無疑問。需要說明的是Segment集成的是ReentrantLock，所以這里加的鎖也就是獨占鎖，也就是說同一個Segment在同一時刻只有能一個put操作。

接下來來就是檢查是否需要擴容，這和HashMap一樣，如果需要的話就擴大一倍，同時進行rehash操作。

查找元素就和get操作是一樣的，得到元素就直接修改其值就好了。這里onlyIfAbsent只是為了實現ConcurrentMap的putIfAbsent操作而已。需要說明以下幾點：

• 如果找到key對于的HashEntry后直接修改就好了，如果找不到那么就需要構造一個新的HashEntry出來加到hash對于的HashEntry的頭部，同時就的頭部就加到新的頭部后面。這是因為HashEntry的next是final類型的，所以只能修改頭節點才能加元素加入鏈表中。
• 如果增加了新的操作后，就需要將count+1寫回去。前面說過count是volatile類型，而讀取操作沒有加鎖，所以只能把元素真正寫回Segment中的時候才能修改count值，這個要放到整個操作的最后。
• 在將新的HashEntry寫入table中時是通過構造函數來設置value值的，這意味對table的賦值可能在設置value之前，也就是說得到了一個半構造完的HashEntry。這就是重排序可能引起的問題。所以在讀取操作中，一旦讀到了一個value為空的value是就需要加鎖重新讀取一次。為什么要加鎖？加鎖意味著前一個寫操作的鎖釋放，也就是前一個鎖的數據已經完成寫完了了，根據happens-before法則，前一個寫操作的結果對當前讀線程就可見了。當然在JDK 6.0以后不一定存在此問題。
• 在Segment中table變量是volatile類型，多次讀取volatile類型的開銷要不非volatile開銷要大，而且編譯器也無法優化，所以在put操作中首先建立一個臨時變量tab指向table，多次讀寫tab的效率要比volatile類型的table要高，JVM也能夠對此進行優化。

清單3 Segment的put操作

V put(K key, int hash, V value, boolean onlyIfAbsent) {
    lock();
    try {
        int c = count;
        if (c++ > threshold) // ensure capacity
            rehash();
        HashEntry<K,V>[] tab = table;
        int index = hash & (tab.length - 1);
        HashEntry<K,V> first = tab[index];
        HashEntry<K,V> e = first;
        while (e != null && (e.hash != hash || !key.equals(e.key)))
            e = e.next;

        V oldValue;
        if (e != null) {
            oldValue = e.value;
            if (!onlyIfAbsent)
                e.value = value;
        }
        else {
            oldValue = null;
            ++modCount;
            tab[index] = new HashEntry<K,V>(key, hash, first, value);
            count = c; // write-volatile
        }
        return oldValue;
    } finally {
        unlock();
    }
}

remove 操作

清單4 描述了Segment刪除一個元素的過程。同put一樣，remove也需要加鎖，這是因為對table可能會有變更。由于HashEntry的next節點是final類型的，所以一旦刪除鏈表中間一個元素，就需要將刪除之前或者之后的元素重新加入新的鏈表。而Segment采用的是將刪除元素之前的元素一個個重新加入刪除之后的元素之前（也就是鏈表頭結點）來完成新鏈表的構造。

清單4 Segment的remove操作

V remove(Object key, int hash, Object value) {
    lock();
    try {
        int c = count - 1;
        HashEntry<K,V>[] tab = table;
        int index = hash & (tab.length - 1);
        HashEntry<K,V> first = tab[index];
        HashEntry<K,V> e = first;
        while (e != null && (e.hash != hash || !key.equals(e.key)))
            e = e.next;

        V oldValue = null;
        if (e != null) {
            V v = e.value;
            if (value == null || value.equals(v)) {
                oldValue = v;
                // All entries following removed node can stay
                // in list, but all preceding ones need to be
                // cloned.
                ++modCount;
                HashEntry<K,V> newFirst = e.next;
                for (HashEntry<K,V> p = first; p != e; p = p.next)
                    newFirst = new HashEntry<K,V>(p.key, p.hash,
                                                  newFirst, p.value);
                tab[index] = newFirst;
                count = c; // write-volatile
            }
        }
        return oldValue;
    } finally {
        unlock();
    }
}

下面的示意圖描述了如何刪除一個已經存在的元素的。假設我們要刪除B3元素。首先定位到B3所在的Segment，然后再定位到Segment的table中的B1元素，也就是Bx所在的鏈表。然后遍歷鏈表找到B3，找到之后就從頭結點B1開始構建新的節點B1（藍色）加到B4的前面，繼續B1后面的節點B2構造B2（藍色），加到由藍色的B1和B4構成的新的鏈表。繼續下去，直到遇到B3后終止，這樣就構造出來一個新的鏈表B2（藍色）->B1（藍色）->B4->B5，然后將此鏈表的頭結點B2（藍色）設置到Segment的table中。這樣就完成了元素B3的刪除操作。需要說明的是，盡管就的鏈表仍然存在(B1->B2->B3->B4->B5)，但是由于沒有引用指向此鏈表，所以此鏈表中無引用的（B1->B2->B3）最終會被GC回收掉。這樣做的一個好處是，如果某個讀操作在刪除時已經定位到了舊的鏈表上，那么此操作仍然將能讀到數據，只不過讀取到的是舊數據而已，這在多線程里面是沒有問題的。

除了對單個元素操作外，還有對全部的Segment的操作，比如size()操作等。

size操作

size操作涉及到統計所有Segment的大小，這樣就會遍歷所有的Segment，如果每次加鎖就會導致整個Map都被鎖住了，任何需要鎖的操作都將無法進行。這里用到了一個比較巧妙的方案解決此問題。

在Segment中有一個變量modCount，用來記錄Segment結構變更的次數，結構變更包括增加元素和刪除元素，每增加一個元素操作就+1，每進行一次刪除操作+1，每進行一次清空操作(clear)就+1。也就是說每次涉及到元素個數變更的操作modCount都會+1，而且一直是增大的，不會減小。

遍歷兩次ConcurrentHashMap中的segments，每次遍歷是記錄每一個Segment的modCount，比較兩次遍歷的modCount值的和是否相同，如果相同就返回在遍歷過程中獲取的Segment的count的和，也就是所有元素的個數。如果不相同就重復再做一次。重復一次還不相同就將所有Segment鎖住，一個一個的獲取其大小(count)，最后將這些count加起來得到總的大小。當然了最后需要將鎖一一釋放。清單5 描述了這個過程。

這里有一個比較高級的話題是為什么在讀取modCount的時候總是先要讀取count一下。為什么不是先讀取modCount然后再讀取count的呢？也就是說下面的兩條語句能否交換下順序？

sum += segments[i].count;

mcsum += mc[i] = segments[i].modCount;

答案是不能！為什么？這是因為modCount總是在加鎖的情況下才發生變化，所以不會發生多線程同時修改的情況，也就是沒必要時volatile類型。另外總是在count修改的情況下修改modCount，而count是一個volatile變量。于是這里就充分利用了volatile的特性。

根據happens-before法則，第（3）條：對volatile字段的寫入操作happens-before于每一個后續的同一個字段的讀操作。也就是說一個操作C在volatile字段的寫操作之后，那么volatile寫操作之前的所有操作都對此操作C可見。所以修改modCount總是在修改count之前，也就是說如果讀取到了一個count的值，那么在count變化之前的modCount也就能夠讀取到，換句話說就是如果看到了count值的變化，那么就一定看到了modCount值的變化。而如果上面兩條語句交換下順序就無法保證這個結果一定存在了。

在ConcurrentHashMap.containsValue中，可以看到每次遍歷segments時都會執行int c = segments[i].count;，但是接下來的語句中又不用此變量c，盡管如此JVM仍然不能將此語句優化掉，因為這是一個volatile字段的讀取操作，它保證了一些列操作的happens-before順序，所以是至關重要的。在這里可以看到：

ConcurrentHashMap將volatile發揮到了極致！

另外isEmpty操作于size操作類似，不再累述。

清單5 ConcurrentHashMap的size操作

public int size() {
    final Segment<K,V>[] segments = this.segments;
    long sum = 0;
    long check = 0;
    int[] mc = new int[segments.length];
    // Try a few times to get accurate count. On failure due to
    // continuous async changes in table, resort to locking.
    for (int k = 0; k < RETRIES_BEFORE_LOCK; ++k) {
        check = 0;
        sum = 0;
        int mcsum = 0;
        for (int i = 0; i < segments.length; ++i) {
            sum += segments[i].count;
            mcsum += mc[i] = segments[i].modCount;
        }
        if (mcsum != 0) {
            for (int i = 0; i < segments.length; ++i) {
                check += segments[i].count;
                if (mc[i] != segments[i].modCount) {
                    check = -1; // force retry
                    break;
                }
            }
        }
        if (check == sum)
            break;
    }
    if (check != sum) { // Resort to locking all segments
        sum = 0;
        for (int i = 0; i < segments.length; ++i)
            segments[i].lock();
        for (int i = 0; i < segments.length; ++i)
            sum += segments[i].count;
        for (int i = 0; i < segments.length; ++i)
            segments[i].unlock();
    }
    if (sum > Integer.MAX_VALUE)
        return Integer.MAX_VALUE;
    else
        return (int)sum;
}

ConcurrentSkipListMap/Set

本來打算介紹下ConcurrentSkipListMap的，結果打開源碼一看，徹底放棄了。那里面的數據結構和算法我估計研究一周也未必能夠完全弄懂。很久以前我看TreeMap的時候就頭大，想想那些復雜的“紅黑二叉樹”我頭都大了。這些都歸咎于從前沒有好好學習《數據結構和算法》，現在再回頭看這些復雜的算法感覺非常頭疼，為了減少腦細胞的死亡，暫且還是不要惹這些“玩意兒”。有興趣的可以看看參考資料4 中對TreeMap的介紹。

參考資料：

1. Hash this
2. Single-word Wang/Jenkins Hash in ConcurrentHashMap
3. 指令重排序與happens-before法則
4. 通過分析 JDK 源代碼研究 TreeMap 紅黑樹算法實現