ConcurrentHashMap實現原理以及原始碼解析

ConcurrentHashMap是Java1.5中引用的一個執行緒安全的支援高併發的HashMap集合類。

1、執行緒不安全的HashMap
因為多執行緒環境下，使用Hashmap進行put操作會引起死迴圈，導致CPU利用率接近100%，所以在併發情況下不能使用HashMap。

2、效率低下的HashTable
HashTable容器使用synchronized來保證執行緒安全，但線上程競爭激烈的情況下HashTable的效率非常低下。
因為當一個執行緒訪問HashTable的同步方法時，其他執行緒訪問HashTable的同步方法時，可能會進入阻塞或輪詢狀態。
如執行緒1使用put進行新增元素，執行緒2不但不能使用put方法新增元素，並且也不能使用get方法來獲取元素，所以競爭越激烈效率越低。

3、鎖分段技術
HashTable容器在競爭激烈的併發環境下表現出效率低下的原因，是因為所有訪問HashTable的執行緒都必須競爭同一把鎖，
那假如容器裡有多把鎖，每一把鎖用於鎖容器其中一部分資料，那麼當多執行緒訪問容器裡不同資料段的資料時，執行緒間就不會存在鎖競爭，
從而可以有效的提高併發訪問效率，這就是ConcurrentHashMap所使用的鎖分段技術。
首先將資料分成一段一段的儲存，然後給每一段資料配一把鎖，當一個執行緒佔用鎖訪問其中一個段資料的時候，其他段的資料也能被其他執行緒訪問。
有些方法需要跨段，比如size()和containsValue()，它們可能需要鎖定整個表而而不僅僅是某個段，這需要按順序鎖定所有段，操作完畢後，又按順序釋放所有段的鎖。
這裡“按順序”是很重要的，否則極有可能出現死鎖，在ConcurrentHashMap內部，段陣列是final的，並且其成員變數實際上也是final的，
但是，僅僅是將陣列宣告為final的並不保證陣列成員也是final的，這需要實現上的保證。這可以確保不會出現死鎖，因為獲得鎖的順序是固定的。
    
4、 ConcurrentHashMap資料結構

ConcurrentHashMap 類中包含兩個靜態內部類 HashEntry 和 Segment。
HashEntry 用來封裝對映表的鍵 / 值對;Segment 用來充當鎖的角色，每個 Segment 物件守護整個雜湊對映表的若干個桶。
每個桶是由若干個 HashEntry 物件連結起來的連結串列。一個 ConcurrentHashMap 例項中包含由若干個 Segment 物件組成的陣列。
每個Segment守護者一個HashEntry數組裡的元素,當對HashEntry陣列的資料進行修改時，必須首先獲得它對應的Segment鎖。

5、HashEntry類

	static final class HashEntry<K,V> { 
			final K key;                 // 宣告 key 為 final 型
			final int hash;              // 宣告 hash 值為 final 型 
			volatile V value;           // 宣告 value 為 volatile 型
			final HashEntry<K,V> next;  // 宣告 next 為 final 型 


			HashEntry(K key, int hash, HashEntry<K,V> next, V value)  { 
				this.key = key; 
				this.hash = hash; 
				this.next = next; 
				this.value = value; 
			} 
	 }

每個HashEntry代表Hash表中的一個節點，在其定義的結構中可以看到，除了value值沒有定義final，其餘的都定義為final型別，我們知道Java中關鍵詞final修飾的域成為最終域。
用關鍵詞final修飾的變數一旦賦值，就不能改變，也稱為修飾的標識為常量。這就意味著我們刪除或者增加一個節點的時候，就必須從頭開始重新建立Hash鏈，因為next引用值需要改變。  
由於 HashEntry 的 next 域為 final 型，所以新節點只能在連結串列的表頭處插入。 例如將A,B,C插入空桶中，插入後的結構為： 

