當(dāng)前位置：首頁(yè) > 编程资源 > 编程问答 >内容正文

编程问答

史上最详细的ConcurrentHashMap详解--源码分析

發(fā)布時(shí)間：2024/3/12 编程问答 33 豆豆

生活随笔收集整理的這篇文章主要介紹了史上最详细的ConcurrentHashMap详解--源码分析小編覺(jué)得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.

ps.本文所有源碼都是基于jdk1.6
首先說(shuō)明一點(diǎn)，ConcurrentHashMap并不是可以完全替換Hashtable的，因?yàn)镃oncurrentHashMap的get、clear函數(shù)是弱一致的（后面會(huì)說(shuō)到），而Hashtable是強(qiáng)一致的。有作者是這么解釋的：我們將“一致性強(qiáng)度”和“擴(kuò)展性”之間的對(duì)比交給用戶來(lái)權(quán)衡，所以大多數(shù)集合都提供了synchronized和concurrent兩個(gè)版本。不過(guò)真正需要“強(qiáng)一致性”的場(chǎng)景可能非常少，我們大多應(yīng)用中ConcurrentHashMap是滿足的。

ConcurrentHashMap的數(shù)據(jù)結(jié)構(gòu)

不得不說(shuō)，ConcurrentHashMap設(shè)計(jì)的相當(dāng)巧妙，它有多個(gè)段，每個(gè)段下面都是一個(gè)Hashtable（相似），所以每個(gè)段上都有一把鎖（分布式鎖），各個(gè)段之間的鎖互不影響，可以實(shí)現(xiàn)并發(fā)操作。話不多說(shuō)，上代碼。

final Segment<K,V>[] segments;

可以看到ConcurrentHashMap實(shí)際上就是一個(gè)Segment數(shù)組，那么Segment是什么呢？

static final class Segment<K,V> extends ReentrantLock implements Serializable {...transient volatile HashEntry<K,V>[] table; //發(fā)現(xiàn)Segment實(shí)際上是HashEntry數(shù)組... }

那HashEntry又是什么呢？

static final class HashEntry<K,V> {final K key;final int hash;volatile V value;final HashEntry<K,V> next;... }

HashEntry是一個(gè)單鏈表
所以ConcurrentHashMap的數(shù)據(jù)結(jié)構(gòu)如下圖：

這里每一個(gè)segment所指向的數(shù)據(jù)結(jié)構(gòu)，其實(shí)就是一個(gè)Hashtable，所以說(shuō)每一個(gè)segment都有一把獨(dú)立的鎖，來(lái)保證當(dāng)訪問(wèn)不同的segment時(shí)互不影響。

ConcurrentHashMap的基礎(chǔ)方法

基礎(chǔ)方法分為這么幾種：
1、段內(nèi)加鎖的：put，putIfAbsent，remove，replace等
2、不加鎖的：get，containsKey等
3、整個(gè)數(shù)據(jù)結(jié)構(gòu)加鎖的：size，containsValue等

構(gòu)造函數(shù)

public ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel) {if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)throw new IllegalArgumentException(); //參數(shù)合法性校驗(yàn)if (concurrencyLevel > MAX_SEGMENTS)concurrencyLevel = MAX_SEGMENTS; //比如我輸入的concurrencyLevel=12，那么sshift = 4，ssize =16，所以sshift是意思就是1左移了幾次比concurrencyLevel大，ssize就是那個(gè)大于等于concurrencyLevel的最小2的冪次方的數(shù)int sshift = 0;int ssize = 1;while (ssize < concurrencyLevel) {++sshift;ssize <<= 1; //ssize = ssize << 1 , ssize = ssize * 2}segmentShift = 32 - sshift;segmentMask = ssize - 1; //segmentMask的二進(jìn)制是一個(gè)全是1的數(shù) this.segments = Segment.newArray(ssize); //segment個(gè)數(shù)是ssize，也就是上圖黃色方塊數(shù)，默認(rèn)16個(gè)if (initialCapacity > MAXIMUM_CAPACITY)initialCapacity = MAXIMUM_CAPACITY;int c = initialCapacity / ssize;if (c * ssize < initialCapacity)++c;int cap = 1;while (cap < c)cap <<= 1; for (int i = 0; i < this.segments.length; ++i)this.segments[i] = new Segment<K,V>(cap, loadFactor);//上圖淺綠色塊的個(gè)數(shù)是cap，是一個(gè)2的冪次方的數(shù)，默認(rèn)是1，也就是每個(gè)segment下都構(gòu)造了cap大小的table數(shù)組 } Segment(int initialCapacity, float lf) {loadFactor = lf;setTable(HashEntry.<K,V>newArray(initialCapacity));//構(gòu)造了一個(gè)initialCapacity大小的table }

