ConcurrentHashMap

ConcurrentHashMap是一個(gè)線程安全的Hash Table，它的主要功能是提供了一組和HashTable功能相同但是線程安全的方法,。ConcurrentHashMap可以做到讀取數(shù)據(jù)不加鎖，并且其內(nèi)部的結(jié)構(gòu)可以讓其在進(jìn)行寫操作的時(shí)候能夠?qū)㈡i的粒度保持地盡量地小，不用對整個(gè)ConcurrentHashMap加鎖,。

ConcurrentHashMap的內(nèi)部結(jié)構(gòu)

ConcurrentHashMap為了提高本身的并發(fā)能力，在內(nèi)部采用了一個(gè)叫做Segment的結(jié)構(gòu),，一個(gè)Segment其實(shí)就是一個(gè)類Hash Table的結(jié)構(gòu),，Segment內(nèi)部維護(hù)了一個(gè)鏈表數(shù)組，我們用下面這一幅圖來看下ConcurrentHashMap的內(nèi)部結(jié)構(gòu)：

從上面的結(jié)構(gòu)我們可以了解到,，ConcurrentHashMap定位一個(gè)元素的過程需要進(jìn)行兩次Hash操作,，第一次Hash定位到Segment，第二次Hash定位到元素所在的鏈表的頭部,，因此,，這一種結(jié)構(gòu)的帶來的副作用是Hash的過程要比普通的HashMap要長，但是帶來的好處是寫操作的時(shí)候可以只對元素所在的Segment進(jìn)行加鎖即可,，不會(huì)影響到其他的Segment,，這樣,，在最理想的情況下，ConcurrentHashMap可以最高同時(shí)支持Segment數(shù)量大小的寫操作（剛好這些寫操作都非常平均地分布在所有的Segment上）,，所以,，通過這一種結(jié)構(gòu)，ConcurrentHashMap的并發(fā)能力可以大大的提高,。

Segment

我們再來具體了解一下Segment的數(shù)據(jù)結(jié)構(gòu)：

Java代碼  

1. static final class Segment<K,V> extends ReentrantLock implements Serializable {  
2.     transient volatile int count;  
3.     transient int modCount;  
4.     transient int threshold;  
5.     transient volatile HashEntry<K,V>[] table;  
6.     final float loadFactor;  
7. }  

詳細(xì)解釋一下Segment里面的成員變量的意義：

• count：Segment中元素的數(shù)量
• modCount：對table的大小造成影響的操作的數(shù)量（比如put或者remove操作）
• threshold：閾值,，Segment里面元素的數(shù)量超過這個(gè)值依舊就會(huì)對Segment進(jìn)行擴(kuò)容
• table：鏈表數(shù)組，數(shù)組中的每一個(gè)元素代表了一個(gè)鏈表的頭部
• loadFactor：負(fù)載因子,，用于確定threshold

HashEntry

Segment中的元素是以HashEntry的形式存放在鏈表數(shù)組中的,，看一下HashEntry的結(jié)構(gòu)：

Java代碼  

1. static final class HashEntry<K,V> {  
2.     final K key;  
3.     final int hash;  
4.     volatile V value;  
5.     final HashEntry<K,V> next;  
6. }  

可以看到HashEntry的一個(gè)特點(diǎn)，除了value以外,，其他的幾個(gè)變量都是final的,，這樣做是為了防止鏈表結(jié)構(gòu)被破壞，出現(xiàn)ConcurrentModification的情況,。

ConcurrentHashMap的初始化

下面我們來結(jié)合源代碼來具體分析一下ConcurrentHashMap的實(shí)現(xiàn),，先看下初始化方法：

Java代碼  

