前言

上篇文章介紹了 HashMap 原始碼後，在部落格平臺廣受好評，讓本來己經不打算更新這個系列的我，彷彿被打了一頓雞血。真的，被讀者認可的感覺，就是這麼奇妙。

然後，有讀者希望我能出一版 ConcurrentHashMap 的解析。所以，今天的這篇文章，我準備講述一下 ConcurrentHashMap 分別在JDK1.7和 JDK1.8 的原始碼。文章較長，建議小夥伴們可以先收藏再看哦~

說一下為什麼我要把原始碼解析寫的這麼詳細吧。一方面，可以記錄下當時自己的思考過程，也方便後續自己複習翻閱；另一方面，記錄下來還能夠幫助看到文章的小夥伴加深對原始碼的理解，簡直是一舉兩得的事情。

更正錯誤

另外，上一篇文章，有個錯誤點，卻沒有讀者給我指正出來。o(╥﹏╥)o 。因此，我只能自己在此更正一下。見下面截圖，

put 方法，在新值替換舊值那裡，應該是隻有一種情況的，e 不包括新值。圖中的方框也標註出來了。因為，判斷 e=p.next==null ， 然後新的節點是賦值給 p.next 了，並沒有賦值給 e，此時 e 依舊是空的。所以 e!=null，代表當前的 e 是已經存在的舊值。

文章編寫過程，難免出現作者考慮不周的地方，如果有朋友發現有錯誤的地方，還請不吝賜教，指正出來。知錯能改，善莫大焉，對於技術，我們應該懷有一顆嚴謹的心態~

文章目錄

這篇文章，我打算從以下幾個方面來講。

1）多執行緒下的 HashMap 有什麼問題？

2）怎樣保證執行緒安全，為什麼選用 ConcurrentHashMap？

3）ConcurrentHashMap 1.7 原始碼解析

• 底層儲存結構
• 常用變數
• 建構函式
• put() 方法
• ensureSegment() 方法
• scanAndLockForPut() 方法
• rehash() 擴容機制
• get() 獲取元素方法
• remove() 方法
• size() 方法是怎麼統計元素個數的

4）ConcurrentHashMap 1.8 原始碼解析

• put()方法詳解
• initTable()初始化表
• addCount()方法
• fullAddCount()方法
• transfer()是怎樣擴容和遷移元素的
• helpTransfer()方法幫助遷移元素

多執行緒下 HashMap 有什麼問題？

在上一篇文章中，已經講解了 HashMap 1.7 死迴圈的成因，也正因為如此，我們才說 HashMap 在多執行緒下是不安全的。但是，在JDK1.8 的 HashMap 改為採用尾插法，已經不存在死迴圈的問題了，為什麼也會執行緒不安全呢？

我們以 put 方法為例（1.8），

假如現在有兩個執行緒都執行到了上圖中的劃線處。當執行緒一判斷為空之後，CPU 時間片到了，被掛起。執行緒二也執行到此處判斷為空，繼續執行下一句，建立了一個新節點，插入到此下標位置。然後，執行緒一解掛，同樣認為此下標的元素為空，因此也建立了一個新節點放在此下標處，因此造成了元素的覆蓋。

所以，可以看到不管是 JDK1.7 還是 1.8 的 HashMap 都存線上程安全的問題。那麼，在多執行緒環境下，應該怎樣去保證執行緒安全呢？

怎樣保證執行緒安全，為什麼選用 ConcurrentHashMap？

首先，你可能想到，在多執行緒環境下用 Hashtable 來解決執行緒安全的問題。這樣確實是可以的，但是同樣的它也有缺點，我們看下最常用的 put 方法和 get 方法。

可以看到，不管是往 map 裡邊新增元素還是獲取元素，都會用 synchronized 關鍵字加鎖。當有多個元素之前存在資源競爭時，只能有一個執行緒可以獲取到鎖，操作資源。更不能忍的是，一個簡單的讀取操作，互相之間又不影響，為什麼也不能同時進行呢？

所以，hashtable 的缺點顯而易見，它不管是 get 還是 put 操作，都是鎖住了整個 table，效率低下，因此 並不適合高併發場景。

也許，你還會想起來一個集合工具類 Collections，生成一個SynchronizedMap。其實，它和 Hashtable 差不多，同樣的原因，鎖住整張表，效率低下。

所以，思考一下，既然鎖住整張表的話，併發效率低下，那我把整張表分成 N 個部分，並使元素儘量均勻的分佈到每個部分中，分別給他們加鎖，互相之間並不影響，這種方式豈不是更好 。這就是在 JDK1.7 中 ConcurrentHashMap 採用的方案，被叫做鎖分段技術，每個部分就是一個 Segment（段）。

但是，在JDK1.8中，完全重構了，採用的是 Synchronized + CAS ，把鎖的粒度進一步降低，而放棄了 Segment 分段。（此時的 Synchronized 已經升級了，效率得到了很大提升，鎖升級可以瞭解一下）

ConcurrentHashMap 1.7 原始碼解析

我們看下在 JDK1.7中 ConcurrentHashMap 是怎麼實現的。牆裂建議，在本文之前瞭解一下多執行緒的基本知識，如JMM記憶體模型，volatile關鍵字作用，CAS和自旋，ReentranLock重入鎖。

底層儲存結構

在 JDK1.7中，本質上還是採用連結串列+陣列的形式儲存鍵值對的。但是，為了提高併發，把原來的整個 table 劃分為 n 個 Segment 。所以，從整體來看，它是一個由 Segment 組成的陣列。然後，每個 Segment 裡邊是由 HashEntry 組成的陣列，每個 HashEntry之間又可以形成連結串列。我們可以把每個 Segment 看成是一個小的 HashMap，其內部結構和 HashMap 是一模一樣的。

當對某個 Segment 加鎖時，如圖中 Segment2，並不會影響到其他 Segment 的讀寫。每個 Segment 內部自己操作自己的資料。這樣一來，我們要做的就是儘可能的讓元素均勻的分佈在不同的 Segment中。最理想的狀態是，所有執行的執行緒操作的元素都是不同的 Segment，這樣就可以降低鎖的競爭。

廢話了這麼多，還是來看底層原始碼吧，因為所有的思想都在程式碼裡體現。借用 Linus的一句話，“No BB . Show me the code ” （改編版，哈哈）

常用變數

先看下 1.7 中常用的變數和內部類都有哪些，這有助於我們瞭解 ConcurrentHashMap 的整體結構。

//預設初始化容量，這個和 HashMap中的容量是一個概念，表示的是整個 Map的容量
static final int DEFAULT_INITIAL_CAPACITY = 16;

//預設載入因子
static final float DEFAULT_LOAD_FACTOR = 0.75f;

//預設的併發級別，這個引數決定了 Segment 陣列的長度
static final int DEFAULT_CONCURRENCY_LEVEL = 16;

//最大的容量
static final int MAXIMUM_CAPACITY = 1 << 30;

//每個Segment中table陣列的最小長度為2，且必須是2的n次冪。
//由於每個Segment是懶載入的，用的時候才會初始化，因此為了避免使用時立即調整大小，設定了最小容量2
static final int MIN_SEGMENT_TABLE_CAPACITY = 2;

//用於限制Segment數量的最大值，必須是2的n次冪
static final int MAX_SEGMENTS = 1 << 16; // slightly conservative

//在size方法和containsValue方法，會優先採用樂觀的方式不加鎖，直到重試次數達到2，才會對所有Segment加鎖
//這個值的設定，是為了避免無限次的重試。後邊size方法會詳講怎麼實現樂觀機制的。
static final int RETRIES_BEFORE_LOCK = 2;

//segment掩碼值，用於根據元素的hash值定位所在的 Segment 下標。後邊會細講
final int segmentMask;

//和 segmentMask 配合使用來定位 Segment 的陣列下標，後邊講。
final int segmentShift;