6、segment類

	static final class Segment<K,V> extends ReentrantLock implements Serializable {  
	 private static final long serialVersionUID = 2249069246763182397L;  
			 /** 
			  * 在本 segment 範圍內，包含的 HashEntry 元素的個數
			  * 該變數被宣告為 volatile 型,保證每次讀取到最新的資料
			  */  
			 transient volatile int count;  


			 /** 
			  *table 被更新的次數
			  */  
			 transient int modCount;  


			 /** 
			  * 當 table 中包含的 HashEntry 元素的個數超過本變數值時，觸發 table 的再雜湊
			  */  
			 transient int threshold;  


			 /** 
			  * table 是由 HashEntry 物件組成的陣列
			  * 如果雜湊時發生碰撞，碰撞的 HashEntry 物件就以連結串列的形式連結成一個連結串列
			  * table 陣列的陣列成員代表雜湊對映表的一個桶
			  * 每個 table 守護整個 ConcurrentHashMap 包含桶總數的一部分
			  * 如果併發級別為 16，table 則守護 ConcurrentHashMap 包含的桶總數的 1/16 
			  */  
			 transient volatile HashEntry<K,V>[] table;  


			 /** 
			  * 裝載因子
			  */  
			 final float loadFactor;  
	 }

Segment 類繼承於 ReentrantLock 類，從而使得 Segment 物件能充當鎖的角色。每個 Segment 物件用來守護其（成員物件 table 中）包含的若干個桶。
table 是一個由 HashEntry 物件組成的陣列。table 陣列的每一個數組成員就是雜湊對映表的一個桶。  
每一個 Segment 物件都有一個 count 物件來表示本 Segment 中包含的 HashEntry 物件的總數。  
之所以在每個 Segment 物件中包含一個計數器，而不是在 ConcurrentHashMap 中使用全域性的計數器，是為了避免出現“熱點域”而影響 ConcurrentHashMap 的併發性。
下圖是依次插入 ABC 三個 HashEntry 節點後，Segment 的結構示意圖。
 

7、ConcurrentHashMap 類  
預設的情況下，每個ConcurrentHashMap 類會建立16個併發的segment，每個segment裡面包含多個Hash表，每個Hash鏈都是有HashEntry節點組成的。  
如果鍵能均勻雜湊，每個 Segment 大約守護整個散列表中桶總數的 1/16。  

	public class ConcurrentHashMap<K, V> extends AbstractMap<K, V> 
		   implements ConcurrentMap<K, V>, Serializable { 
	 
	       /** 
		* 雜湊對映表的預設初始容量為 16，即初始預設為 16 個桶
		* 在建構函式中沒有指定這個引數時，使用本引數
		*/ 
	   static final     int DEFAULT_INITIAL_CAPACITY= 16; 
	 
	   /** 
		* 雜湊對映表的預設裝載因子為 0.75，該值是 table 中包含的 HashEntry 元素的個數與
	* table 陣列長度的比值
		* 當 table 中包含的 HashEntry 元素的個數超過了 table 陣列的長度與裝載因子的乘積時，
	* 將觸發 再雜湊
		* 在建構函式中沒有指定這個引數時，使用本引數
		*/ 
	   static final float DEFAULT_LOAD_FACTOR= 0.75f; 
	 
	   /** 
		* 散列表的預設併發級別為 16。該值表示當前更新執行緒的估計數
		* 在建構函式中沒有指定這個引數時，使用本引數
		*/ 
	   static final int DEFAULT_CONCURRENCY_LEVEL= 16; 
	 
	   /** 
		* segments 的掩碼值
		* key 的雜湊碼的高位用來選擇具體的 segment 
		*/ 
	   final int segmentMask; 
	 
	   /** 
		* 偏移量
		*/ 
	   final int segmentShift; 
	 
	   /** 
		* 由 Segment 物件組成的陣列
		*/ 
	   final Segment<K,V>[] segments; 
	 
	   /** 
		* 建立一個帶有指定初始容量、載入因子和併發級別的新的空對映。
		*/ 
	   public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) { 
		   if(!(loadFactor > 0) || initialCapacity < 0 || 
	concurrencyLevel <= 0) 
			   throw new IllegalArgumentException(); 
	 
		   if(concurrencyLevel > MAX_SEGMENTS) 
			   concurrencyLevel = MAX_SEGMENTS; 
	 
		   // 尋找最佳匹配引數（不小於給定引數的最接近的 2 次冪） 
		   int sshift = 0; 
		   int ssize = 1; 
		   while(ssize < concurrencyLevel) { 
			   ++sshift; 
			   ssize <<= 1; 
		   } 
		   segmentShift = 32 - sshift;       // 偏移量值
		   segmentMask = ssize - 1;           // 掩碼值 
		   this.segments = Segment.newArray(ssize);   // 建立陣列
	 
		   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) 
			   // 初始化每個陣列元素引用的 Segment 物件
	this.segments[i] = new Segment<K,V>(cap, loadFactor); 
	   } 
	 
	   /** 
		* 建立一個帶有預設初始容量 (16)、預設載入因子 (0.75) 和 預設併發級別 (16) 
	 * 的空雜湊對映表。
		*/ 
	   public ConcurrentHashMap() { 
		   // 使用三個預設引數，呼叫上面過載的建構函式來建立空雜湊對映表
	this(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL); 
	}	