put函數(shù)

public V put(K key, V value) {if (value == null)throw new NullPointerException(); //明確指定value不能為nullint hash = hash(key.hashCode());return segmentFor(hash).put(key, hash, value, false); //segmentFor如下，定位segment下標(biāo)，然后再put }final Segment<K,V> segmentFor(int hash) { //尋找segment的下標(biāo)return segments[(hash >>> segmentShift) & segmentMask]; //前面說(shuō)了segmentMask是一個(gè)2進(jìn)制全是1的數(shù)，那么&segmentMask就保證了下標(biāo)小于等于segmentMask，與HashMap的尋下標(biāo)相似。 }//真正的put操作 V put(K key, int hash, V value, boolean onlyIfAbsent) {lock(); //先加鎖，可以看到，put操作是在segment里面加鎖的，也就是每個(gè)segment都可以加一把鎖try {int c = count;if (c++ > threshold) // ensure capacityrehash(); //判斷容量，如果不夠了就擴(kuò)容HashEntry<K,V>[] tab = table; //將table賦給一個(gè)局部變量tab，這是因?yàn)閠able是volatile變量，讀寫volatile變量的開(kāi)銷很大，編譯器也不能對(duì)volatile變量的讀寫做任何優(yōu)化，直接多次訪問(wèn)非volatile實(shí)例變量沒(méi)有多大影響，編譯器會(huì)做相應(yīng)優(yōu)化。int index = hash & (tab.length - 1); //尋找table的下標(biāo)HashEntry<K,V> first = tab[index];HashEntry<K,V> e = first;while (e != null && (e.hash != hash || !key.equals(e.key)))e = e.next; //遍歷單鏈表，找到key相同的為止，如果沒(méi)找到，e指向鏈表尾V oldValue;if (e != null) { //如果有相同的key，那么直接替換oldValue = e.value;if (!onlyIfAbsent)e.value = value;}else { //否則在鏈表表頭插入新的結(jié)點(diǎn)oldValue = null;++modCount;tab[index] = new HashEntry<K,V>(key, hash, first, value);count = c; // write-volatile}return oldValue;} finally {unlock();} }

該方法也是在持有段鎖的情況下執(zhí)行的，首先判斷是否需要rehash，需要就先rehash，擴(kuò)容都是針對(duì)單個(gè)段的，也就是單個(gè)段的數(shù)據(jù)數(shù)量大于設(shè)定的量的時(shí)候會(huì)觸發(fā)擴(kuò)容。接著是找是否存在同樣一個(gè)key的結(jié)點(diǎn)，如果存在就直接替換這個(gè)結(jié)點(diǎn)的值。否則創(chuàng)建一個(gè)新的結(jié)點(diǎn)并添加到hash鏈的頭部，這時(shí)一定要修改modCount和count的值，同樣修改count的值一定要放在最后一步。put方法調(diào)用了rehash方法，reash方法實(shí)現(xiàn)得也很精巧，主要利用了table的大小為2^n，和HashMap的擴(kuò)容基本一樣，這里就不介紹了。

還有一個(gè)叫putIfAbsent(K key, V value)的函數(shù)，這個(gè)函數(shù)的實(shí)現(xiàn)和put幾乎一模一樣，作用是，如果map中不存在這個(gè)key，那么插入這個(gè)數(shù)據(jù)，如果存在這個(gè)key，那么不覆蓋原來(lái)的value，也就是不插入了。

remove函數(shù)

>remove操作也交給了段內(nèi)的remove，如下： V remove(Object key, int hash, Object value) {lock(); //段內(nèi)先獲得一把鎖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) { //如果找到該keyV 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; //newFirst此時(shí)為要?jiǎng)h除結(jié)點(diǎn)的nextfor (HashEntry<K,V> p = first; p != e; p = p.next)newFirst = new HashEntry<K,V>(p.key,p.hash,newFirst, p.value);//從頭遍歷鏈表將要?jiǎng)h除結(jié)點(diǎn)的前面所有結(jié)點(diǎn)復(fù)制一份插入到newFirst之前，如下圖tab[index] = newFirst;count = c; // write-volatile}}return oldValue;} finally {unlock();} }

為什么用這么方式刪除呢，細(xì)心的同學(xué)會(huì)發(fā)現(xiàn)上面定義的HashEntry的key和next都是final類型的，所以不能改變next的指向，所以又復(fù)制了一份指向刪除的結(jié)點(diǎn)的next。

get函數(shù)

public V get(Object key) {int hash = hash(key.hashCode()); //雙hash，和HashMap一樣，也是為了更好的離散化return segmentFor(hash).get(key, hash); //先尋找segment的下標(biāo)，然后再get }final Segment<K,V> segmentFor(int hash) { //尋找segment的下標(biāo)return segments[(hash >>> segmentShift) & segmentMask]; //前面說(shuō)了segmentMask是一個(gè)2進(jìn)制全是1的數(shù)，那么&segmentMask就保證了下標(biāo)小于等于segmentMask，與HashMap的尋下標(biāo)相似。 }V get(Object key, int hash) {// count是一個(gè)volatile變量，count非常巧妙，每次put和remove之后的最后一步都要更新count，就是為了get的時(shí)候不讓編譯器對(duì)代碼進(jìn)行重排序，來(lái)保證if (count != 0) { HashEntry<K,V> e = getFirst(hash); //尋找table的下標(biāo)，也就是鏈表的表頭while (e != null) {if (e.hash == hash && key.equals(e.key)) {V v = e.value;if (v != null)return v;return readValueUnderLock(e); // recheck 加鎖讀，這個(gè)加鎖讀不用重新計(jì)算位置，而是直接拿e的值}e = e.next;}}return null; }HashEntry<K,V> getFirst(int hash) {HashEntry<K,V>[] tab = table;return tab[hash & (tab.length - 1)]; }

讀完上面代碼我有一個(gè)疑問(wèn)，就是如果找到的key對(duì)應(yīng)的value是null的話，加鎖再讀一次，既然上面put操作不允許value是null，那讀到的value為什么會(huì)有null的情況呢？我們分析一下這種情況，就是put操作和get操作同時(shí)進(jìn)行的時(shí)候，可以分為兩種情況：
1、put的key已經(jīng)存在，由于value不是final的，可以直接更新，且value是volatile的，所以修改會(huì)立馬對(duì)get線程可見(jiàn)，而不用等到put方法結(jié)束。
2、put的key不存在，那么將在鏈表表頭插入一個(gè)數(shù)據(jù)，那么將new HashEntry賦值給tab[index]是否能立刻對(duì)執(zhí)行g(shù)et的線程可見(jiàn)呢，我們知道每次put完之后都要更新一個(gè)count變量（寫），而每次get數(shù)據(jù)的時(shí)候，再最一開(kāi)始都要讀一個(gè)count變量（讀），而且發(fā)現(xiàn)這個(gè)count是volatile的，而對(duì)同一個(gè)volatile變量，有volatile寫 happens-before volatile讀，所以如果寫發(fā)生在讀之前，那么new HashEntry賦值給tab[index]是對(duì)get線程可見(jiàn)的，但是如果寫沒(méi)有發(fā)生在讀之前，就無(wú)法保證new HashEntry賦值給tab[index]要先于get函數(shù)的getFirst(hash)，也就是說(shuō)，如果某個(gè)Segment實(shí)例中的put將一個(gè)Entry加入到了table中，在未執(zhí)行count賦值操作之前有另一個(gè)線程執(zhí)行了同一個(gè)Segment實(shí)例中的get，來(lái)獲取這個(gè)剛加入的Entry中的value，那么是有可能取不到的，這也就是get的弱一致性。但是什么時(shí)候會(huì)找到key但是讀到的value是null呢，仔細(xì)看下put操作的語(yǔ)句：tab[index] = new HashEntry(key, hash, first, value)，在這條語(yǔ)句中，HashEntry構(gòu)造函數(shù)中對(duì)value的賦值以及對(duì)tab[index]的賦值可能被重新排序，舉個(gè)例子就是這條語(yǔ)句有可能先執(zhí)行對(duì)key賦值，再執(zhí)行對(duì)tab[index]的賦值，最后對(duì)value賦值，如果在對(duì)tab和key都賦值但是對(duì)value還沒(méi)賦值的情況下的get就是一個(gè)空值。
詳細(xì)可以看看這篇文章：http://ifeve.com/concurrenthashmap-weakly-consistent/
這也就是說(shuō)無(wú)鎖的get操作是一個(gè)弱一致性的操作。

clear函數(shù)

public void clear() {for (int i = 0; i < segments.length; ++i) //循環(huán)clear掉每個(gè)段中的內(nèi)容segments[i].clear(); } void clear() {if (count != 0) {lock(); //段內(nèi)加鎖try {HashEntry<K,V>[] tab = table;for (int i = 0; i < tab.length ; i++)tab[i] = null;++modCount;count = 0; // write-volatile} finally {unlock();}} }

因?yàn)闆](méi)有全局的鎖，在清除完一個(gè)segments之后，正在清理下一個(gè)segments的時(shí)候，已經(jīng)清理segments可能又被加入了數(shù)據(jù)，因此clear返回的時(shí)候，ConcurrentHashMap中是可能存在數(shù)據(jù)的。因此，clear方法是弱一致的。

size函數(shù)

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; //循環(huán)相加每個(gè)段內(nèi)數(shù)據(jù)的個(gè)數(shù)mcsum += mc[i] = segments[i].modCount; //循環(huán)相加每個(gè)段內(nèi)的modCount}if (mcsum != 0) { //如果是0，代表根本沒(méi)有過(guò)數(shù)據(jù)更改，也就是size是0for (int i = 0; i < segments.length; ++i) {check += segments[i].count; //再次循環(huán)相加每個(gè)段內(nèi)數(shù)據(jù)的個(gè)數(shù)，這里為什么會(huì)再算一次呢，后面會(huì)說(shuō)if (mc[i] != segments[i].modCount) { check = -1; // force retry //如果modCount和之前統(tǒng)計(jì)的不一致了，check直接賦值-1，重新再來(lái)break;}}}if (check == sum)break;}if (check != sum) { // Resort to locking all segmentssum = 0;for (int i = 0; i < segments.length; ++i) //循環(huán)獲取所有segment的鎖segments[i].lock();for (int i = 0; i < segments.length; ++i) //在持有所有段的鎖的時(shí)候進(jìn)行count的相加sum += segments[i].count;for (int i = 0; i < segments.length; ++i) //循環(huán)釋放所有段的鎖segments[i].unlock();}if (sum > Integer.MAX_VALUE) //returnreturn Integer.MAX_VALUE;elsereturn (int)sum; }

如果我們要統(tǒng)計(jì)整個(gè)ConcurrentHashMap里元素的大小，就必須統(tǒng)計(jì)所有Segment里元素的大小后求和。Segment里的全局變量count是一個(gè)volatile變量，那么在多線程場(chǎng)景下，我們是不是直接把所有Segment的count相加就可以得到整個(gè)ConcurrentHashMap大小了呢？不是的，雖然相加時(shí)可以獲取每個(gè)Segment的count的最新值，但是拿到之后可能累加前使用的count發(fā)生了變化，那么統(tǒng)計(jì)結(jié)果就不準(zhǔn)了。所以最安全的做法，是在統(tǒng)計(jì)size的時(shí)候把所有Segment的put，remove和clean方法全部鎖住，但是這種做法顯然非常低效，因?yàn)樵诶奂觕ount操作過(guò)程中，之前累加過(guò)的count發(fā)生變化的幾率非常小，所以ConcurrentHashMap的做法是先嘗試2次通過(guò)不鎖住Segment的方式來(lái)統(tǒng)計(jì)各個(gè)Segment大小，如果統(tǒng)計(jì)的過(guò)程中，容器的count發(fā)生了變化，則再采用加鎖的方式來(lái)統(tǒng)計(jì)所有Segment的大小。那么ConcurrentHashMap是如何判斷在統(tǒng)計(jì)的時(shí)候容器是否發(fā)生了變化呢？使用modCount變量，在put , remove和clear方法里操作元素前都會(huì)將變量modCount進(jìn)行加1，那么在統(tǒng)計(jì)size前后比較modCount是否發(fā)生變化，從而得知容器的大小是否發(fā)生變化。size()的實(shí)現(xiàn)還有一點(diǎn)需要注意，必須要先segments[i].count，才能segments[i].modCount，這是因?yàn)閟egment[i].count是對(duì)volatile變量的訪問(wèn)，接下來(lái)segments[i].modCount才能得到幾乎最新的值，這里和get方法的方式是一樣的，也是一個(gè)volatile寫 happens-before volatile讀的問(wèn)題。