// Segment 組成的陣列，每一個 Segment 都可以看做是一個特殊的 HashMap
final Segment<K,V>[] segments;

//Segment 物件，繼承自 ReentrantLock 可重入鎖。
//其內部的屬性和方法和 HashMap 神似，只是多了一些拓展功能。
static final class Segment<K,V> extends ReentrantLock implements Serializable {
	
	//這是在 scanAndLockForPut 方法中用到的一個引數，用於計算最大重試次數
	//獲取當前可用的處理器的數量，若大於1，則返回64，否則返回1。
	static final int MAX_SCAN_RETRIES =
		Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;

	//用於表示每個Segment中的 table，是一個用HashEntry組成的陣列。
	transient volatile HashEntry<K,V>[] table;

	//Segment中的元素個數，每個Segment單獨計數（下邊的幾個引數同樣的都是單獨計數）
	transient int count;

	//每次 table 結構修改時，如put，remove等，此變數都會自增
	transient int modCount;

	//當前Segment擴容的閾值，同HashMap計算方法一樣也是容量乘以載入因子
	//需要知道的是，每個Segment都是單獨處理擴容的，互相之間不會產生影響
	transient int threshold;

	//載入因子
	final float loadFactor;

	//Segment建構函式
	Segment(float lf, int threshold, HashEntry<K,V>[] tab) {
		this.loadFactor = lf;
		this.threshold = threshold;
		this.table = tab;
	}
	
	...
	// put(),remove(),rehash() 方法都在此類定義
}

// HashEntry，存在於每個Segment中，它就類似於HashMap中的Node，用於儲存鍵值對的具體資料和維護單向連結串列的關係
static final class HashEntry<K,V> {
	//每個key通過雜湊運算後的結果，用的是 Wang/Jenkins hash 的變種演算法，此處不細講，感興趣的可自行查閱相關資料
	final int hash;
	final K key;
	//value和next都用 volatile 修飾，用於保證記憶體可見性和禁止指令重排序
	volatile V value;
	//指向下一個節點
	volatile HashEntry<K,V> next;

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

建構函式

ConcurrentHashMap 有五種建構函式，但是最終都會呼叫同一個建構函式，所以只需要搞明白這一個核心的建構函式就可以了。

PS: 文章註釋中 (1)(2)(3) 等序號都是用來方便做標記，不是計算值

public ConcurrentHashMap(int initialCapacity,
                             float loadFactor, int concurrencyLevel) {
	//檢驗引數是否合法。值得說的是，併發級別一定要大於0，否則就沒辦法實現分段鎖了。
	if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
		throw new IllegalArgumentException();
	//併發級別不能超過最大值
	if (concurrencyLevel > MAX_SEGMENTS)
		concurrencyLevel = MAX_SEGMENTS;
	// Find power-of-two sizes best matching arguments
	//偏移量，是為了對hash值做位移操作，計算元素所在的Segment下標，put方法詳講
	int sshift = 0;
	//用於設定最終Segment陣列的長度，必須是2的n次冪
	int ssize = 1;
	//這裡就是計算 sshift 和 ssize 值的過程  (1) 
	while (ssize < concurrencyLevel) {
		++sshift;
		ssize <<= 1;
	}
	this.segmentShift = 32 - sshift;
	//Segment的掩碼
	this.segmentMask = ssize - 1;
	if (initialCapacity > MAXIMUM_CAPACITY)
		initialCapacity = MAXIMUM_CAPACITY;
	//c用於輔助計算cap的值   (2)
	int c = initialCapacity / ssize;
	if (c * ssize < initialCapacity)
		++c;
	// cap 用於確定某個Segment的容量，即Segment中HashEntry陣列的長度
	int cap = MIN_SEGMENT_TABLE_CAPACITY;
	//(3)
	while (cap < c)
		cap <<= 1;
	// create segments and segments[0]
	//這裡用 loadFactor做為載入因子，cap乘以載入因子作為擴容閾值，建立長度為cap的HashEntry陣列，
	//三個引數，建立一個Segment物件，儲存到S0物件中。後邊在 ensureSegment 方法會用到S0作為原型物件去建立對應的Segment。
	Segment<K,V> s0 =
		new Segment<K,V>(loadFactor, (int)(cap * loadFactor),
						 (HashEntry<K,V>[])new HashEntry[cap]);
	//創建出長度為 ssize 的一個 Segment陣列
	Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];
	//把S0存到Segment陣列中去。在這裡，我們就可以發現，此時只是建立了一個Segment陣列，
	//但是並沒有把陣列中的每個Segment物件創建出來，僅僅建立了一個Segment用來作為原型物件。
	UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
	this.segments = ss;
}				

上邊的註釋中留了 (1)(2)(3) 三個地方還沒有細說。我們現在假設一組資料，把涉及到的幾個變數計算出來，就能明白這些引數的含義了。

//假設呼叫了預設構造，都用的是預設引數，即 initialCapacity 和 concurrencyLevel 都是16
//(1)  sshift 和 ssize 值的計算過程為，每次迴圈，都會把 sshift 自增1，並且 ssize 左移一位，即乘以2，
//直到 ssize 的值大於等於 concurrencyLevel 的值 16。
sshfit=0,1,2,3,4
ssize=1,2,4,8,16
//可以看到，初始他們的值分別是0和1，最終結果是4和16
//sshfit是為了輔助計算segmentShift值，ssize是為了確定Segment陣列長度。
//(2)  此時,計算c的值，
c = 16/16 = 1;
//判斷 c * 16 < 16 是否為真，真的話 c 自增1，此處為false，因此 c的值為1不變。
//(3)  此時，由於c為1， cap為2 ，因此判斷 cap < c 為false，最終cap為2。
//總結一下，以上三個步驟，最終都是為了確定以下幾個關鍵引數的值，
//確定 segmentShift ，這個用於後邊計算hash值的偏移量，此處即為 32-4=28，
//確定 ssize，必須是一個大於等於 concurrencyLevel 的一個2的n次冪值
//確定 cap，必須是一個大於等於2的一個2的n次冪值
//感興趣的小夥伴，還可以用另外幾組引數來計算上邊的引數值，可以加深理解引數的含義。
//例如initialCapacity和concurrencyLevel分別傳入10和5，或者傳入33和16

put()方法

put 方法的總體流程是，

1. 通過雜湊演算法計算出當前 key 的 hash 值
2. 通過這個 hash 值找到它所對應的 Segment 陣列的下標
3. 再通過 hash 值計算出它在對應 Segment 的 HashEntry陣列 的下標
4. 找到合適的位置插入元素