1. public ConcurrentHashMap(int initialCapacity,  
2.                          float loadFactor, int concurrencyLevel) {  
3.     if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)  
4.         throw new IllegalArgumentException();  
5.    
6.     if (concurrencyLevel > MAX_SEGMENTS)  
7.         concurrencyLevel = MAX_SEGMENTS;  
8.    
9.     // Find power-of-two sizes best matching arguments  
10.     int sshift = 0;  
11.     int ssize = 1;  
12.     while (ssize < concurrencyLevel) {  
13.         ++sshift;  
14.         ssize <<= 1;  
15.     }  
16.     segmentShift = 32 - sshift;  
17.     segmentMask = ssize - 1;  
18.     this.segments = Segment.newArray(ssize);  
19.    
20.     if (initialCapacity > MAXIMUM_CAPACITY)  
21.         initialCapacity = MAXIMUM_CAPACITY;  
22.     int c = initialCapacity / ssize;  
23.     if (c * ssize < initialCapacity)  
24.         ++c;  
25.     int cap = 1;  
26.     while (cap < c)  
27.         cap <<= 1;  
28.    
29.     for (int i = 0; i < this.segments.length; ++i)  
30.         this.segments[i] = new Segment<K,V>(cap, loadFactor);  
31. }  

CurrentHashMap的初始化一共有三個(gè)參數(shù)，一個(gè)initialCapacity,，表示初始的容量,，一個(gè)loadFactor，表示負(fù)載參數(shù),，最后一個(gè)是concurrentLevel,，代表ConcurrentHashMap內(nèi)部的Segment的數(shù)量，ConcurrentLevel一經(jīng)指定,，不可改變,，后續(xù)如果ConcurrentHashMap的元素?cái)?shù)量增加導(dǎo)致ConrruentHashMap需要擴(kuò)容，ConcurrentHashMap不會(huì)增加Segment的數(shù)量,，而只會(huì)增加Segment中鏈表數(shù)組的容量大小,，這樣的好處是擴(kuò)容過程不需要對整個(gè)ConcurrentHashMap做rehash，而只需要對Segment里面的元素做一次rehash就可以了,。

整個(gè)ConcurrentHashMap的初始化方法還是非常簡單的,，先是根據(jù)concurrentLevel來new出Segment，這里Segment的數(shù)量是不大于concurrentLevel的最大的2的指數(shù),，就是說Segment的數(shù)量永遠(yuǎn)是2的指數(shù)個(gè),，這樣的好處是方便采用移位操作來進(jìn)行hash，加快hash的過程,。接下來就是根據(jù)intialCapacity確定Segment的容量的大小,，每一個(gè)Segment的容量大小也是2的指數(shù)，同樣使為了加快hash的過程。

這邊需要特別注意一下兩個(gè)變量,，分別是segmentShift和segmentMask,，這兩個(gè)變量在后面將會(huì)起到很大的作用，假設(shè)構(gòu)造函數(shù)確定了Segment的數(shù)量是2的n次方,，那么segmentShift就等于32減去n,，而segmentMask就等于2的n次方減一。

ConcurrentHashMap的get操作

前面提到過ConcurrentHashMap的get操作是不用加鎖的,，我們這里看一下其實(shí)現(xiàn)：

Java代碼  

1. public V get(Object key) {  
2.     int hash = hash(key.hashCode());  
3.     return segmentFor(hash).get(key, hash);  
4. }  

看第三行,，segmentFor這個(gè)函數(shù)用于確定操作應(yīng)該在哪一個(gè)segment中進(jìn)行，幾乎對ConcurrentHashMap的所有操作都需要用到這個(gè)函數(shù),，我們看下這個(gè)函數(shù)的實(shí)現(xiàn)：

Java代碼  

1. final Segment<K,V> segmentFor(int hash) {  
2.     return segments[(hash >>> segmentShift) & segmentMask];  
3. }  

這個(gè)函數(shù)用了位操作來確定Segment，根據(jù)傳入的hash值向右無符號(hào)右移segmentShift位,，然后和segmentMask進(jìn)行與操作,，結(jié)合我們之前說的segmentShift和segmentMask的值，就可以得出以下結(jié)論：假設(shè)Segment的數(shù)量是2的n次方,，根據(jù)元素的hash值的高n位就可以確定元素到底在哪一個(gè)Segment中,。

