前言
上篇文章介紹了 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 方法的總體流程是，
- 通過哈希算法計算出當前 key 的 hash 值
- 通過這個 hash 值找到它所對應的 Segment 數組的下標
- 再通過 hash 值計算出它在對應 Segment 的 HashEntry數組 的下標
- 找到合適的位置插入元素
//這是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 。（會做一些其他校驗來判斷是否遷移全部完成，看代碼）。
//這個類是一個標志，用來代表當前桶（數組中的某個下標位置）的元素已經全部遷移完成
static final class ForwardingNode<K,V> extends Node<K,V> {
final Node<K,V>[] nextTable;
ForwardingNode(Node<K,V>[] tab) {
//把當前桶的頭結點的 hash 值設置為 -1，表明已經遷移完成，
//這個節點中並不存儲有效的數據
super(MOVED, null, null, null);
this.nextTable = tab;
}
}
//遷移數據
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
//根據當前CPU核心數，確定每次推進的步長，最小值為16.（為了方便我們以2為例）
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
//從 addCount 方法，只會有一個線程跳轉到這里，初始化新數組
if (nextTab == null) { // initiating
try {
@SuppressWarnings("unchecked")
//新數組長度為原數組的兩倍
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
nextTab = nt;
} catch (Throwable ex) { // try to cope with OOME
sizeCtl = Integer.MAX_VALUE;
return;
}
//用 nextTable 指代新數組
nextTable = nextTab;
//這里就把推進的下標值初始化為原數組長度（以16為例）
transferIndex = n;
}
//新數組長度
int nextn = nextTab.length;
//創建一個標志類
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
//是否向前推進的標志
boolean advance = true;
//是否所有線程都全部遷移完成的標志
boolean finishing = false; // to ensure sweep before committing nextTab
//i 代表當前線程正在遷移的桶的下標，bound代表它本次可以遷移的范圍下限
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
//需要向前推進
while (advance) {
int nextIndex, nextBound;
// (1) 先看 (3) 。i每次自減 1，直到 bound。若超過bound范圍，或者finishing標志為true，則不用向前推進。
//若未全部完成遷移，且 i 並未走到 bound，則跳轉到 (7)，處理當前桶的元素遷移。
if (--i >= bound || finishing)
advance = false;
// (2) 每次執行，都會把 transferIndex 最新的值同步給 nextIndex
//若 transferIndex小於等於0，則說明原數組中的每個桶位置，都有線程在處理遷移了，
//於是，需要跳出while循環，並把 i設為 -1，以跳轉到④判斷在處理的線程是否已經全部完成。
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
// (3) 第一個線程會先走到這里，確定它的數據遷移范圍。(2)處會更新 nextIndex為 transferIndex 的最新值
//因此第一次 nextIndex=n=16，nextBound代表當次遷移的數據范圍下限，減去步長即可，
//所以，第一次時，nextIndex=16，nextBound=16-2=14。后續，每次都會間隔一個步長。
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
//bound代表當次數據遷移下限
bound = nextBound;
//第一次的i為15，因為長度16的數組，最后一個元素的下標為15
i = nextIndex - 1;
//表明不需要向前推進，只有當把當前范圍內的數據全部遷移完成后，才可以向前推進
advance = false;
}
}
// (4)
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
//若全部線程遷移完成
if (finishing) {
nextTable = null;
//更新table為新表
table = nextTab;
//擴容閾值改為原來數組長度的 3/2 ，即新長度的 3/4，也就是新數組長度的0.75倍
sizeCtl = (n << 1) - (n >>> 1);
return;
}
//到這，說明當前線程已經完成了自己的所有遷移（無論參與了幾次遷移），
//則把 sc 減1，表明參與擴容的線程數減少 1。
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
//在 addCount 方法最后，我們強調，遷移開始時，會設置 sc=(rs << RESIZE_STAMP_SHIFT) + 2
//每當有一個線程參與遷移，sc 就會加 1，每當有一個線程完成遷移，sc 就會減 1。
//因此，這里就是去校驗當前 sc 是否和初始值是否相等。相等，則說明全部線程遷移完成。
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
return;
//只有此處，才會把finishing 設置為true。
finishing = advance = true;
//這里非常有意思，會把 i 從 -1 修改為16，
//目的就是，讓 i 再從后向前掃描一遍數組，檢查是否所有的桶都已被遷移完成，參看 (6)
i = n; // recheck before commit
}
}
// (5) 若i的位置元素為空，則說明當前桶的元素已經被遷移完成，就把頭結點設置為fwd標志。
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
// (6) 若當前桶的頭結點是 ForwardingNode ，說明遷移完成，則向前推進
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
//(7) 處理當前桶的數據遷移。
else {
synchronized (f) { //給頭結點加鎖
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;
//若hash值大於等於0，則說明是普通鏈表節點
if (fh >= 0) {
int runBit = fh & n;
//這里是 1.7 的 CHM 的 rehash 方法和 1.8 HashMap的 resize 方法的結合體。
//會分成兩條鏈表，一條鏈表和原來的下標相同，另一條鏈表是原來的下標加數組長度的位置
//然后找到 lastRun 節點，從它到尾結點整體遷移。
//lastRun前邊的節點則單個遷移，但是需要注意的是，這里是頭插法。
//另外還有一點和1.7不同，1.7 lastRun前邊的節點是復制過去的，而這里是直接遷移的，沒有復制操作。
//所以，最后會有兩條鏈表，一條鏈表從 lastRun到尾結點是正序的，而lastRun之前的元素是倒序的，
//另外一條鏈表，從頭結點開始就是倒敘的。看下圖。
Node<K,V> lastRun = f;
for (Node<K,V> p = f.next; p != null; p = p.next) {
int b = p.hash & n;
if (b != runBit) {
runBit = b;
lastRun = p;
}
}
if (runBit == 0) {
ln = lastRun;
hn = null;
}
else {
hn = lastRun;
ln = null;
}
for (Node<K,V> p = f; p != lastRun; p = p.next) {
int ph = p.hash; K pk = p.key; V pv = p.val;
if ((ph & n) == 0)
ln = new Node<K,V>(ph, pk, pv, ln);
else
hn = new Node<K,V>(ph, pk, pv, hn);
}
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
setTabAt(tab, i, fwd);
advance = true;
}
//樹節點
else if (f instanceof TreeBin) {
TreeBin<K,V> t = (TreeBin<K,V>)f;
TreeNode<K,V> lo = null, loTail = null;
TreeNode<K,V> hi = null, hiTail = null;
int lc = 0, hc = 0;
for (Node<K,V> e = t.first; e != null; e = e.next) {
int h = e.hash;
TreeNode<K,V> p = new TreeNode<K,V>
(h, e.key, e.val, null, null);
if ((h & n) == 0) {
if ((p.prev = loTail) == null)
lo = p;
else
loTail.next = p;
loTail = p;
++lc;
}
else {
if ((p.prev = hiTail) == null)
hi = p;
else
hiTail.next = p;
hiTail = p;
++hc;
}
}
ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
(hc != 0) ? new TreeBin<K,V>(lo) : t;
hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
(lc != 0) ? new TreeBin<K,V>(hi) : t;
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
setTabAt(tab, i, fwd);
advance = true;
}
}
}
}
}
}
遷移后的新數組鏈表方向示意圖，以 runBit =0 為例。
helpTransfer()方法
最后再看 put 方法中的這個方法，就比較簡單了。
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
Node<K,V>[] nextTab; int sc;
//頭結點為 ForwardingNode ，並且新數組已經初始化
if (tab != null && (f instanceof ForwardingNode) &&
(nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {
int rs = resizeStamp(tab.length);
while (nextTab == nextTable && table == tab &&
(sc = sizeCtl) < 0) {
//若校驗標識失敗，或者已經擴容完成，或推進下標到頭，則退出
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || transferIndex <= 0)
break;
//當前線程需要幫助遷移，sc值加1
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
transfer(tab, nextTab);
break;
}
}
return nextTab;
}
return table;
}
JDK1.8 的 CHM 最主要的邏輯基本上都講完了，其它方法原理類同。1.8 的 ConcurrentHashMap 實現原理還是比較簡單的，但是代碼實現比較復雜。相對於 1.7 來說，鎖的粒度降低了，效率也提高了。
結語
看到這里的小伙伴，我想說你是最棒的！
求點贊：
上篇文章收藏比點贊還要多，我要哭了 o(╥﹏╥)o。
原創不易，小伙伴們如果覺得文章還不錯的話，動動手指，關注我，點個贊吧~