//這是Map的put方法
public V put(K key, V value) {
	Segment<K,V> s;
	//不支援value為空
	if (value == null)
		throw new NullPointerException();
	//通過 Wang/Jenkins 演算法的一個變種演算法，計算出當前key對應的hash值
	int hash = hash(key);
	//上邊我們計算出的 segmentShift為28，因此hash值右移28位，說明此時用的是hash的高4位，
	//然後把它和掩碼15進行與運算，得到的值一定是一個 0000 ~ 1111 範圍內的值，即 0~15 。
	int j = (hash >>> segmentShift) & segmentMask;
	//這裡是用Unsafe類的原子操作找到Segment陣列中j下標的 Segment 物件
	if ((s = (Segment<K,V>)UNSAFE.getObject          // nonvolatile; recheck
		 (segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegment
		//初始化j下標的Segment
		s = ensureSegment(j);
	//在此Segment中新增元素
	return s.put(key, hash, value, false);
}

上邊有一個這樣的方法， UNSAFE.getObject (segments, (j << SSHIFT) + SBASE。它是為了通過Unsafe這個類，找到 j 最新的實際值。這個計算 (j << SSHIFT) + SBASE ，在後邊非常常見，我們只需要知道它代表的是 j 的一個偏移量，通過偏移量，就可以得到 j 的實際值。可以類比，AQS 中的 CAS 操作。 Unsafe中的操作，都需要一個偏移量，看下圖，

(j << SSHIFT) + SBASE 就相當於圖中的 stateOffset偏移量。只不過圖中是 CAS 設定新值，而我們這裡是取 j 的最新值。 後邊很多這樣的計算方式，就不贅述了。接著看 s.put 方法，這才是最終確定元素位置的方法。

//Segment中的 put 方法
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
	//這裡通過tryLock嘗試加鎖，如果加鎖成功，返回null，否則執行 scanAndLockForPut方法
	//這裡說明一下，tryLock 和 lock 是 ReentrantLock 中的方法，
	//區別是 tryLock 不會阻塞，搶鎖成功就返回true，失敗就立馬返回false，
	//而 lock 方法是，搶鎖成功則返回，失敗則會進入同步佇列，阻塞等待獲取鎖。
	HashEntry<K,V> node = tryLock() ? null :
		scanAndLockForPut(key, hash, value);
	V oldValue;
	try {
		//當前Segment的table陣列
		HashEntry<K,V>[] tab = table;
		//這裡就是通過hash值，與tab陣列長度取模，找到其所在HashEntry陣列的下標
		int index = (tab.length - 1) & hash;
		//當前下標位置的第一個HashEntry節點
		HashEntry<K,V> first = entryAt(tab, index);
		for (HashEntry<K,V> e = first;;) {
			//如果第一個節點不為空
			if (e != null) {
				K k;
				//並且第一個節點，就是要插入的節點，則替換value值，否則繼續向後查詢
				if ((k = e.key) == key ||
					(e.hash == hash && key.equals(k))) {
					//替換舊值
					oldValue = e.value;
					if (!onlyIfAbsent) {
						e.value = value;
						++modCount;
					}
					break;
				}
				e = e.next;
			}
			//說明當前index位置不存在任何節點，此時first為null，
			//或者當前index存在一條連結串列，並且已經遍歷完了還沒找到相等的key，此時first就是連結串列第一個元素
			else {
				//如果node不為空，則直接頭插
				if (node != null)
					node.setNext(first);
				//否則，建立一個新的node，並頭插
				else
					node = new HashEntry<K,V>(hash, key, value, first);
				int c = count + 1;
				//如果當前Segment中的元素大於閾值，並且tab長度沒有超過容量最大值，則擴容
				if (c > threshold && tab.length < MAXIMUM_CAPACITY)
					rehash(node);
				//否則，就把當前node設定為index下標位置新的頭結點
				else
					setEntryAt(tab, index, node);
				++modCount;
				//更新count值
				count = c;
				//這種情況說明舊值肯定為空
				oldValue = null;
				break;
			}
		}
	} finally {
		//需要注意ReentrantLock必須手動解鎖
		unlock();
	}
	//返回舊值
	return oldValue;
}

這裡說明一下計算 Segment 陣列下標和計算 HashEntry 陣列下標的不同點：

//下邊的hash值是通過雜湊運算後的hash值，不是hashCode
//計算 Segment 下標
 (hash >>> segmentShift) & segmentMask 
 //計算 HashEntry 陣列下標
 (tab.length - 1) & hash

思考一下，為什麼它們的演算法不一樣呢？ 計算 Segment 陣列下標是用的 hash值高几位（這裡以高 4 位為例）和掩碼做與運算，而計算 HashEntry 陣列下標是直接用的 hash 值和陣列長度減1做與運算。

我的理解是，這是為了儘量避免當前 hash 值計算出來的 Segment 陣列下標和計算出來的 HashEntry 陣列下標趨於相同。簡單說，就是為了避免分配到同一個 Segment 中的元素扎堆現象，即避免它們都被分配到同一條連結串列上，導致連結串列過長。同時，也是為了減少併發。下面做一個運算，幫助理解一下（假設不用高 4 位運算，而是正常情況都用低位做運算）。

//我們以併發級別16，HashEntry陣列容量 4 為例，則它們參與運算的掩碼分別為 15 和 3
//hash值
0110 1101 0110 1111 0110 1110 0010 0010
//segmentMask = 15   ，標記為 (1)
0000 0000 0000 0000 0000 0000 0000 1111
//tab.length - 1 = 3     ，標記為 (2)
0000 0000 0000 0000 0000 0000 0000 0011
//用 hash 分別和 15 ，3 做與運算，會發現得到的結果是一樣，都是十進位制 2.
//這表明，當前 hash值被分配到下標為 2 的 Segment 中，同時，被分配到下標為 2 的 HashEntry 陣列中
//現在若有另外一個 hash 值 h2，和第一個hash值，高位不同，但是低4位相同，
1010 1101 0110 1111 0110 1110 0010 0010
//我們會發現，最後它也會被分配到下標為 2 的 Segment 和 HashEntry 陣列，就會和第一個元素形成連結串列。
//所以，為了避免這種扎堆現象，讓元素儘量均勻分配，就讓 hash 的高 4 位和 (1)處做與 運算，而用低位和 (2)處做與運算
//這樣計算後，它們所在的Segment下標分別為 6(0110), 10(1010)，即使它們在HashEntry陣列中的下標都為 2(0010)，也無所謂
//因為它們並不在一個 Segment 中，也就不會在同一個 HashEntry 陣列中，更不會形成連結串列。
//更重要的是，它們不會有併發，因為在各自不同的 Segment 自己操作自己的加鎖解鎖，互不影響

可能有的小夥伴就會打岔了，那如果兩個 hash 值，低位和高位都相同，怎麼辦呢。如果是這樣，我只能說，這個 hash 演算法也太爛了吧。（這裡的 hash 演算法也會盡量避免這種情況，當然只是減少機率，並不能杜絕）

我有個大膽的想法，這裡的高低位不同的計算方式，是不是後邊 1.8 HashMap 讓 hash 高低位做異或運算的引子呢？不得而知。。

put 方法比較簡單，只要能看懂 HashMap 中的 put 方法，這裡也沒問題。主要是它呼叫的子方法比較複雜，下邊一個一個講解。

ensureSegment()方法

回到 Map的 put 方法，判斷 j 下標的 Segment為空後，則需要呼叫此方法，初始化一個 Segment 物件，以確保拿到的物件一定是不為空的，否則無法執行s.put了。

//k為 (hash >>> segmentShift) & segmentMask 演算法計算出來的值
private Segment<K,V> ensureSegment(int k) {
	final Segment<K,V>[] ss = this.segments;
	//u代表 k 的偏移量，用於通過 UNSAFE 獲取主記憶體最新的實際 K 值
	long u = (k << SSHIFT) + SBASE; // raw offset
	Segment<K,V> seg;
	//從記憶體中取到最新的下標位置的 Segment 物件，判斷是否為空，(1)
	if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {
		//之前建構函式說了，s0是作為一個原型物件，用於建立新的 Segment 物件
		Segment<K,V> proto = ss[0]; // use segment 0 as prototype
		//容量
		int cap = proto.table.length;
		//載入因子
		float lf = proto.loadFactor;
		//擴容閾值
		int threshold = (int)(cap * lf);
		//把 Segment 對應的 HashEntry 陣列先創建出來
		HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];
		//再次檢查 K 下標位置的 Segment 是否為空， (2)
		if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
			== null) { // recheck
			//此處把 Segment 物件創建出來，並賦值給 s，
			Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);
			//迴圈檢查 K 下標位置的 Segment 是否為空， (3)
			//若不為空，則說明有其它執行緒搶先建立成功，並且已經成功同步到主記憶體中了，
			//則把它取出來，並返回
			while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
				   == null) {
				//CAS，若當前下標的Segment物件為空，就把它替換為最新創建出來的 s 物件。
				//若成功，就跳出迴圈，否則，就一直自旋直到成功，或者 seg 不為空（其他執行緒成功導致）。
				if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
					break;
			}
		}
	}
	return seg;
}