在確定了需要在哪一個(gè)segment中進(jìn)行操作以后，接下來的事情就是調(diào)用對應(yīng)的Segment的get方法：

Java代碼  

1. V get(Object key, int hash) {  
2.     if (count != 0) { // read-volatile  
3.         HashEntry<K,V> e = getFirst(hash);  
4.         while (e != null) {  
5.             if (e.hash == hash && key.equals(e.key)) {  
6.                 V v = e.value;  
7.                 if (v != null)  
8.                     return v;  
9.                 return readValueUnderLock(e); // recheck  
10.             }  
11.             e = e.next;  
12.         }  
13.     }  
14.     return null;  
15. }  

先看第二行代碼,，這里對count進(jìn)行了一次判斷,，其中count表示Segment中元素的數(shù)量，我們可以來看一下count的定義：

Java代碼  

1. transient volatile int count;  

可以看到count是volatile的,，實(shí)際上這里里面利用了volatile的語義：

因?yàn)閷?shí)際上put、remove等操作也會(huì)更新count的值,，所以當(dāng)競爭發(fā)生的時(shí)候,，volatile的語義可以保證寫操作在讀操作之前，也就保證了寫操作對后續(xù)的讀操作都是可見的,，這樣后面get的后續(xù)操作就可以拿到完整的元素內(nèi)容,。

然后，在第三行,，調(diào)用了getFirst()來取得鏈表的頭部：

Java代碼  

1. HashEntry<K,V> getFirst(int hash) {  
2.     HashEntry<K,V>[] tab = table;  
3.     return tab[hash & (tab.length - 1)];  
4. }  

同樣,，這里也是用位操作來確定鏈表的頭部，hash值和HashTable的長度減一做與操作,，最后的結(jié)果就是hash值的低n位,，其中n是HashTable的長度以2為底的結(jié)果。

在確定了鏈表的頭部以后,，就可以對整個(gè)鏈表進(jìn)行遍歷,，看第4行，取出key對應(yīng)的value的值,，如果拿出的value的值是null,，則可能這個(gè)key,，value對正在put的過程中，如果出現(xiàn)這種情況,，那么就加鎖來保證取出的value是完整的,，如果不是null，則直接返回value,。

ConcurrentHashMap的put操作

看完了get操作,，再看下put操作，put操作的前面也是確定Segment的過程,，這里不再贅述,，直接看關(guān)鍵的segment的put方法：

Java代碼  

1. V put(K key, int hash, V value, boolean onlyIfAbsent) {  
2.     lock();  
3.     try {  
4.         int c = count;  
5.         if (c++ > threshold) // ensure capacity  
6.             rehash();  
7.         HashEntry<K,V>[] tab = table;  
8.         int index = hash & (tab.length - 1);  
9.         HashEntry<K,V> first = tab[index];  
10.         HashEntry<K,V> e = first;  
11.         while (e != null && (e.hash != hash || !key.equals(e.key)))  
12.             e = e.next;  
13.    
14.         V oldValue;  
15.         if (e != null) {  
16.             oldValue = e.value;  
17.             if (!onlyIfAbsent)  
18.                 e.value = value;  
19.         }  
20.         else {  
21.             oldValue = null;  
22.             ++modCount;  
23.             tab[index] = new HashEntry<K,V>(key, hash, first, value);  
24.             count = c; // write-volatile  
25.         }  
26.         return oldValue;  
27.     } finally {  
28.         unlock();  
29.     }  
30. }  

首先對Segment的put操作是加鎖完成的，然后在第五行,，如果Segment中元素的數(shù)量超過了閾值（由構(gòu)造函數(shù)中的loadFactor算出）這需要進(jìn)行對Segment擴(kuò)容,，并且要進(jìn)行rehash，關(guān)于rehash的過程大家可以自己去了解,，這里不詳細(xì)講了,。