上面18行代碼，拋出了一個(gè)問(wèn)題，就是為什么會(huì)再算一遍，上面說(shuō)只需要比較modCount不變不就可以了么？但是仔細(xì)分析，就會(huì)發(fā)現(xiàn)，在13行14行代碼那里，計(jì)算完count之后，計(jì)算modCount之前有可能count的值又變了，那么18行的代碼主要是解決這個(gè)問(wèn)題。

containsValue函數(shù)

containsKey函數(shù)比較簡(jiǎn)單，也是不加鎖的簡(jiǎn)易get，下面說(shuō)一下containsValue有一個(gè)有意思的地方

public boolean containsValue(Object value) {if (value == null)throw new NullPointerException();// See explanation of modCount use abovefinal Segment<K,V>[] segments = this.segments;int[] mc = new int[segments.length];// Try a few times without lockingfor (int k = 0; k < RETRIES_BEFORE_LOCK; ++k) {int sum = 0;int mcsum = 0;for (int i = 0; i < segments.length; ++i) {int c = segments[i].count; //注意這行，聲明了一個(gè)c變量，并且賦值了，但是下面卻完全沒(méi)有用到。mcsum += mc[i] = segments[i].modCount;if (segments[i].containsValue(value))return true;}boolean cleanSweep = true; //默認(rèn)結(jié)構(gòu)沒(méi)變if (mcsum != 0) {for (int i = 0; i < segments.length; ++i) {int c = segments[i].count;if (mc[i] != segments[i].modCount) { //如果結(jié)構(gòu)變了，cleanSweep = falsecleanSweep = false;break;}}}if (cleanSweep) //如果沒(méi)變，直接返回falsereturn false;}// Resort to locking all segmentsfor (int i = 0; i < segments.length; ++i)segments[i].lock();boolean found = false;try {for (int i = 0; i < segments.length; ++i) {if (segments[i].containsValue(value)) {found = true;break;}}} finally {for (int i = 0; i < segments.length; ++i)segments[i].unlock();}return found; }

注意上面代碼15行處，里面有語(yǔ)句int c = segments[i].count; 但是c卻從來(lái)沒(méi)有被使用過(guò)，即使如此，編譯器也不能做優(yōu)化將這條語(yǔ)句去掉，因?yàn)榇嬖趯?duì)volatile變量count的讀取，這條語(yǔ)句存在的唯一目的就是保證segments[i].modCount讀取到幾乎最新的值，上面有說(shuō)道這個(gè)問(wèn)題。

本文章參考文章：
1、http://ifeve.com/concurrenthashmap-weakly-consistent/
2、http://www.iteye.com/topic/344876

總結(jié)

以上是生活随笔為你收集整理的史上最详细的ConcurrentHashMap详解--源码分析的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。

上一篇： Three.js中添加指南针
下一篇： cad指北针lisp_用CAD里的LIS