可以發現，我標註了上邊 (1)(2)(3) 個地方，每次都判斷最新的Segment是否為空。可能有的小夥伴就會迷惑，為什麼做這麼多次判斷，我直接去自旋不就好了，反正最後都要自旋的。

我的理解是，在多執行緒環境下，因為不確定是什麼時候會有其它執行緒 CAS 成功，有可能發生在以上的任意時刻。所以，只要發現一旦記憶體中的物件已經存在了，則說明已經有其它執行緒把Segment物件建立好，並CAS成功同步到主記憶體了。此時，就可以直接返回，而不需要往下執行了。這樣做，是為了程式碼執行效率考慮。

scanAndLockForPut()方法

put 方法第一步搶鎖失敗之後，就會執行此方法，

private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
	//根據hash值定位到它對應的HashEntry陣列的下標位置，並找到連結串列的第一個節點
	//注意，這個操作會從主記憶體中獲取到最新的狀態，以確保獲取到的first是最新值
	HashEntry<K,V> first = entryForHash(this, hash);
	HashEntry<K,V> e = first;
	HashEntry<K,V> node = null;
	//重試次數，初始化為 -1
	int retries = -1; // negative while locating node
	//若搶鎖失敗，就一直迴圈，直到成功獲取到鎖。有三種情況
	while (!tryLock()) {
		HashEntry<K,V> f; // to recheck first below
		//1.若 retries 小於0，
		if (retries < 0) {
			if (e == null) {
				//若 e 節點和 node 都為空，則建立一個 node 節點。這裡只是預測性的建立一個node節點
				if (node == null) // speculatively create node
					node = new HashEntry<K,V>(hash, key, value, null);
				retries = 0;
			}
			//如當前遍歷到的 e 節點不為空，則判斷它的key是否等於傳進來的key，若是則把 retries 設為0
			else if (key.equals(e.key))
				retries = 0;
			//否則，繼續向後遍歷節點
			else
				e = e.next;
		}
		//2.若是重試次數超過了最大嘗試次數，則呼叫lock方法加鎖。表明不再重試，我下定決心了一定要獲取到鎖。
		//要麼當前執行緒可以獲取到鎖，要麼獲取不到就去排隊等待獲取鎖。獲取成功後，再 break。
		else if (++retries > MAX_SCAN_RETRIES) {
			lock();
			break;
		}
		//3.若 retries 的值為偶數，並且從記憶體中再次獲取到最新的頭節點，判斷若不等於first
		//則說明有其他執行緒修改了當前下標位置的頭結點，於是需要更新頭結點資訊。
		else if ((retries & 1) == 0 &&
				 (f = entryForHash(this, hash)) != first) {
			//更新頭結點資訊，並把重試次數重置為 -1，繼續下一次迴圈，從最新的頭結點遍歷當前連結串列。
			e = first = f; // re-traverse if entry changed
			retries = -1;
		}
	}
	return node;
}

這個方法邏輯比較複雜，會一直迴圈嘗試獲取鎖，若獲取成功，則返回。否則的話，每次迴圈時，都會同時遍歷當前連結串列。若遍歷完了一次，還沒找到和key相等的節點，就會預先建立一個節點。注意，這裡只是預測性的建立一個新節點，也有可能在這之前，就已經獲取鎖成功了。

同時，當重試次每偶數次時，就會檢查一次當前最新的頭結點是否被改變。因為若有變化的話，還需要從最新的頭結點開始遍歷連結串列。

還有一種情況，就是迴圈次數達到了最大限制，則停止迴圈，用阻塞的方式去獲取鎖。這時，也就停止了遍歷連結串列的動作，當前執行緒也不會再做其他預熱(warm up)的事情。

關於為什麼預測性的建立新節點，原始碼中原話是這樣的：

Since traversal speed doesn't matter, we might as well help warm up the associated code and accesses as well.

解釋一下就是，因為遍歷速度無所謂，所以，我們可以預先(warm up)做一些相關聯程式碼的準備工作。這裡相關聯程式碼，指的就是迴圈中，在獲取鎖成功或者呼叫 lock 方法之前做的這些事情，當然也包括建立新節點。

在put 方法中可以看到，有一句是判斷 node 是否為空，若建立了，就直接頭插。否則的話，它也會自己建立這個新節點。

scanAndLockForPut 這個方法可以確保返回時，當前執行緒一定是獲取到鎖的狀態。

rehash()方法

當 put 方法時，發現元素個數超過了閾值，則會擴容。需要注意的是，每個Segment只管它自己的擴容，互相之間並不影響。換句話說，可以出現這個 Segment的長度為2，另一個Segment的長度為4的情況（只要是2的n次冪）。

//node為建立的新節點
private void rehash(HashEntry<K,V> node) {
	//當前Segment中的舊錶
	HashEntry<K,V>[] oldTable = table;
	//舊的容量
	int oldCapacity = oldTable.length;
	//新容量為舊容量的2倍
	int newCapacity = oldCapacity << 1;
	//更新新的閾值
	threshold = (int)(newCapacity * loadFactor);
	//用新的容量建立一個新的 HashEntry 陣列
	HashEntry<K,V>[] newTable =
		(HashEntry<K,V>[]) new HashEntry[newCapacity];
	//當前的掩碼，用於計算節點在新陣列中的下標
	int sizeMask = newCapacity - 1;
	//遍歷舊錶
	for (int i = 0; i < oldCapacity ; i++) {
		HashEntry<K,V> e = oldTable[i];
		//如果e不為空，說明當前連結串列不為空
		if (e != null) {
			HashEntry<K,V> next = e.next;
			//計算hash值再新陣列中的下標位置
			int idx = e.hash & sizeMask;
			//如果e不為空，且它的下一個節點為空，則說明這條連結串列只有一個節點，
			//直接把這個節點放到新陣列的對應下標位置即可
			if (next == null)   //  Single node on list
				newTable[idx] = e;
			//否則，處理當前連結串列的節點遷移操作
			else { // Reuse consecutive sequence at same slot
				//記錄上一次遍歷到的節點
				HashEntry<K,V> lastRun = e;
				//對應上一次遍歷到的節點在新陣列中的新下標
				int lastIdx = idx;
				for (HashEntry<K,V> last = next;
					 last != null;
					 last = last.next) {
					//計算當前遍歷到的節點的新下標
					int k = last.hash & sizeMask;
					//若 k 不等於 lastIdx，則說明此次遍歷到的節點和上次遍歷到的節點不在同一個下標位置
					//需要把 lastRun 和 lastIdx 更新為當前遍歷到的節點和下標值。
					//若相同，則不處理，繼續下一次 for 迴圈。
					if (k != lastIdx) {
						lastIdx = k;
						lastRun = last;
					}
				}
				//把和 lastRun 節點的下標位置相同的連結串列最末尾的幾個連續的節點放到新陣列的對應下標位置
				newTable[lastIdx] = lastRun;
				//再把剩餘的節點，複製到新陣列
				//從舊陣列的頭結點開始遍歷，直到 lastRun 節點，因為 lastRun節點後邊的節點都已經遷移完成了。
				for (HashEntry<K,V> p = e; p != lastRun; p = p.next) {
					V v = p.value;
					int h = p.hash;
					int k = h & sizeMask;
					HashEntry<K,V> n = newTable[k];
					//用的是複製節點資訊的方式，並不是把原來的節點直接遷移，區別於lastRun處理方式
					newTable[k] = new HashEntry<K,V>(h, p.key, v, n);
				}
			}
		}
	}
	//所有節點都遷移完成之後，再處理傳進來的新的node節點，把它頭插到對應的下標位置
	int nodeIndex = node.hash & sizeMask; // add the new node
	//頭插node節點
	node.setNext(newTable[nodeIndex]);
	newTable[nodeIndex] = node;
	//更新當前Segment的table資訊
	table = newTable;
}