第8和第9行的操作就是getFirst的過程，確定鏈表頭部的位置,。

第11行這里的這個(gè)while循環(huán)是在鏈表中尋找和要put的元素相同key的元素,，如果找到，就直接更新更新key的value,，如果沒有找到,，則進(jìn)入21行這里，生成一個(gè)新的HashEntry并且把它加到整個(gè)Segment的頭部,，然后再更新count的值,。

ConcurrentHashMap的remove操作

Remove操作的前面一部分和前面的get和put操作一樣，都是定位Segment的過程,，然后再調(diào)用Segment的remove方法：

Java代碼  

1. V remove(Object key, int hash, Object value) {  
2.     lock();  
3.     try {  
4.         int c = count - 1;  
5.         HashEntry<K,V>[] tab = table;  
6.         int index = hash & (tab.length - 1);  
7.         HashEntry<K,V> first = tab[index];  
8.         HashEntry<K,V> e = first;  
9.         while (e != null && (e.hash != hash || !key.equals(e.key)))  
10.             e = e.next;  
11.    
12.         V oldValue = null;  
13.         if (e != null) {  
14.             V v = e.value;  
15.             if (value == null || value.equals(v)) {  
16.                 oldValue = v;  
17.                 // All entries following removed node can stay  
18.                 // in list, but all preceding ones need to be  
19.                 // cloned.  
20.                 ++modCount;  
21.                 HashEntry<K,V> newFirst = e.next;  
22.                 for (HashEntry<K,V> p = first; p != e; p = p.next)  
23.                     newFirst = new HashEntry<K,V>(p.key, p.hash,  
24.                                                   newFirst, p.value);  
25.                 tab[index] = newFirst;  
26.                 count = c; // write-volatile  
27.             }  
28.         }  
29.         return oldValue;  
30.     } finally {  
31.         unlock();  
32.     }  
33. }  

首先remove操作也是確定需要?jiǎng)h除的元素的位置,，不過這里刪除元素的方法不是簡單地把待刪除元素的前面的一個(gè)元素的next指向后面一個(gè)就完事了，我們之前已經(jīng)說過HashEntry中的next是final的,，一經(jīng)賦值以后就不可修改,，在定位到待刪除元素的位置以后，程序就將待刪除元素前面的那一些元素全部復(fù)制一遍,，然后再一個(gè)一個(gè)重新接到鏈表上去,，看一下下面這一幅圖來了解這個(gè)過程：

假設(shè)鏈表中原來的元素如上圖所示，現(xiàn)在要?jiǎng)h除元素3,，那么刪除元素3以后的鏈表就如下圖所示：

ConcurrentHashMap的size操作

在前面的章節(jié)中,，我們涉及到的操作都是在單個(gè)Segment中進(jìn)行的，但是ConcurrentHashMap有一些操作是在多個(gè)Segment中進(jìn)行，比如size操作,，ConcurrentHashMap的size操作也采用了一種比較巧的方式,，來盡量避免對所有的Segment都加鎖。

前面我們提到了一個(gè)Segment中的有一個(gè)modCount變量,，代表的是對Segment中元素的數(shù)量造成影響的操作的次數(shù),，這個(gè)值只增不減，size操作就是遍歷了兩次Segment,，每次記錄Segment的modCount值,，然后將兩次的modCount進(jìn)行比較，如果相同,，則表示期間沒有發(fā)生過寫入操作,，就將原先遍歷的結(jié)果返回，如果不相同,，則把這個(gè)過程再重復(fù)做一次,，如果再不相同，則就需要將所有的Segment都鎖住,，然后一個(gè)一個(gè)遍歷了，具體的實(shí)現(xiàn)大家可以看ConcurrentHashMap的源碼,，這里就不貼了,。

PS. 原文在：Java并發(fā)編程之ConcurrentHashMap  請大家關(guān)注：黃金檔

PS：本篇文章的也提供了PDF下載：concurrentHashMap.pdf