Java集合框架之ConcurrentHashMap
參考此文件瞭解造成死迴圈的原因http://coolshell.cn/articles/9606.html。
所以，在多執行緒使用場景中，應該儘量避免使用執行緒不安全的HashMap，而使用執行緒安全的ConcurrentHashMap。
1.段分鎖技術
HashTable容器在競爭激烈的併發環境下表現出效率低下的原因是所有訪問HashTable的執行緒都必須競爭同一把鎖，那假如容器裡有多把鎖，每一把鎖用於鎖容器其中一部分資料，那麼當多執行緒訪問容器裡不同資料段的資料時，執行緒間就不會存在鎖競爭，從而可以有效的提高併發訪問效率，這就是ConcurrentHashMap所使用的鎖分段技術，首先將資料分成一段一段的儲存，然後給每一段資料配一把鎖，當一個執行緒佔用鎖訪問其中一個段資料的時候，其他段的資料也能被其他執行緒訪問
2.ConcurrentHashMap的結構
ConcurrentHashMap的類圖如下所示：
ConcurrentHashMap是由Segment陣列結構和HashEntry陣列結構組成。Segment是一種可重入鎖ReentrantLock，在ConcurrentHashMap裡扮演鎖的角色，HashEntry則用於儲存鍵值對資料。一個ConcurrentHashMap裡包含一個Segment陣列，Segment的結構和HashMap類似，是一種陣列和連結串列結構， 一個Segment裡包含一個HashEntry陣列，每個HashEntry是一個連結串列結構的元素， 每個Segment守護著一個HashEntry數組裡的元素,當對HashEntry陣列的資料進行修改時，必須首先獲得它對應的Segment鎖。
HsahEntry的結構如下：
HashEntry| 1 2 3 4 5 6 7 8 9 10 11 12 | staticfinalclassHashEntry<K,V> { finalinthash; finalK key; volatileV value; volatileHashEntry<K,V> next;//之前版本次變數是final型別的。 HashEntry(inthash, K key, V value, HashEntry<K,V> next) { this.hash = hash; this.key = key; this.value = value; this.next = next; } } |
3.ConcurrentHashMap的初始化
ConcurrentHashMap的初始化| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 | publicConcurrentHashMap(intinitialCapacity, floatloadFactor,intconcurrencyLevel) { if(!(loadFactor >0) || initialCapacity <0|| concurrencyLevel <=0) thrownewIllegalArgumentException(); if(concurrencyLevel > MAX_SEGMENTS) concurrencyLevel = MAX_SEGMENTS; // Find power-of-two sizes best matching arguments intsshift =0; intssize =1; // 必須保證segments陣列的長度是2的N次方 while(ssize < concurrencyLevel) { ++sshift; ssize <<=1; } this.segmentShift =32- sshift; this.segmentMask = ssize -1; if(initialCapacity > MAXIMUM_CAPACITY) initialCapacity = MAXIMUM_CAPACITY; intc = initialCapacity / ssize; if(c * ssize < initialCapacity) ++c; intcap = MIN_SEGMENT_TABLE_CAPACITY; //必須保證HashEntry陣列的長度是2的N次方 while(cap < c) cap <<=1; // create segments and segments[0] Segment<K,V> s0 = newSegment<K,V>(loadFactor, (int)(cap * loadFactor), (HashEntry<K,V>[])newHashEntry[cap]); Segment<K,V>[] ss = (Segment<K,V>[])newSegment[ssize]; UNSAFE.putOrderedObject(ss, SBASE, s0);// ordered write of segments[0] this.segments = ss; } |
ConcurrentHashMap初始化方法是通過initialCapacity，loadFactor, concurrencyLevel幾個引數來初始化segments陣列，它們的預設初始值依次為16，0.75，16.
由上面的程式碼可知segments陣列的長度ssize通過concurrencyLevel計算得出。為了能通過按位與的雜湊演算法來定位segments陣列的索引，必須保證segments陣列的長度是2的N次方（power-of-two size），所以必須計算出一個是大於或等於concurrencyLevel的最小的2的N次方值來作為segments陣列的長度。假如concurrencyLevel等於14，15或16，ssize都會等於16，即容器裡鎖的個數也是16。segmentShift和segmentMask兩個全域性變數在定位segment時的雜湊演算法裡需要使用。
初始化每個Segment。輸入引數initialCapacity是ConcurrentHashMap的初始化容量，loadfactor是每個segment的負載因子，在構造方法裡需要通過這兩個引數來初始化陣列中的每個segment。變數cap就是segment裡HashEntry陣列的長度，它等於initialCapacity除以ssize的倍數c，如果c大於1，就會取大於等於c的2的N次方值，所以cap不是1，就是2的N次方。segment的容量threshold＝(int)cap*loadFactor。
4.定位Segment
既然ConcurrentHashMap使用分段鎖Segment來保護不同段的資料，那麼在插入和獲取元素的時候，必須先通過雜湊演算法定位到Segment。可以看到ConcurrentHashMap會首先使用Wang/Jenkins hash的變種演算法對元素的hashCode進行一次再雜湊。
再hash演算法| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 | privateinthash(Object k) { inth = hashSeed; if((0!= h) && (kinstanceofString)) { returnsun.misc.Hashing.stringHash32((String) k); } h ^= k.hashCode(); // 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); returnh ^ (h >>>16); } //int index = (tab.length - 1) & hash; |
之所以進行再雜湊，其目的是為了減少雜湊衝突，使元素能夠均勻的分佈在不同的Segment上，從而提高容器的存取效率。假如雜湊的質量差到極點，那麼所有的元素都在一個Segment中，不僅存取元素緩慢，分段鎖也會失去意義。可以做了一個測試，不通過再雜湊而直接執行雜湊計算。
測試| 1 2 3 4 | System.out.println(Integer.parseInt("0001111",2) &15); System.out.println(Integer.parseInt("0011111",2) &15); System.out.println(Integer.parseInt("0111111",2) &15); System.out.println(Integer.parseInt("1111111",2) &15); |
計算後輸出的雜湊值全是15，通過這個例子可以發現如果不進行再雜湊，雜湊衝突會非常嚴重，因為只要低位一樣，無論高位是什麼數，其雜湊值總是一樣。我們再把上面的二進位制資料進行再雜湊後結果如下，為了方便閱讀，不足32位的高位補了0，每隔四位用豎線分割下。
測試| 1 2 3 4 | 0100｜0111｜0110｜0111｜1101｜1010｜0100｜1110 1111｜0111｜0100｜0011｜0000｜0001｜1011｜1000 0111｜0111｜0110｜1001｜0100｜0110｜0011｜1110 1000｜0011｜0000｜0000｜1100｜1000｜0001｜1010 |
可以發現每一位的資料都雜湊開了，通過這種再雜湊能讓數字的每一位都能參加到雜湊運算當中，從而減少雜湊衝突。ConcurrentHashMap通過以下雜湊演算法定位segment。
定位segment的hash演算法 Collapse source| 1 2 3 4 5 6 7 8 9 | //JDK1.7 privateSegment<K,V> segmentForHash(inth) { longu = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE; return(Segment<K,V>) UNSAFE.getObjectVolatile(segments, u); } //JDK 1.6 finalSegment<K,V> segmentFor(inthash) { returnsegments[(hash >>> segmentShift) & segmentMask]; } |
預設情況下segmentShift為28，segmentMask為15，再雜湊後的數最大是32位二進位制資料，向右無符號移動28位，意思是讓高4位參與到hash運算中， (hash >>> segmentShift) & segmentMask的運算結果分別是4，15，7和8，可以看到hash值沒有發生衝突。
5.ConcurrentHashMap的Put操作
由於put方法裡需要對共享變數進行寫入操作，所以為了執行緒安全，在操作共享變數時必須得加鎖。Put方法首先定位到Segment，然後在Segment裡進行插入操作。插入操作需要經歷兩個步驟，第一步判斷是否需要對Segment裡的HashEntry陣列進行擴容，第二步定位新增元素的位置然後放在HashEntry數組裡。擴容的時候首先會建立一個兩倍於原容量的陣列，然後將原數組裡的元素進行再hash後插入到新的數組裡。為了高效ConcurrentHashMap不會對整個容器進行擴容，而只對某個segment進行擴容。
ConcurrentHashMap的put方法| 1 2 3 4 5 6 7 8 9 10 11 | publicV put(K key, V value) { Segment<K,V> s; if(value ==null) thrownewNullPointerException(); inthash = hash(key); intj = (hash >>> segmentShift) & segmentMask; if((s = (Segment<K,V>)UNSAFE.getObject// nonvolatile; recheck (segments, (j << SSHIFT) + SBASE)) ==null)// in ensureSegment s = ensureSegment(j); returns.put(key, hash, value,false); } |
6.ConcurrentHashMap的Get操作
Segment的get操作實現非常簡單和高效。先經過一次再雜湊，然後使用這個雜湊值通過雜湊運算定位到segment，再通過雜湊演算法定位到元素，程式碼如下：
ConcurrentHashMap的get方法| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 | publicV get(Object key) { Segment<K,V> s;// manually integrate access methods to reduce overhead HashEntry<K,V>[] tab; inth = hash(key); longu = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE; if((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) !=null&& (tab = s.table) !=null) { for(HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile (tab, ((long)(((tab.length -1) & h)) << TSHIFT) + TBASE); e !=null; e = e.next) { K k; if((k = e.key) == key || (e.hash == h && key.equals(k))) returne.value; } } returnnull; } |
get操作的高效之處在於整個get過程不需要加鎖，除非讀到的值是空的才會加鎖重讀，我們知道HashTable容器的get方法是需要加鎖的，那麼ConcurrentHashMap的get操作是如何做到不加鎖的呢？原因是它的get方法裡將要使用的共享變數都定義成volatile，如用於統計當前Segement大小的count欄位和用於儲存值的HashEntry的value。定義成volatile的變數，能夠線上程之間保持可見性，能夠被多執行緒同時讀，並且保證不會讀到過期的值，但是隻能被單執行緒寫（有一種情況可以被多執行緒寫，就是寫入的值不依賴於原值），在get操作裡只需要讀不需要寫共享變數count和value，所以可以不用加鎖。之所以不會讀到過期的值，是根據java記憶體模型的happen before原則，對volatile欄位的寫入操作先於讀操作，即使兩個執行緒同時修改和獲取volatile變數，get操作也能拿到最新的值，這是用volatile替換鎖的經典應用場景。
在定位元素的程式碼裡我們可以發現定位HashEntry和定位Segment的雜湊演算法雖然一樣，都與陣列的長度減去一相與，但是相與的值不一樣，定位Segment使用的是元素的hashcode通過再雜湊後得到的值的高位，而定位HashEntry直接使用的是再雜湊後的值。其目的是避免兩次雜湊後的值一樣，導致元素雖然在Segment裡雜湊開了，但是卻沒有在HashEntry裡雜湊開。
7.ConcurrentHashMap的Remove操作
整個操作是先定位到段，然後委託給段的remove操作。當多個刪除操作併發進行時，只要它們所在的段不相同，它們就可以同時進行。
ConcurrentHashMap的remove方法| 1 2 3 4 5 | publicV remove(Object key) { inthash = hash(key); Segment<K,V> s = segmentForHash(hash); returns ==null?null: s.remove(key, hash,null); } |
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 | finalV remove(Object key,inthash, Object value) { if(!tryLock()) scanAndLock(key, hash); V oldValue =null; try{ HashEntry<K,V>[] tab = table; intindex = (tab.length -1) & hash; HashEntry<K,V> e = entryAt(tab, index); HashEntry<K,V> pred =null; while(e !=null) { K k; HashEntry<K,V> next = e.next; if((k = e.key) == key || (e.hash == hash && key.equals(k))) { V v = e.value; if(value ==null|| value == v || value.equals(v)) { if(pred ==null) setEntryAt(tab, index, next); else pred.setNext(next); ++modCount; --count; oldValue = v; } break; } pred = e; e = next; } }finally{ unlock(); } returnoldValue; } |
整個操作是在持有段鎖的情況下執行的，首先定位到要刪除的節點e。接下來，如果不存在這個節點就直接返回null，否則尾結點指向e的下一個結點。下面是個示意圖。
刪除元素之前：
刪除元素3之後：
整個remove實現並不複雜，但是需要注意如下幾點。第一，當要刪除的結點存在時，刪除的最後一步操作要將count的值減一。這必須是最後一步操作，否則讀取操作可能看不到之前對段所做的結構性修改。第二，remove執行的開始就將table賦給一個區域性變數tab，這是因為table是volatile變數，讀寫volatile變數的開銷很大。編譯器也不能對volatile變數的讀寫做任何優化，直接多次訪問非volatile例項變數沒有多大影響，編譯器會做相應優化。
8.ConcurrentHashMap的Size操作
ConcurrentHashMap的Size方法| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 | publicintsize() { finalSegment<K,V>[] segments =this.segments; intsize; booleanoverflow;// true if size overflows 32 bits longsum;// sum of modCounts longlast = 0L;// previous sum intretries = -1;// first iteration isn't retry try{ for(;;) { if(retries++ == RETRIES_BEFORE_LOCK) {//RETRIES_BEFORE_LOCK為2 for(intj =0; j < segments.length; ++j) ensureSegment(j).lock();// force creation } sum = 0L; size =0; overflow =false; for(intj =0; j < segments.length; ++j) { Segment<K,V> seg = segmentAt(segments, j); if(seg !=null) { sum += seg.modCount; intc = seg.count; if(c <0|| (size += c) <0) overflow =true; } } if(sum == last)// 判斷Segment結構是否發生變化 break; last = sum; } }finally{ if(retries > RETRIES_BEFORE_LOCK) { for(intj =0; j < segments.length; ++j) segmentAt(segments, j).unlock(); } } returnoverflow ? Integer.MAX_VALUE : size; } |
如果我們要統計整個ConcurrentHashMap裡元素的大小，就必須統計所有Segment裡元素的大小後求和。Segment裡的全域性變數count是一個volatile變數，那麼在多執行緒場景下，我們是不是直接把所有Segment的count相加就可以得到整個ConcurrentHashMap大小了呢？不是的，雖然相加時可以獲取每個Segment的count的最新值，但是拿到之後可能累加前使用的count發生了變化，那麼統計結果就不準了。所以最安全的做法，是在統計size的時候把所有Segment的put，remove和clean方法全部鎖住，但是這種做法顯然非常低效。 因為在累加count操作過程中，之前累加過的count發生變化的機率非常小，所以ConcurrentHashMap的做法是先嚐試3次通過不鎖住Segment的方式來統計各個Segment大小，如果統計的過程中，容器的count發生了變化，則再採用加鎖的方式來統計所有Segment的大小。
那麼ConcurrentHashMap是如何判斷在統計的時候容器是否發生了變化呢？使用modCount變數，在put , remove和clean方法裡操作元素前都會將變數modCount進行加1，那麼在統計size前後比較modCount是否發生變化，從而得知容器的大小是否發生變化。
9.ConcurrentHashMap的ContainsKey操作
ConcurrentHashMap的ContainsKe方法| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 | publicbooleancontainsKey(Object key) { Segment<K,V> s; HashEntry<K,V>[] tab; inth = hash(key); longu = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE; if((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) !=null&& (tab = s.table) !=null) { for(HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile (tab, ((long)(((tab.length -1) & h)) << TSHIFT) + TBASE); e !=null; e = e.next) { K k; if((k = e.key) == key || (e.hash == h && key.equals(k))) returntrue; } } returnfalse; } |
過程簡單，不需要跨段操作，不需要讀取值，不需要加鎖。
10.ConcurrentHashMap的ContainsValue操作
ConcurrentHashMap的ContainsValue方法| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 | publicbooleancontainsValue(Object value) { // Same idea as size() if(value ==null) thrownewNullPointerException(); finalSegment<K,V>[] segments =this.segments; booleanfound =false; longlast =0; intretries = -1; try{ outer:for(;;) { if(retries++ == RETRIES_BEFORE_LOCK) { for(intj =0; j < segments.length; ++j) ensureSegment(j).lock();// force creation } longhashSum = 0L; intsum =0; for(intj =0; j < segments.length; ++j) { HashEntry<K,V>[] tab; Segment<K,V> seg = segmentAt(segments, j); if(seg !=null&& (tab = seg.table) !=null) { for(inti =0; i < tab.length; i++) { HashEntry<K,V> e; for(e = entryAt(tab, i); e !=null; e = e.next) { V v = e.value; if(v !=null&& value.equals(v)) { found =true; breakouter; } } } sum += seg.modCount; } } if(retries >0&& sum == last) break; last = sum; } }finally{ if(retries > RETRIES_BEFORE_LOCK) { for(intj =0; j < segments.length; ++j) segmentAt(segments, j).unlock(); } } returnfound; } |
實現方法類似於size()。
11.ConcurrentHashMap的isEmpty操作
ConcurrentHashMap的isEmpty方法| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 | publicbooleanisEmpty() { longsum = 0L; finalSegment<K,V>[] segments =this.segments; for(intj =0; j < segments.length; ++j) { Segment<K,V> seg = segmentAt(segments, j); if(seg !=null) { if(seg.count !=0) returnfalse; sum += seg.modCount; } } if(sum != 0L) {// recheck unless no modifications for(intj =0; j < segments.length; ++j) { Segment<K,V> seg = segmentAt(segments, j); if(seg !=null) { if(seg.count !=0) returnfalse; sum -= seg.modCount; } } if(sum != 0L) returnfalse; } returntrue; } |
轉載於:https://my.oschina.net/zouqun/blog/405530