上邊的遷移過程和 lastRun 和 lastIdx 變數可能不太好理解，我畫個圖就明白了。以其中一條連結串列處理方式為例。

從頭結點開始向後遍歷，找到當前連結串列的最後幾個下標相同的連續的節點。如上圖，雖然開頭出現了有兩個節點的下標都是 k2, 但是中間出現一個不同的下標 k1，打斷了下標連續相同，因此從下一個k2，又重新開始算。好在後邊三個連續的節點下標都是相同的，因此倒數第三個節點被標記為 lastRun，且變數無變化。

從lastRun節點到尾結點的這部分就可以整體遷移到新陣列的對應下標位置了，因為它們的下標都是相同的，可以這樣統一處理。

另外從頭結點到 lastRun 之前的節點，無法統一處理，只能一個一個去複製了。且注意，這裡不是直接遷移，而是複製節點到新的陣列，舊的節點會在不久的將來，因為沒有引用指向，被 JVM 垃圾回收處理掉。

（不知道為啥這個方法名起為 rehash，其實擴容時 hash 值並沒有重新計算，變化的只是它們所在的下標而已。我猜測，可能是，借用了 1.7 HashMap 中的說法吧。。。）

get()

put 方法搞明白了之後，其實 get 方法就很好理解了。也是先定位到 Segment，然後再定位到 HashEntry 。

public V get(Object key) {
	Segment<K,V> s; // manually integrate access methods to reduce overhead
	HashEntry<K,V>[] tab;
	//計算hash值
	int h = hash(key);
	//同樣的先定位到 key 所在的Segment ，然後從主記憶體中取出最新的節點
	long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
	if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
		(tab = s.table) != null) {
		//若Segment不為空，且連結串列也不為空，則遍歷查詢節點
		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;
			//找到則返回它的 value 值，否則返回 null
			if ((k = e.key) == key || (e.hash == h && key.equals(k)))
				return e.value;
		}
	}
	return null;
}

remove()

remove 方法和 put 方法類似，也不用做過多特殊的介紹，

public V remove(Object key) {
	int hash = hash(key);
	//定位到Segment
	Segment<K,V> s = segmentForHash(hash);
	//若 s為空，則返回 null，否則執行 remove
	return s == null ? null : s.remove(key, hash, null);
}

public boolean remove(Object key, Object value) {
	int hash = hash(key);
	Segment<K,V> s;
	return value != null && (s = segmentForHash(hash)) != null &&
		s.remove(key, hash, value) != null;
}

final V remove(Object key, int hash, Object value) {
	//嘗試加鎖，若失敗，則執行 scanAndLock ，此方法和 scanAndLockForPut 方法類似
	if (!tryLock())
		scanAndLock(key, hash);
	V oldValue = null;
	try {
		HashEntry<K,V>[] tab = table;
		int index = (tab.length - 1) & hash;
		//從主記憶體中獲取對應 table 的最新的頭結點
		HashEntry<K,V> e = entryAt(tab, index);
		HashEntry<K,V> pred = null;
		while (e != null) {
			K k;
			HashEntry<K,V> next = e.next;
			//匹配到 key
			if ((k = e.key) == key ||
				(e.hash == hash && key.equals(k))) {
				V v = e.value;
				// value 為空，或者 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();
	}
	return oldValue;
}

size()

size 方法需要重點說明一下。愛思考的小夥伴可能就會想到，併發情況下，有可能在統計期間，陣列元素個數不停的變化，而且，整個表還被分成了 N個 Segment，怎樣統計才能保證結果的準確性呢？ 我們一起來看下吧。

public int size() {
	// Try a few times to get accurate count. On failure due to
	// continuous async changes in table, resort to locking.
	//segment陣列
	final Segment<K,V>[] segments = this.segments;
	//統計所有Segment中元素的總個數
	int size;
	//如果size大小超過32位，則標記為溢位為true
	boolean overflow; 
	//統計每個Segment中的 modcount 之和
	long sum;         
	//上次記錄的 sum 值
	long last = 0L;   
	//重試次數，初始化為 -1
	int retries = -1; 
	try {
		for (;;) {
			//如果超過重試次數，則不再重試，而是把所有Segment都加鎖，再統計 size
			if (retries++ == RETRIES_BEFORE_LOCK) {
				for (int j = 0; j < segments.length; ++j)
					//強制加鎖
					ensureSegment(j).lock(); // force creation
			}
			sum = 0L;
			size = 0;
			overflow = false;
			//遍歷所有Segment
			for (int j = 0; j < segments.length; ++j) {
				Segment<K,V> seg = segmentAt(segments, j);
				//若當前遍歷到的Segment不為空，則統計它的 modCount 和 count 元素個數
				if (seg != null) {
					//累加當前Segment的結構修改次數，如put，remove等操作都會影響modCount
					sum += seg.modCount;
					int c = seg.count;
					//若當前Segment的元素個數 c 小於0 或者 size 加上 c 的結果小於0，則認為溢位
					//因為若超過了 int 最大值，就會返回負數
					if (c < 0 || (size += c) < 0)
						overflow = true;
				}
			}
			//當此次嘗試，統計的 sum 值和上次統計的值相同，則說明這段時間內，
			//並沒有任何一個 Segment 的結構發生改變，就可以返回最後的統計結果
			if (sum == last)
				break;
			//不相等，則說明有 Segment 結構發生了改變，則記錄最新的結構變化次數之和 sum，
			//並賦值給 last，用於下次重試的比較。
			last = sum;
		}
	} finally {
		//如果超過了指定重試次數，則說明表中的所有Segment都被加鎖了，因此需要把它們都解鎖
		if (retries > RETRIES_BEFORE_LOCK) {
			for (int j = 0; j < segments.length; ++j)
				segmentAt(segments, j).unlock();
		}
	}
	//若結果溢位，則返回 int 最大值，否則正常返回 size 值 
	return overflow ? Integer.MAX_VALUE : size;
}

其實原始碼中前兩行的註釋也說的非常清楚了。我們先採用樂觀的方式，認為在統計 size 的過程中，並沒有發生 put， remove 等會改變 Segment 結構的操作。 但是，如果發生了，就需要重試。如果重試2次都不成功(執行三次，第一次不能叫做重試)，就只能強制把所有 Segment 都加鎖之後，再統計了，以此來得到準確的結果。

ConcurrentHashMap 1.8 原始碼分析

需要說明的是，JDK 1.8 的 CHM（ConcurrentHashMap） 實現，完全重構了 1.7 。不再有 Segment 的概念，只是為了相容 1.7 才申明瞭一下，並沒有用到。因此，不再使用分段鎖，而是給陣列中的每一個頭節點（為了方便，以後都叫桶）都加鎖，鎖的粒度降低了。並且，用的是 Synchronized 鎖。

可能有的小夥伴就有疑惑了，不是都說同步鎖是重量級鎖嗎，這樣不是會影響併發效率嗎？

確實之前同步鎖是一個重量級鎖，但是在 JDK1.6 之後進行了各種優化之後，它已經不再那麼重了。引入了偏向鎖，輕量級鎖，以及鎖升級的概念，而且，據說在更細粒度的程式碼層面上，同步鎖已經可以媲美 Lock 鎖，甚至是趕超了。 除此之外，它還有很多優點，這裡不再展開了。感興趣的可以自行查閱同步鎖的鎖升級過程，以及它和 Lock 鎖的區別。

在 1.8 CHM 中，底層儲存結構和 1.8 的 HashMap 是一樣的，都是陣列+連結串列+紅黑樹。不同的就是，多了一些併發的處理。

文章開頭我們提到了，在 1.8 HashMap 中的執行緒安全問題，就是因為在多個執行緒同時操作同一個桶的頭結點時，會發生值的覆蓋情況。那麼，順著這個思路，我們看一下在 CHM 中它是怎麼避免這種情況發生的吧。

PS： 由於1.8的 CHM 和 HashMap 結構和基本屬性變數，還有初始化邏輯都差不多，只是多了一些併發情況需要用到的引數和內部類，因此，不再單獨拎出來介紹。在方法中用到的時候，再詳細解釋。

put()方法

因此，從 put 方法開始，我們看下，它在插入新元素的時候，是怎麼保證執行緒安全的吧。

public V put(K key, V value) {
	return putVal(key, value, false);
}

final V putVal(K key, V value, boolean onlyIfAbsent) {
	//可以看到，在併發情況下，key 和 value 都是不支援為空的。
	if (key == null || value == null) throw new NullPointerException();
	//這裡和1.8 HashMap 的hash 方法大同小異，只是多了一個操作，如下
	//( h ^ (h >>> 16)) & HASH_BITS;  HASH_BITS = 0x7fffffff;
	// 0x7fffffff ，二進位制為 0111 1111 1111 1111 1111 1111 1111 1111 。
	//所以，hash值除了做了高低位異或運算，還多了一步，保證最高位的 1 個 bit 位總是0。
	//這裡，我並沒有明白它的意圖，僅僅是保證計算出來的hash值不超過 Integer 最大值，且不為負數嗎。
	//同 HashMap 的hash 方法對比一下，會發現連原始碼註釋都是相同的，並沒有多說明其它的。
	//我個人認為意義不大，因為最後 hash 是為了和 capacity -1 做與運算，而 capacity 最大值為 1<<30，
	//即 0100 0000 0000 0000 0000 0000 0000 0000 ，減1為 0011 1111 1111 1111 1111 1111 1111 1111。
	//即使 hash 最高位為 1(無所謂0)，也不影響最後的結果，最高位也總會是0.
	int hash = spread(key.hashCode());
	//用來計算當前連結串列上的元素個數
	int binCount = 0;
	for (Node<K,V>[] tab = table;;) {
		Node<K,V> f; int n, i, fh;
		//如果表為空，則說明還未初始化。
		if (tab == null || (n = tab.length) == 0)
			//初始化表，只有一個執行緒可以初始化成功。
			tab = initTable();
		//若表已經初始化，則找到當前 key 所在的桶，並且判斷是否為空
		else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
			//若當前桶為空，則通過 CAS 原子操作，把新節點插入到此位置，
			//這保證了只有一個執行緒可以 CAS 成功，其它執行緒都會失敗。
			if (casTabAt(tab, i, null,
						 new Node<K,V>(hash, key, value, null)))
				break;                   // no lock when adding to empty bin
		}
		//若所在桶不為空，則判斷節點的 hash 值是否為 MOVED（值是-1）
		else if ((fh = f.hash) == MOVED)
			//若為-1，說明當前陣列正在進行擴容，則需要當前執行緒幫忙遷移資料
			tab = helpTransfer(tab, f);
		else {
			V oldVal = null;
			//這裡用加同步鎖的方式，來保證執行緒安全，給桶中第一個節點物件加鎖
			synchronized (f) {
				//recheck 一下，保證當前桶的第一個節點無變化，後邊很多這樣類似的操作，不再贅述
				if (tabAt(tab, i) == f) {
					//如果hash值大於等於0，說明是正常的連結串列結構
					if (fh >= 0) {
						binCount = 1;
						//從頭結點開始遍歷，每遍歷一次，binCount計數加1
						for (Node<K,V> e = f;; ++binCount) {
							K ek;
							//如果找到了和當前 key 相同的節點，則用新值替換舊值
							if (e.hash == hash &&
								((ek = e.key) == key ||
								 (ek != null && key.equals(ek)))) {
								oldVal = e.val;
								if (!onlyIfAbsent)
									e.val = value;
								break;
							}
							Node<K,V> pred = e;
							//若遍歷到了尾結點，則把新節點尾插進去
							if ((e = e.next) == null) {
								pred.next = new Node<K,V>(hash, key,
														  value, null);
								break;
							}
						}
					}
					//否則判斷是否是樹節點。這裡提一下，TreeBin只是頭結點對TreeNode的再封裝
					else if (f instanceof TreeBin) {
						Node<K,V> p;
						binCount = 2;
						if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
													   value)) != null) {
							oldVal = p.val;
							if (!onlyIfAbsent)
								p.val = value;
						}
					}
				}
			}
			//注意下，這個判斷是在同步鎖外部，因為 treeifyBin內部也有同步鎖，並不影響
			if (binCount != 0) {
				//如果節點個數大於等於 8，則轉化為紅黑樹
				if (binCount >= TREEIFY_THRESHOLD)
					treeifyBin(tab, i);
				//把舊節點值返回
				if (oldVal != null)
					return oldVal;
				break;
			}
		}
	}
	//給元素個數加 1，並有可能會觸發擴容，比較複雜，稍後細講
	addCount(1L, binCount);
	return null;
}