8、 用分離鎖實現多個執行緒間的併發寫操作 

   插入資料後的ConcurrentHashMap的儲存形式

 
（1）Put方法的實現  

	 首先，根據 key 計算出對應的 hash 值：
        public V put(K key, V value) { 
       if (value == null)          //ConcurrentHashMap 中不允許用 null 作為對映值
           throw new NullPointerException(); 
       int hash = hash(key.hashCode());        // 計算鍵對應的雜湊碼
       // 根據雜湊碼找到對應的 Segment 
       return segmentFor(hash).put(key, hash, value, false); 
	}
	 根據 hash 值找到對應的 Segment：
	/** 
	* 使用 key 的雜湊碼來得到 segments 陣列中對應的 Segment 
	*/ 
	final Segment<K,V> segmentFor(int hash) { 
	// 將雜湊值右移 segmentShift 個位，並在高位填充 0 
	// 然後把得到的值與 segmentMask 相“與”
	// 從而得到 hash 值對應的 segments 陣列的下標值
	// 最後根據下標值返回雜湊碼對應的 Segment 物件
		return segments[(hash >>> segmentShift) & segmentMask]; 
	}
	在這個 Segment 中執行具體的 put 操作：
		V put(K key, int hash, V value, boolean onlyIfAbsent) { 
           lock();  // 加鎖，這裡是鎖定某個 Segment 物件而非整個 ConcurrentHashMap 
           try { 
               int c = count; 
 
               if (c++ > threshold)     // 如果超過再雜湊的閾值
                   rehash();              // 執行再雜湊，table 陣列的長度將擴充一倍
 
               HashEntry<K,V>[] tab = table; 
               // 把雜湊碼值與 table 陣列的長度減 1 的值相“與”
               // 得到該雜湊碼對應的 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;    // 設定 value 值
               } 
               else {                        // 鍵 / 值對不存在 
                   oldValue = null; 
                   ++modCount;         // 要新增新節點到連結串列中，所以 modCont 要加 1  
                   // 建立新節點，並新增到連結串列的頭部 
                   tab[index] = new HashEntry<K,V>(key, hash, first, value); 
                   count = c;               // 寫 count 變數
               } 
               return oldValue; 
           } finally { 
               unlock();                     // 解鎖
           } 
       `}

這裡的加鎖操作是針對（鍵的 hash 值對應的）某個具體的 Segment，鎖定的是該 Segment 而不是整個 ConcurrentHashMap。
因為插入鍵 / 值對操作只是在這個 Segment 包含的某個桶中完成，不需要鎖定整個ConcurrentHashMap。
此時，其他寫執行緒對另外 15 個Segment 的加鎖並不會因為當前執行緒對這個 Segment 的加鎖而阻塞。
同時，所有讀執行緒幾乎不會因本執行緒的加鎖而阻塞（除非讀執行緒剛好讀到這個 Segment 中某個 HashEntry 的 value 域的值為 null，此時需要加鎖後重新讀取該值）。

（2）Get方法的實現

	V get(Object key, int hash) { 
		if(count != 0) {       // 首先讀 count 變數
			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; 
					// 如果讀到 value 域為 null，說明發生了重排序，加鎖後重新讀取
					return readValueUnderLock(e); 
				} 
				e = e.next; 
			} 
		} 
		return null; 
	}
	 V readValueUnderLock(HashEntry<K,V> e) {  
		 lock();  
		 try {  
			 return e.value;  
		 } finally {  
			 unlock();  
		 }  
	 }

ConcurrentHashMap中的讀方法不需要加鎖，所有的修改操作在進行結構修改時都會在最後一步寫count 變數，通過這種機制保證get操作能夠得到幾乎最新的結構更新。

（3）Remove方法的實現

	V remove(Object key, int hash, Object value) { 
		lock(); //加鎖
		try{ 
			int c = count - 1; 
			HashEntry<K,V>[] tab = table; 
			//根據雜湊碼找到 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; 
					++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;      //寫 count 變數
				} 
			} 
			return oldValue; 
		} finally{ 
			unlock(); //解鎖
		} 
	}

整個操作是在持有段鎖的情況下執行的，空白行之前的行主要是定位到要刪除的節點e。
如果不存在這個節點就直接返回null，否則就要將e前面的結點複製一遍，尾結點指向e的下一個結點。
e後面的結點不需要複製，它們可以重用。

中間那個for迴圈是做什麼用的呢？從程式碼來看，就是將定位之後的所有entry克隆並拼回前面去，但有必要嗎？
每次刪除一個元素就要將那之前的元素克隆一遍？這點其實是由entry的不變性來決定的，仔細觀察entry定義，發現除了value，其他所有屬性都是用final來修飾的，
這意味著在第一次設定了next域之後便不能再改變它，取而代之的是將它之前的節點全都克隆一次。至於entry為什麼要設定為不變性，這跟不變性的訪問不需要同步從而節省時間有關。


（4）containsKey方法的實現，它不需要讀取值。

	boolean containsKey(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))  
					 return true;  
				 e = e.next;  
			 }  
		 }  
		 return false;  
	 } 

（5）size()
我們要統計整個ConcurrentHashMap裡元素的大小，就必須統計所有Segment裡元素的大小後求和。
Segment裡的全域性變數count是一個volatile變數，那麼在多執行緒場景下，我們是不是直接把所有Segment的count相加就可以得到整個ConcurrentHashMap大小了呢？
不是的，雖然相加時可以獲取每個Segment的count的最新值，但是拿到之後可能累加前使用的count發生了變化，那麼統計結果就不準了。
所以最安全的做法，是在統計size的時候把所有Segment的put，remove和clean方法全部鎖住，但是這種做法顯然非常低效。

因為在累加count操作過程中，之前累加過的count發生變化的機率非常小，所以ConcurrentHashMap的做法是先嚐試2次通過不鎖住Segment的方式來統計各個Segment大小，如果統計的過程中，容器的count發生了變化，則再採用加鎖的方式來統計所有Segment的大小。

那麼ConcurrentHashMap是如何判斷在統計的時候容器是否發生了變化呢？使用modCount變數，在put , remove和clean方法裡操作元素前都會將變數modCount進行加1，那麼在統計size前後比較modCount是否發生變化，從而得知容器的大小是否發生變化。



9、總結
1.在使用鎖來協調多執行緒間併發訪問的模式下，減小對鎖的競爭可以有效提高併發性。
有兩種方式可以減小對鎖的競爭：
減小請求同一個鎖的頻率。
減少持有鎖的時間。

2.ConcurrentHashMap 的高併發性主要來自於三個方面：

用分離鎖實現多個執行緒間的更深層次的共享訪問。
用 HashEntery 物件的不變性來降低執行讀操作的執行緒在遍歷連結串列期間對加鎖的需求。
通過對同一個 Volatile 變數的寫 / 讀訪問，協調不同執行緒間讀 / 寫操作的記憶體可見性。

使用分離鎖，減小了請求同一個鎖的頻率。







參考資料:
JDK API HashMap
Concurrenthashmap 原始碼 
探索 ConcurrentHashMap 高併發性的實現機制
http://www.cnblogs.com/ITtangtang/p/3948786.html

每天努力一點，每天都在進步。