initTable()方法

先看下當陣列為空時，是怎麼初始化表的。

private final Node<K,V>[] initTable() {
	Node<K,V>[] tab; int sc;
	//迴圈判斷表是否為空，直到初始化成功為止。
	while ((tab = table) == null || tab.length == 0) {
		//sizeCtl 這個值有很多情況，預設值為0，
		//當為 -1 時，說明有其它執行緒正在對錶進行初始化操作
		//當表初始化成功後，又會把它設定為擴容閾值
		//當為一個小於 -1 的負數，用來表示當前有幾個執行緒正在幫助擴容(後邊細講)
		if ((sc = sizeCtl) < 0)
			//若 sc 小於0，其實在這裡就是-1，因為此時表是空的，不會發生擴容，sc只能為正數或者-1
			//因此，當前執行緒放棄 CPU 時間片，只是自旋。
			Thread.yield(); // lost initialization race; just spin
		//通過 CAS 把 sc 的值設定為-1，表明當前執行緒正在進行表的初始化，其它失敗的執行緒就會自旋
		else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
			try {
				//重新檢查一下表是否為空
				if ((tab = table) == null || tab.length == 0) {
					//如果sc大於0，則為sc，否則返回預設容量 16。
					//當呼叫有參構造建立 Map 時，sc的值是大於0的。
					int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
					@SuppressWarnings("unchecked")
					//建立陣列
					Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
					table = tab = nt;
					//n減去 1/4 n ，即為 0.75n ，表示擴容閾值
					sc = n - (n >>> 2);
				}
			} finally {
				//更新 sizeCtl 為擴容閾值
				sizeCtl = sc;
			}
			//若當前執行緒初始化表成功，則跳出迴圈。其它自旋的執行緒因為判斷陣列不為空，也會停止自旋
			break;
		}
	}
	return tab;
}

addCount()方法

若 put 方法元素插入成功之後，則會呼叫此方法，傳入引數為 addCount(1L, binCount)。這個方法的目的很簡單，就是把整個 table 的元素個數加 1 。但是，實現比較難。

我們先思考一下，如果讓我們自己去實現這樣的統計元素個數，怎麼實現？

類比 1.8 的 HashMap ，我們可以搞一個 size 變數來儲存個數統計。但是，這是在多執行緒環境下，需要考慮併發的問題。因此，可以把 size 設定為 volatile 的，保證可見性，然後通過 CAS 樂觀鎖來自增 1。

這樣雖然也可以實現。但是，設想一下現在有非常多的執行緒，都在同一時間操作這個 size 變數，將會造成特別嚴重的競爭。所以，基於此，這裡做了更好的優化。讓這些競爭的執行緒，分散到不同的物件裡邊，單獨操作它自己的資料(計數變數)，用這樣的方式儘量降低競爭。到最後需要統計 size 的時候，再把所有物件裡邊的計數相加就可以了。

上邊提到的 size ，在此用 baseCount 表示。分散到的物件用 CounterCell 表示，物件裡邊的計數變數用 value 表示。注意這裡的變數都是 volatile 修飾的。

當需要修改元素數量時，執行緒會先去 CAS 修改 baseCount 加1，若成功即返回。若失敗，則執行緒被分配到某個 CounterCell ，然後操作 value 加1。若成功，則返回。否則，給當前執行緒重新分配一個 CounterCell，再嘗試給 value 加1。（這裡簡略的說，實際更復雜）

CounterCell 會組成一個數組，也會涉及到擴容問題。所以，先畫一個示意圖幫助理解一下。

//執行緒被分配到的格子
@sun.misc.Contended static final class CounterCell {
	//此格子內記錄的 value 值
    volatile long value;
    CounterCell(long x) { value = x; }
}

//用來儲存執行緒和執行緒生成的隨機數的對應關係
static final int getProbe() {
	return UNSAFE.getInt(Thread.currentThread(), PROBE);
}

// x為1,check代表連結串列上的元素個數
private final void addCount(long x, int check) {
	CounterCell[] as; long b, s;
	//此處要進入if有兩種情況
	//1.陣列不為空，說明陣列已經被建立好了。
	//2.若陣列為空，說明陣列還未建立，很有可能競爭的執行緒非常少，因此就直接 CAS 操作 baseCount
	//若 CAS 成功，則方法跳轉到 (2)處，若失敗，則需要考慮給當前執行緒分配一個格子（指CounterCell物件）
	if ((as = counterCells) != null ||
		!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
		CounterCell a; long v; int m;
		//字面意思，是無競爭，這裡先標記為 true，表示還沒有產生執行緒競爭
		boolean uncontended = true;
		//這裡有三種情況，會進入 fullAddCount 方法
		//1.若陣列為空，進方法 (1)
		//2.ThreadLocalRandom.getProbe() 方法會給當前執行緒生成一個隨機數（可以簡單的認為也是一個hash值）
		//然後用隨機數與陣列長度取模，計算它所在的格子。若當前執行緒所分配到的格子為空，進方法 (1)。
		//3.若陣列不為空，且執行緒所在格子不為空，則嘗試 CAS 修改此格子對應的 value 值加1。
		//若修改成功，則跳轉到 (3)，若失敗，則把 uncontended 值設為 fasle，說明產生了競爭，然後進方法 (1)
		if (as == null || (m = as.length - 1) < 0 ||
			(a = as[ThreadLocalRandom.getProbe() & m]) == null ||
			!(uncontended =
			  U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
			//方法(1), 這個方法的目的是讓當前執行緒一定把 1 加成功。情況更多，更復雜，稍後講。
			fullAddCount(x, uncontended);
			return;
		}
		//(3)能走到這，說明陣列不為空，且修改 baseCount失敗，
		//且執行緒被分配到的格子不為空，且修改 value 成功。
		//但是這裡沒明白為什麼小於等於1，就直接返回了，這裡我懷疑之前的方法漏掉了binCount=0的情況。
		//而且此處若返回了，後邊怎麼判斷擴容？（存疑）
		if (check <= 1)
			return;
		//計算總共的元素個數
		s = sumCount();
	}
	//(2)這裡用於檢查是否需要擴容（下邊這部分很多邏輯不懂的話，等後邊講完擴容，再回來看就理解了）
	if (check >= 0) {
		Node<K,V>[] tab, nt; int n, sc;
		//若元素個數達到擴容閾值，且tab不為空，且tab陣列長度小於最大容量
		while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
			   (n = tab.length) < MAXIMUM_CAPACITY) {
			//這裡假設陣列長度n就為16，這個方法返回的是一個固定值，用於當做一個擴容的校驗標識
			//可以跳轉到最後，看詳細計算過程，0000 0000 0000 0000 1000 0000 0001 1011
			int rs = resizeStamp(n);
			//若sc小於0，說明正在擴容
			if (sc < 0) {
			    //sc的結構類似這樣，1000 0000 0001 1011 0000 0000 0000 0001
				//sc的高16位是資料校驗標識，低16位代表當前有幾個執行緒正在幫助擴容,RESIZE_STAMP_SHIFT=16
				//因此判斷校驗標識是否相等，不相等則退出迴圈
				//sc == rs + 1,sc == rs + MAX_RESIZERS 這兩個應該是用來判斷擴容是否已經完成，但是計算方法存疑
				//感興趣的可以看這個地址，應該是一個 bug ，
				// https://bugs.java.com/bugdatabase/view_bug.do?bug_id=JDK-8214427
				//nextTable=null 說明需要擴容的新陣列還未建立完成
				//transferIndex這個引數小於等於0，說明已經不需要其它執行緒幫助擴容了，
				//但是並不說明已經擴容完成，因為有可能還有執行緒正在遷移元素。稍後擴容細講就明白了。
				if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
					sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
					transferIndex <= 0)
					break;
				//到這裡說明當前執行緒可以幫助擴容，因此sc值加一，代表擴容的執行緒數加1
				if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
					transfer(tab, nt);
			}
			//當sc大於0，說明sc代表擴容閾值，因此第一次擴容之前肯定走這個分支，用於初始化新表 nextTable
			//rs<<16
			//1000 0000 0001 1011 0000 0000 0000 0000
			//+2
			//1000 0000 0001 1011 0000 0000 0000 0010
			//這個值，轉為十進位制就是 -2145714174，用於標識，這是擴容時，初始化新表的狀態，
			//擴容時，需要用到這個引數校驗是否所有執行緒都全部幫助擴容完成。
			else if (U.compareAndSwapInt(this, SIZECTL, sc,
										 (rs << RESIZE_STAMP_SHIFT) + 2))
				//擴容，第二個引數代表新表，傳入null，則說明是第一次初始化新表(nextTable)
				transfer(tab, null);
			s = sumCount();
		}
	}
}

//計算表中的元素總個數
final long sumCount() {
	CounterCell[] as = counterCells; CounterCell a;
	//baseCount，以這個值作為累加基準
	long sum = baseCount;
	if (as != null) {
		//遍歷 counterCells 陣列，得到每個物件中的value值
		for (int i = 0; i < as.length; ++i) {
			if ((a = as[i]) != null)
				//累加 value 值
				sum += a.value;
		}
	}
	//此時得到的就是元素總個數
	return sum;
}	

//擴容時的校驗標識
static final int resizeStamp(int n) {
	return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1));
}

//Integer.numberOfLeadingZeros方法的作用是返回 n 的最高位為1的前面的0的個數
//n=16，
0000 0000 0000 0000 0000 0000 0001 0000
//前面有27個0，即27
0000 0000 0000 0000 0000 0000 0001 1011
//RESIZE_STAMP_BITS為16，然後 1<<(16-1)，即 1<<15
0000 0000 0000 0000 1000 0000 0000 0000
//它們做或運算，得到 rs 的值
0000 0000 0000 0000 1000 0000 0001 1011

fullAddCount()方法

上邊的 addCount 方法還沒完，別忘了有可能元素個數加 1 的操作還未成功，就走到 fullAddCount 這個方法了。看方法名，就知道了，全力增加計數值，一定要成功（奧利給）。 這個方法和擴容遷移方法是最難的，保持耐心~

//傳過來的引數分別為 1 ， false
private final void fullAddCount(long x, boolean wasUncontended) {
	int h;
	//如果當前執行緒的隨機數為0，則強制初始化一個值
	if ((h = ThreadLocalRandom.getProbe()) == 0) {
		ThreadLocalRandom.localInit();      // force initialization
		h = ThreadLocalRandom.getProbe();
		//此時把 wasUncontended 設為true，認為無競爭
		wasUncontended = true;
	}
	//用來表示比 contend（競爭）更嚴重的碰撞，若為true，表示可能需要擴容，以減少碰撞衝突
	boolean collide = false;                // True if last slot nonempty
	//迴圈內，外層if判斷分三種情況，內層判斷又分為六種情況
	for (;;) {
		CounterCell[] as; CounterCell a; int n; long v;
		//1. 若counterCells陣列不為空。  建議先看下邊的2和3兩種情況，再回頭看這個。 
		if ((as = counterCells) != null && (n = as.length) > 0) {
			// (1) 若當前執行緒所在的格子（CounterCell物件）為空
			if ((a = as[(n - 1) & h]) == null) {
				if (cellsBusy == 0) {    
					//若無鎖，則樂觀的建立一個 CounterCell 物件。
					CounterCell r = new CounterCell(x); 
					//嘗試加鎖
					if (cellsBusy == 0 &&
						U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
						boolean created = false;
						//加鎖成功後，再 recheck 一下陣列是否不為空，且當前格子為空
						try {               
							CounterCell[] rs; int m, j;
							if ((rs = counterCells) != null &&
								(m = rs.length) > 0 &&
								rs[j = (m - 1) & h] == null) {
								//把新建立的物件賦值給當前格子
								rs[j] = r;
								created = true;
							}
						} finally {
							//手動釋放鎖
							cellsBusy = 0;
						}
						//若當前格子建立成功，且上邊的賦值成功，則說明加1成功，退出迴圈
						if (created)
							break;
						//否則，繼續下次迴圈
						continue;           // Slot is now non-empty
					}
				}
				//若cellsBusy=1，說明有其它執行緒搶鎖成功。或者若搶鎖的 CAS 操作失敗，都會走到這裡，
				//則當前執行緒需跳轉到(9)重新生成隨機數，進行下次迴圈判斷。
				collide = false;
			}
			/**
			*後邊這幾種情況，都是陣列和當前隨機到的格子都不為空的情況。
			*且注意每種情況，若執行成功，且不break，continue，則都會執行(9)，重新生成隨機數，進入下次迴圈判斷
			*/
			// (2) 到這，說明當前方法在被呼叫之前已經 CAS 失敗過一次，若不明白可回頭看下 addCount 方法，
			//為了減少競爭，則跳轉到⑨處重新生成隨機數，並把 wasUncontended 設定為true ，認為下一次不會產生競爭
			else if (!wasUncontended)       // CAS already known to fail
				wasUncontended = true;      // Continue after rehash
			// (3) 若 wasUncontended 為 true 無競爭，則嘗試一次 CAS。若成功，則結束迴圈，若失敗則判斷後邊的 (4)(5)(6)。
			else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))
				break;
			// (4) 結合 (6) 一起看，(4)(5)(6)都是 wasUncontended=true，且CAS修改value失敗的情況。
			//若陣列有變化，或者陣列長度大於等於當前CPU的核心數，則把 collide 改為 false
			//因為陣列若有變化，說明是由擴容引起的；長度超限，則說明已經無法擴容，只能認為無碰撞。
			//這裡很有意思，認真思考一下，當擴容超限後，則會達到一個平衡，即 (4)(5) 反覆執行，直到 (3) 中CAS成功，跳出迴圈。
			else if (counterCells != as || n >= NCPU)
				collide = false;            // At max size or stale
			// (5) 若陣列無變化，且陣列長度小於CPU核心數時，且 collide 為 false，就把它改為 true，說明下次迴圈可能需要擴容
			else if (!collide)
				collide = true;
			// (6) 若陣列無變化，且陣列長度小於CPU核心數時，且 collide 為 true，說明衝突比較嚴重，需要擴容了。
			else if (cellsBusy == 0 &&
					 U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
				try {
					//recheck
					if (counterCells == as) {// Expand table unless stale
						//建立一個容量為原來兩倍的陣列
						CounterCell[] rs = new CounterCell[n << 1];
						//轉移舊陣列的值
						for (int i = 0; i < n; ++i)
							rs[i] = as[i];
						//更新陣列
						counterCells = rs;
					}
				} finally {
					cellsBusy = 0;
				}
				//認為擴容後，下次不會產生衝突了，和(4)處邏輯照應
				collide = false;
				//當次擴容後，就不需要重新生成隨機數了
				continue;                   // Retry with expanded table
			}
			// (9)，重新生成一個隨機數，進行下一次迴圈判斷
			h = ThreadLocalRandom.advanceProbe(h);
		}
		//2.這裡的 cellsBusy 引數非常有意思，是一個volatile的 int值，用來表示自旋鎖的標誌，
		//可以類比 AQS 中的 state 引數，用來控制鎖之間的競爭，並且是獨佔模式。簡化版的AQS。
		//cellsBusy 若為0，說明無鎖，執行緒都可以搶鎖，若為1，表示已經有執行緒拿到了鎖，則其它執行緒不能搶鎖。
		else if (cellsBusy == 0 && counterCells == as &&
				 U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
			boolean init = false;
			try {    
				//這裡再重新檢測下 counterCells 陣列引用是否有變化
				if (counterCells == as) {
					//初始化一個長度為 2 的陣列
					CounterCell[] rs = new CounterCell[2];
					//根據當前執行緒的隨機數值，計算下標，只有兩個結果 0 或 1，並初始化物件
					rs[h & 1] = new CounterCell(x);
					//更新陣列引用
					counterCells = rs;
					//初始化成功的標誌
					init = true;
				}
			} finally {
				//別忘了，需要手動解鎖。
				cellsBusy = 0;
			}
			//若初始化成功，則說明當前加1的操作也已經完成了，則退出整個迴圈。
			if (init)
				break;
		}
		//3.到這，說明陣列為空，且 2 搶鎖失敗，則嘗試直接去修改 baseCount 的值，
		//若成功，也說明加1操作成功，則退出迴圈。
		else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x))
			break;                          // Fall back on using base
	}
}

不得不佩服 Doug Lea 大神，思維這麼縝密，如果是我的話，直接一個 CAS 完事。（手動攤手~）

transfer()方法

需要說明的一點是，雖然我們一直在說幫助擴容，其實更準確的說應該是幫助遷移元素。因為擴容的第一次初始化新表（擴容後的新表）這個動作，只能由一個執行緒完成。其他執行緒都是在幫助遷移元素到新陣列。

這裡還是先看下遷移的示意圖，幫助理解。

為了方便，上邊以原陣列長度 8 為例。在元素遷移的時候，所有執行緒都遵循從後向前推進的規則，即如圖A執行緒是第一個進來的執行緒，會從下標為7的位置，開始遷移資料。

而且當前執行緒遷移時會確定一個範圍，限定它此次遷移的資料範圍，如圖 A 執行緒只能遷移 bound=6到 i=7 這兩個資料。

此時，其它執行緒就不能遷移這部分資料了，只能繼續向前推進，尋找其它可以遷移的資料範圍，且每次推進的步長為固定值 stride（此處假設為2）。如圖中 B執行緒發現 A 執行緒正在遷移6,7的資料，因此只能向前尋找，然後遷移 bound=4 到 i=5 的這兩個資料。

當每個執行緒遷移完成它的範圍內資料時，都會繼續向前推進。那什麼時候是個頭呢？

這就需要維護一個全域性的變數 transferIndex，來表示所有執行緒總共推進到的元素下標位置。如圖，執行緒 A 第一次遷移成功後又向前推進，然後遷移2,3 的資料。此時，若沒有其他執行緒在幫助遷移，則 transferIndex 即為2。

剩餘部分等待下一個執行緒來遷移，或者有任何的 A 和B執行緒已經遷移完成，也可以推進到這裡幫助遷移。直到 transferIndex=0 。（會做一些其他校驗來判斷是否遷移全部完成，看程式碼）。

//這個類是一