前言

上篇文章介紹了 HashMap 源碼后，在博客平臺廣受好評，讓本來己經(jīng)不打算更新這個系列的我，仿佛被打了一頓雞血。真的，被讀者認(rèn)可的感覺，就是這么奇妙。

原文：面試官再問你 HashMap 底層原理，就把這篇文章甩給他看

有讀者評論，希望我能出一版 ConcurrentHashMap 的解析。所以，今天的這篇文章，我準(zhǔn)備講述一下 ConcurrentHashMap 分別在JDK1.7和 JDK1.8 的源碼。文章較長，建議小伙伴們可以先收藏再看哦~

說一下為什么我要把源碼解析寫的這么詳細(xì)吧。一方面，可以記錄下當(dāng)時自己的思考過程，也方便后續(xù)自己復(fù)習(xí)翻閱；另一方面，記錄下來還能夠幫助看到文章的小伙伴加深對源碼的理解，簡直是一舉兩得的事情。

更正錯誤

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

put 方法，在新值替換舊值那里，應(yīng)該是只有一種情況的，e 不包括新值。圖中的方框也標(biāo)注出來了。因為，判斷 e=p.next==null ， 然后新的節(jié)點是賦值給 p.next 了，并沒有賦值給 e，此時 e 依舊是空的。所以 e!=null，代表當(dāng)前的 e 是已經(jīng)存在的舊值。

文章編寫過程，難免出現(xiàn)作者考慮不周的地方，如果有朋友發(fā)現(xiàn)有錯誤的地方，還請不吝賜教，指正出來。知錯能改，善莫大焉，對于技術(shù)，我們應(yīng)該懷有一顆嚴(yán)謹(jǐn)?shù)男膽B(tài)~

文章目錄

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

1）多線程下的 HashMap 有什么問題？

2）怎樣保證線程安全，為什么選用 ConcurrentHashMap？

3）ConcurrentHashMap 1.7 源碼解析

• 底層存儲結(jié)構(gòu)
• 常用變量
• 構(gòu)造函數(shù)
• put() 方法
• ensureSegment() 方法
• scanAndLockForPut() 方法
• rehash() 擴(kuò)容機(jī)制
• get() 獲取元素方法
• remove() 方法
• size() 方法是怎么統(tǒng)計元素個數(shù)的

4）ConcurrentHashMap 1.8 源碼解析

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

多線程下 HashMap 有什么問題？

在上一篇文章中，已經(jīng)講解了 HashMap 1.7 死循環(huán)的成因，也正因為如此，我們才說 HashMap 在多線程下是不安全的。但是，在JDK1.8 的 HashMap 改為采用尾插法，已經(jīng)不存在死循環(huán)的問題了，為什么也會線程不安全呢？

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

假如現(xiàn)在有兩個線程都執(zhí)行到了上圖中的劃線處。當(dāng)線程一判斷為空之后，CPU 時間片到了，被掛起。線程二也執(zhí)行到此處判斷為空，繼續(xù)執(zhí)行下一句，創(chuàng)建了一個新節(jié)點，插入到此下標(biāo)位置。然后，線程一解掛，同樣認(rèn)為此下標(biāo)的元素為空，因此也創(chuàng)建了一個新節(jié)點放在此下標(biāo)處，因此造成了元素的覆蓋。

所以，可以看到不管是 JDK1.7 還是 1.8 的 HashMap 都存在線程安全的問題。那么，在多線程環(huán)境下，應(yīng)該怎樣去保證線程安全呢？

怎樣保證線程安全，為什么選用 ConcurrentHashMap？

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

可以看到，不管是往 map 里邊添加元素還是獲取元素，都會用 synchronized 關(guān)鍵字加鎖。當(dāng)有多個元素之前存在資源競爭時，只能有一個線程可以獲取到鎖，操作資源。更不能忍的是，一個簡單的讀取操作，互相之間又不影響，為什么也不能同時進(jìn)行呢？

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

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

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

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

ConcurrentHashMap 1.7 源碼解析

我們看下在 JDK1.7中 ConcurrentHashMap 是怎么實現(xiàn)的。墻裂建議，在本文之前了解一下多線程的基本知識，如JMM內(nèi)存模型，volatile關(guān)鍵字作用，CAS和自旋，ReentranLock重入鎖。

底層存儲結(jié)構(gòu)

在 JDK1.7中，本質(zhì)上還是采用鏈表+數(shù)組的形式存儲鍵值對的。但是，為了提高并發(fā)，把原來的整個 table 劃分為 n 個 Segment 。所以，從整體來看，它是一個由 Segment 組成的數(shù)組。然后，每個 Segment 里邊是由 HashEntry 組成的數(shù)組，每個 HashEntry之間又可以形成鏈表。我們可以把每個 Segment 看成是一個小的 HashMap，其內(nèi)部結(jié)構(gòu)和 HashMap 是一模一樣的。

當(dāng)對某個 Segment 加鎖時，如圖中 Segment2，并不會影響到其他 Segment 的讀寫。每個 Segment 內(nèi)部自己操作自己的數(shù)據(jù)。這樣一來，我們要做的就是盡可能的讓元素均勻的分布在不同的 Segment中。最理想的狀態(tài)是，所有執(zhí)行的線程操作的元素都是不同的 Segment，這樣就可以降低鎖的競爭。

廢話了這么多，還是來看底層源碼吧，因為所有的思想都在代碼里體現(xiàn)。借用 Linus的一句話，“No BB . Show me the code ” （改編版，哈哈）

常用變量

先看下 1.7 中常用的變量和內(nèi)部類都有哪些，這有助于我們了解 ConcurrentHashMap 的整體結(jié)構(gòu)。

//默認(rèn)初始化容量，這個和 HashMap中的容量是一個概念，表示的是整個 Map的容量
static final int DEFAULT_INITIAL_CAPACITY = 16;

//默認(rèn)加載因子
static final float DEFAULT_LOAD_FACTOR = 0.75f;

//默認(rèn)的并發(fā)級別，這個參數(shù)決定了 Segment 數(shù)組的長度
static final int DEFAULT_CONCURRENCY_LEVEL = 16;

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

//每個Segment中table數(shù)組的最小長度為2，且必須是2的n次冪。
//由于每個Segment是懶加載的，用的時候才會初始化，因此為了避免使用時立即調(diào)整大小，設(shè)定了最小容量2
static final int MIN_SEGMENT_TABLE_CAPACITY = 2;

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

//在size方法和containsValue方法，會優(yōu)先采用樂觀的方式不加鎖，直到重試次數(shù)達(dá)到2，才會對所有Segment加鎖
//這個值的設(shè)定，是為了避免無限次的重試。后邊size方法會詳講怎么實現(xiàn)樂觀機(jī)制的。
static final int RETRIES_BEFORE_LOCK = 2;

//segment掩碼值，用于根據(jù)元素的hash值定位所在的 Segment 下標(biāo)。后邊會細(xì)講
final int segmentMask;

//和 segmentMask 配合使用來定位 Segment 的數(shù)組下標(biāo)，后邊講。
final int segmentShift;

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

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

 //用于表示每個Segment中的 table，是一個用HashEntry組成的數(shù)組。
 transient volatile HashEntry<K,V>[] table;

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

 //每次 table 結(jié)構(gòu)修改時，如put，remove等，此變量都會自增
 transient int modCount;

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

 //加載因子
 final float loadFactor;

 //Segment構(gòu)造函數(shù)
 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，用于存儲鍵值對的具體數(shù)據(jù)和維護(hù)單向鏈表的關(guān)系
static final class HashEntry<K,V> {
 //每個key通過哈希運算后的結(jié)果，用的是 Wang/Jenkins hash 的變種算法，此處不細(xì)講，感興趣的可自行查閱相關(guān)資料
 final int hash;
 final K key;
 //value和next都用 volatile 修飾，用于保證內(nèi)存可見性和禁止指令重排序
 volatile V value;
 //指向下一個節(jié)點
 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;
 }
}

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

ConcurrentHashMap 有五種構(gòu)造函數(shù)，但是最終都會調(diào)用同一個構(gòu)造函數(shù)，所以只需要搞明白這一個核心的構(gòu)造函數(shù)就可以了。

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

public ConcurrentHashMap(int initialCapacity,
                             float loadFactor, int concurrencyLevel) {
 //檢驗參數(shù)是否合法。值得說的是，并發(fā)級別一定要大于0，否則就沒辦法實現(xiàn)分段鎖了。
 if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
  throw new IllegalArgumentException();
 //并發(fā)級別不能超過最大值
 if (concurrencyLevel > MAX_SEGMENTS)
  concurrencyLevel = MAX_SEGMENTS;
 // Find power-of-two sizes best matching arguments
 //偏移量，是為了對hash值做位移操作，計算元素所在的Segment下標(biāo)，put方法詳講
 int sshift = 0;
 //用于設(shè)定最終Segment數(shù)組的長度，必須是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數(shù)組的長度
 int cap = MIN_SEGMENT_TABLE_CAPACITY;
 //(3)
 while (cap < c)
  cap <<= 1;
 // create segments and segments[0]
 //這里用 loadFactor做為加載因子，cap乘以加載因子作為擴(kuò)容閾值，創(chuàng)建長度為cap的HashEntry數(shù)組，
 //三個參數(shù)，創(chuàng)建一個Segment對象，保存到S0對象中。后邊在 ensureSegment 方法會用到S0作為原型對象去創(chuàng)建對應(yīng)的Segment。
 Segment<K,V> s0 =
  new Segment<K,V>(loadFactor, (int)(cap * loadFactor),
       (HashEntry<K,V>[])new HashEntry[cap]);
 //創(chuàng)建出長度為 ssize 的一個 Segment數(shù)組
 Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];
 //把S0存到Segment數(shù)組中去。在這里，我們就可以發(fā)現(xiàn)，此時只是創(chuàng)建了一個Segment數(shù)組，
 //但是并沒有把數(shù)組中的每個Segment對象創(chuàng)建出來，僅僅創(chuàng)建了一個Segment用來作為原型對象。
 UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
 this.segments = ss;
}    
```須是2的n次冪
 int ssize = 1;

上邊的注釋中留了 (1)(2)(3) 三個地方還沒有細(xì)說。我們現(xiàn)在假設(shè)一組數(shù)據(jù)，把涉及到的幾個變量計算出來，就能明白這些參數(shù)的含義了。

```java
//假設(shè)調(diào)用了默認(rèn)構(gòu)造，都用的是默認(rèn)參數(shù)，即 initialCapacity 和 concurrencyLevel 都是16
//(1)  sshift 和 ssize 值的計算過程為，每次循環(huán)，都會把 sshift 自增1，并且 ssize 左移一位，即乘以2，
//直到 ssize 的值大于等于 concurrencyLevel 的值 16。
sshfit=0,1,2,3,4
ssize=1,2,4,8,16
//可以看到，初始他們的值分別是0和1，最終結(jié)果是4和16
//sshfit是為了輔助計算segmentShift值，ssize是為了確定Segment數(shù)組長度。
//(2)  此時,計算c的值，
c = 16/16 = 1;
//判斷 c * 16 < 16 是否為真，真的話 c 自增1，此處為false，因此 c的值為1不變。
//(3)  此時，由于c為1， cap為2 ，因此判斷 cap < c 為false，最終cap為2。
//總結(jié)一下，以上三個步驟，最終都是為了確定以下幾個關(guān)鍵參數(shù)的值，
//確定 segmentShift ，這個用于后邊計算hash值的偏移量，此處即為 32-4=28，
//確定 ssize，必須是一個大于等于 concurrencyLevel 的一個2的n次冪值
//確定 cap，必須是一個大于等于2的一個2的n次冪值
//感興趣的小伙伴，還可以用另外幾組參數(shù)來計算上邊的參數(shù)值，可以加深理解參數(shù)的含義。
//例如initialCapacity和concurrencyLevel分別傳入10和5，或者傳入33和16

put()方法

put 方法的總體流程是，

1. 通過哈希算法計算出當(dāng)前 key 的 hash 值
2. 通過這個 hash 值找到它所對應(yīng)的 Segment 數(shù)組的下標(biāo)
3. 再通過 hash 值計算出它在對應(yīng) Segment 的 HashEntry數(shù)組 的下標(biāo)
4. 找到合適的位置插入元素

//這是Map的put方法
public V put(K key, V value) {
 Segment<K,V> s;
 //不支持value為空
 if (value == null)
  throw new NullPointerException();
 //通過 Wang/Jenkins 算法的一個變種算法，計算出當(dāng)前key對應(yīng)的hash值
 int hash = hash(key);
 //上邊我們計算出的 segmentShift為28，因此hash值右移28位，說明此時用的是hash的高4位，
 //然后把它和掩碼15進(jìn)行與運算，得到的值一定是一個 0000 ~ 1111 范圍內(nèi)的值，即 0~15 。
 int j = (hash >>> segmentShift) & segmentMask;
 //這里是用Unsafe類的原子操作找到Segment數(shù)組中j下標(biāo)的 Segment 對象
 if ((s = (Segment<K,V>)UNSAFE.getObject          // nonvolatile; recheck
   (segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegment
  //初始化j下標(biāo)的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 的實際值?？梢灶惐龋珹QS 中的 CAS 操作。Unsafe中的操作，都需要一個偏移量，看下圖，

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

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

這里說明一下計算 Segment 數(shù)組下標(biāo)和計算 HashEntry 數(shù)組下標(biāo)的不同點：

//下邊的hash值是通過哈希運算后的hash值，不是hashCode
//計算 Segment 下標(biāo)
 (hash >>> segmentShift) & segmentMask 
 //計算 HashEntry 數(shù)組下標(biāo)
 (tab.length - 1) & hash

思考一下，為什么它們的算法不一樣呢？計算 Segment 數(shù)組下標(biāo)是用的 hash值高幾位（這里以高 4 位為例）和掩碼做與運算，而計算 HashEntry 數(shù)組下標(biāo)是直接用的 hash 值和數(shù)組長度減1做與運算。

我的理解是，這是為了盡量避免當(dāng)前 hash 值計算出來的 Segment 數(shù)組下標(biāo)和計算出來的 HashEntry 數(shù)組下標(biāo)趨于相同。簡單說，就是為了避免分配到同一個 Segment 中的元素扎堆現(xiàn)象，即避免它們都被分配到同一條鏈表上，導(dǎo)致鏈表過長。同時，也是為了減少并發(fā)。下面做一個運算，幫助理解一下（假設(shè)不用高 4 位運算，而是正常情況都用低位做運算）。

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

可能有的小伙伴就會打岔了，那如果兩個 hash 值，低位和高位都相同，怎么辦呢。如果是這樣，我只能說，這個 hash 算法也太爛了吧。（這里的 hash 算法也會盡量避免這種情況，當(dāng)然只是減少幾率，并不能杜絕）

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

put 方法比較簡單，只要能看懂 HashMap 中的 put 方法，這里也沒問題。主要是它調(diào)用的子方法比較復(fù)雜，下邊一個一個講解。

ensureSegment()方法

回到 Map的 put 方法，判斷 j 下標(biāo)的 Segment為空后，則需要調(diào)用此方法，初始化一個 Segment 對象，以確保拿到的對象一定是不為空的，否則無法執(zhí)行s.put了。

//k為 (hash >>> segmentShift) & segmentMask 算法計算出來的值
private Segment<K,V> ensureSegment(int k) {
 final Segment<K,V>[] ss = this.segments;
 //u代表 k 的偏移量，用于通過 UNSAFE 獲取主內(nèi)存最新的實際 K 值
 long u = (k << SSHIFT) + SBASE; // raw offset
 Segment<K,V> seg;
 //從內(nèi)存中取到最新的下標(biāo)位置的 Segment 對象，判斷是否為空，(1)
 if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {
  //之前構(gòu)造函數(shù)說了，s0是作為一個原型對象，用于創(chuàng)建新的 Segment 對象
  Segment<K,V> proto = ss[0]; // use segment 0 as prototype
  //容量
  int cap = proto.table.length;
  //加載因子
  float lf = proto.loadFactor;
  //擴(kuò)容閾值
  int threshold = (int)(cap * lf);
  //把 Segment 對應(yīng)的 HashEntry 數(shù)組先創(chuàng)建出來
  HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];
  //再次檢查 K 下標(biāo)位置的 Segment 是否為空， (2)
  if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
   == null) { // recheck
   //此處把 Segment 對象創(chuàng)建出來，并賦值給 s，
   Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);
   //循環(huán)檢查 K 下標(biāo)位置的 Segment 是否為空， (3)
   //若不為空，則說明有其它線程搶先創(chuàng)建成功，并且已經(jīng)成功同步到主內(nèi)存中了，
   //則把它取出來，并返回
   while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
       == null) {
    //CAS，若當(dāng)前下標(biāo)的Segment對象為空，就把它替換為最新創(chuàng)建出來的 s 對象。
    //若成功，就跳出循環(huán)，否則，就一直自旋直到成功，或者 seg 不為空（其他線程成功導(dǎo)致）。
    if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
     break;
   }
  }
 }
 return seg;
}

可以發(fā)現(xiàn)，我標(biāo)注了上邊 (1)(2)(3) 個地方，每次都判斷最新的Segment是否為空?？赡苡械男』锇榫蜁曰?，為什么做這么多次判斷，我直接去自旋不就好了，反正最后都要自旋的。

我的理解是，在多線程環(huán)境下，因為不確定是什么時候會有其它線程 CAS 成功，有可能發(fā)生在以上的任意時刻。所以，只要發(fā)現(xiàn)一旦內(nèi)存中的對象已經(jīng)存在了，則說明已經(jīng)有其它線程把Segment對象創(chuàng)建好，并CAS成功同步到主內(nèi)存了。此時，就可以直接返回，而不需要往下執(zhí)行了。這樣做，是為了代碼執(zhí)行效率考慮。

scanAndLockForPut()方法

put 方法第一步搶鎖失敗之后，就會執(zhí)行此方法，

private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
 //根據(jù)hash值定位到它對應(yīng)的HashEntry數(shù)組的下標(biāo)位置，并找到鏈表的第一個節(jié)點
 //注意，這個操作會從主內(nèi)存中獲取到最新的狀態(tài)，以確保獲取到的first是最新值
 HashEntry<K,V> first = entryForHash(this, hash);
 HashEntry<K,V> e = first;
 HashEntry<K,V> node = null;
 //重試次數(shù)，初始化為 -1
 int retries = -1; // negative while locating node
 //若搶鎖失敗，就一直循環(huán)，直到成功獲取到鎖。有三種情況
 while (!tryLock()) {
  HashEntry<K,V> f; // to recheck first below
  //1.若 retries 小于0，
  if (retries < 0) {
   if (e == null) {
    //若 e 節(jié)點和 node 都為空，則創(chuàng)建一個 node 節(jié)點。這里只是預(yù)測性的創(chuàng)建一個node節(jié)點
    if (node == null) // speculatively create node
     node = new HashEntry<K,V>(hash, key, value, null);
    retries = 0;
   }
   //如當(dāng)前遍歷到的 e 節(jié)點不為空，則判斷它的key是否等于傳進(jìn)來的key，若是則把 retries 設(shè)為0
   else if (key.equals(e.key))
    retries = 0;
   //否則，繼續(xù)向后遍歷節(jié)點
   else
    e = e.next;
  }
  //2.若是重試次數(shù)超過了最大嘗試次數(shù)，則調(diào)用lock方法加鎖。表明不再重試，我下定決心了一定要獲取到鎖。
  //要么當(dāng)前線程可以獲取到鎖，要么獲取不到就去排隊等待獲取鎖。獲取成功后，再 break。
  else if (++retries > MAX_SCAN_RETRIES) {
   lock();
   break;
  }
  //3.若 retries 的值為偶數(shù)，并且從內(nèi)存中再次獲取到最新的頭節(jié)點，判斷若不等于first
  //則說明有其他線程修改了當(dāng)前下標(biāo)位置的頭結(jié)點，于是需要更新頭結(jié)點信息。
  else if ((retries & 1) == 0 &&
     (f = entryForHash(this, hash)) != first) {
   //更新頭結(jié)點信息，并把重試次數(shù)重置為 -1，繼續(xù)下一次循環(huán)，從最新的頭結(jié)點遍歷當(dāng)前鏈表。
   e = first = f; // re-traverse if entry changed
   retries = -1;
  }
 }
 return node;
}

這個方法邏輯比較復(fù)雜，會一直循環(huán)嘗試獲取鎖，若獲取成功，則返回。否則的話，每次循環(huán)時，都會同時遍歷當(dāng)前鏈表。若遍歷完了一次，還沒找到和key相等的節(jié)點，就會預(yù)先創(chuàng)建一個節(jié)點。注意，這里只是預(yù)測性的創(chuàng)建一個新節(jié)點，也有可能在這之前，就已經(jīng)獲取鎖成功了。

同時，當(dāng)重試次每偶數(shù)次時，就會檢查一次當(dāng)前最新的頭結(jié)點是否被改變。因為若有變化的話，還需要從最新的頭結(jié)點開始遍歷鏈表。

還有一種情況，就是循環(huán)次數(shù)達(dá)到了最大限制，則停止循環(huán)，用阻塞的方式去獲取鎖。這時，也就停止了遍歷鏈表的動作，當(dāng)前線程也不會再做其他預(yù)熱(warm up)的事情。

關(guān)于為什么預(yù)測性的創(chuàng)建新節(jié)點，源碼中原話是這樣的：

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

解釋一下就是，因為遍歷速度無所謂，所以，我們可以預(yù)先(warm up)做一些相關(guān)聯(lián)代碼的準(zhǔn)備工作。這里相關(guān)聯(lián)代碼，指的就是循環(huán)中，在獲取鎖成功或者調(diào)用 lock 方法之前做的這些事情，當(dāng)然也包括創(chuàng)建新節(jié)點。

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

scanAndLockForPut 這個方法可以確保返回時，當(dāng)前線程一定是獲取到鎖的狀態(tài)。

rehash()方法

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

//node為創(chuàng)建的新節(jié)點
private void rehash(HashEntry<K,V> node) {
 //當(dāng)前Segment中的舊表
 HashEntry<K,V>[] oldTable = table;
 //舊的容量
 int oldCapacity = oldTable.length;
 //新容量為舊容量的2倍
 int newCapacity = oldCapacity << 1;
 //更新新的閾值
 threshold = (int)(newCapacity * loadFactor);
 //用新的容量創(chuàng)建一個新的 HashEntry 數(shù)組
 HashEntry<K,V>[] newTable =
  (HashEntry<K,V>[]) new HashEntry[newCapacity];
 //當(dāng)前的掩碼，用于計算節(jié)點在新數(shù)組中的下標(biāo)
 int sizeMask = newCapacity - 1;
 //遍歷舊表
 for (int i = 0; i < oldCapacity ; i++) {
  HashEntry<K,V> e = oldTable[i];
  //如果e不為空，說明當(dāng)前鏈表不為空
  if (e != null) {
   HashEntry<K,V> next = e.next;
   //計算hash值再新數(shù)組中的下標(biāo)位置
   int idx = e.hash & sizeMask;
   //如果e不為空，且它的下一個節(jié)點為空，則說明這條鏈表只有一個節(jié)點，
   //直接把這個節(jié)點放到新數(shù)組的對應(yīng)下標(biāo)位置即可
   if (next == null)   //  Single node on list
    newTable[idx] = e;
   //否則，處理當(dāng)前鏈表的節(jié)點遷移操作
   else { // Reuse consecutive sequence at same slot
    //記錄上一次遍歷到的節(jié)點
    HashEntry<K,V> lastRun = e;
    //對應(yīng)上一次遍歷到的節(jié)點在新數(shù)組中的新下標(biāo)
    int lastIdx = idx;
    for (HashEntry<K,V> last = next;
      last != null;
      last = last.next) {
     //計算當(dāng)前遍歷到的節(jié)點的新下標(biāo)
     int k = last.hash & sizeMask;
     //若 k 不等于 lastIdx，則說明此次遍歷到的節(jié)點和上次遍歷到的節(jié)點不在同一個下標(biāo)位置
     //需要把 lastRun 和 lastIdx 更新為當(dāng)前遍歷到的節(jié)點和下標(biāo)值。
     //若相同，則不處理，繼續(xù)下一次 for 循環(huán)。
     if (k != lastIdx) {
      lastIdx = k;
      lastRun = last;
     }
    }
    //把和 lastRun 節(jié)點的下標(biāo)位置相同的鏈表最末尾的幾個連續(xù)的節(jié)點放到新數(shù)組的對應(yīng)下標(biāo)位置
    newTable[lastIdx] = lastRun;
    //再把剩余的節(jié)點，復(fù)制到新數(shù)組
    //從舊數(shù)組的頭結(jié)點開始遍歷，直到 lastRun 節(jié)點，因為 lastRun節(jié)點后邊的節(jié)點都已經(jīng)遷移完成了。
    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];
     //用的是復(fù)制節(jié)點信息的方式，并不是把原來的節(jié)點直接遷移，區(qū)別于lastRun處理方式
     newTable[k] = new HashEntry<K,V>(h, p.key, v, n);
    }
   }
  }
 }
 //所有節(jié)點都遷移完成之后，再處理傳進(jìn)來的新的node節(jié)點，把它頭插到對應(yīng)的下標(biāo)位置
 int nodeIndex = node.hash & sizeMask; // add the new node
 //頭插node節(jié)點
 node.setNext(newTable[nodeIndex]);
 newTable[nodeIndex] = node;
 //更新當(dāng)前Segment的table信息
 table = newTable;
}

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

從頭結(jié)點開始向后遍歷，找到當(dāng)前鏈表的最后幾個下標(biāo)相同的連續(xù)的節(jié)點。如上圖，雖然開頭出現(xiàn)了有兩個節(jié)點的下標(biāo)都是 k2, 但是中間出現(xiàn)一個不同的下標(biāo) k1，打斷了下標(biāo)連續(xù)相同，因此從下一個k2，又重新開始算。好在后邊三個連續(xù)的節(jié)點下標(biāo)都是相同的，因此倒數(shù)第三個節(jié)點被標(biāo)記為 lastRun，且變量無變化。

從lastRun節(jié)點到尾結(jié)點的這部分就可以整體遷移到新數(shù)組的對應(yīng)下標(biāo)位置了，因為它們的下標(biāo)都是相同的，可以這樣統(tǒng)一處理。

另外從頭結(jié)點到 lastRun 之前的節(jié)點，無法統(tǒng)一處理，只能一個一個去復(fù)制了。且注意，這里不是直接遷移，而是復(fù)制節(jié)點到新的數(shù)組，舊的節(jié)點會在不久的將來，因為沒有引用指向，被 JVM 垃圾回收處理掉。

（不知道為啥這個方法名起為 rehash，其實擴(kuò)容時 hash 值并沒有重新計算，變化的只是它們所在的下標(biāo)而已。我猜測，可能是，借用了 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 ，然后從主內(nèi)存中取出最新的節(jié)點
 long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
 if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
  (tab = s.table) != null) {
  //若Segment不為空，且鏈表也不為空，則遍歷查找節(jié)點
  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，否則執(zhí)行 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) {
 //嘗試加鎖，若失敗，則執(zhí)行 scanAndLock ，此方法和 scanAndLockForPut 方法類似
 if (!tryLock())
  scanAndLock(key, hash);
 V oldValue = null;
 try {
  HashEntry<K,V>[] tab = table;
  int index = (tab.length - 1) & hash;
  //從主內(nèi)存中獲取對應(yīng) table 的最新的頭結(jié)點
  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 方法需要重點說明一下。愛思考的小伙伴可能就會想到，并發(fā)情況下，有可能在統(tǒng)計期間，數(shù)組元素個數(shù)不停的變化，而且，整個表還被分成了 N個 Segment，怎樣統(tǒng)計才能保證結(jié)果的準(zhǔn)確性呢？我們一起來看下吧。

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

其實源碼中前兩行的注釋也說的非常清楚了。我們先采用樂觀的方式，認(rèn)為在統(tǒng)計 size 的過程中，并沒有發(fā)生 put， remove 等會改變 Segment 結(jié)構(gòu)的操作。但是，如果發(fā)生了，就需要重試。如果重試2次都不成功(執(zhí)行三次，第一次不能叫做重試)，就只能強(qiáng)制把所有 Segment 都加鎖之后，再統(tǒng)計了，以此來得到準(zhǔn)確的結(jié)果。

ConcurrentHashMap 1.8 源碼分析

需要說明的是，JDK 1.8 的 CHM（ConcurrentHashMap） 實現(xiàn)，完全重構(gòu)了 1.7 。不再有 Segment 的概念，只是為了兼容 1.7 才申明了一下，并沒有用到。因此，不再使用分段鎖，而是給數(shù)組中的每一個頭節(jié)點（為了方便，以后都叫桶）都加鎖，鎖的粒度降低了。并且，用的是 Synchronized 鎖。

可能有的小伙伴就有疑惑了，不是都說同步鎖是重量級鎖嗎，這樣不是會影響并發(fā)效率嗎？

確實之前同步鎖是一個重量級鎖，但是在 JDK1.6 之后進(jìn)行了各種優(yōu)化之后，它已經(jīng)不再那么重了。引入了偏向鎖，輕量級鎖，以及鎖升級的概念，而且，據(jù)說在更細(xì)粒度的代碼層面上，同步鎖已經(jīng)可以媲美 Lock 鎖，甚至是趕超了。除此之外，它還有很多優(yōu)點，這里不再展開了。感興趣的可以自行查閱同步鎖的鎖升級過程，以及它和 Lock 鎖的區(qū)別。

在 1.8 CHM 中，底層存儲結(jié)構(gòu)和 1.8 的 HashMap 是一樣的，都是數(shù)組+鏈表+紅黑樹。不同的就是，多了一些并發(fā)的處理。

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

PS：由于1.8的 CHM 和 HashMap 結(jié)構(gòu)和基本屬性變量，還有初始化邏輯都差不多，只是多了一些并發(fā)情況需要用到的參數(shù)和內(nèi)部類，因此，不再單獨拎出來介紹。在方法中用到的時候，再詳細(xì)解釋。

put()方法

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

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

final V putVal(K key, V value, boolean onlyIfAbsent) {
 //可以看到，在并發(fā)情況下，key 和 value 都是不支持為空的。
 if (key == null || value == null) throw new NullPointerException();
 //這里和1.8 HashMap 的hash 方法大同小異，只是多了一個操作，如下
 //( h ^ (h >>> 16)) & HASH_BITS;  HASH_BITS = 0x7fffffff;
 // 0x7fffffff ，二進(jìn)制為 0111 1111 1111 1111 1111 1111 1111 1111 。
 //所以，hash值除了做了高低位異或運算，還多了一步，保證最高位的 1 個 bit 位總是0。
 //這里，我并沒有明白它的意圖，僅僅是保證計算出來的hash值不超過 Integer 最大值，且不為負(fù)數(shù)嗎。
 //同 HashMap 的hash 方法對比一下，會發(fā)現(xiàn)連源碼注釋都是相同的，并沒有多說明其它的。
 //我個人認(rèn)為意義不大，因為最后 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)，也不影響最后的結(jié)果，最高位也總會是0.
 int hash = spread(key.hashCode());
 //用來計算當(dāng)前鏈表上的元素個數(shù)
 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();
  //若表已經(jīng)初始化，則找到當(dāng)前 key 所在的桶，并且判斷是否為空
  else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
   //若當(dāng)前桶為空，則通過 CAS 原子操作，把新節(jié)點插入到此位置，
   //這保證了只有一個線程可以 CAS 成功，其它線程都會失敗。
   if (casTabAt(tab, i, null,
       new Node<K,V>(hash, key, value, null)))
    break;                   // no lock when adding to empty bin
  }
  //若所在桶不為空，則判斷節(jié)點的 hash 值是否為 MOVED（值是-1）
  else if ((fh = f.hash) == MOVED)
   //若為-1，說明當(dāng)前數(shù)組正在進(jìn)行擴(kuò)容，則需要當(dāng)前線程幫忙遷移數(shù)據(jù)
   tab = helpTransfer(tab, f);
  else {
   V oldVal = null;
   //這里用加同步鎖的方式，來保證線程安全，給桶中第一個節(jié)點對象加鎖
   synchronized (f) {
    //recheck 一下，保證當(dāng)前桶的第一個節(jié)點無變化，后邊很多這樣類似的操作，不再贅述
    if (tabAt(tab, i) == f) {
     //如果hash值大于等于0，說明是正常的鏈表結(jié)構(gòu)
     if (fh >= 0) {
      binCount = 1;
      //從頭結(jié)點開始遍歷，每遍歷一次，binCount計數(shù)加1
      for (Node<K,V> e = f;; ++binCount) {
       K ek;
       //如果找到了和當(dāng)前 key 相同的節(jié)點，則用新值替換舊值
       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;
       //若遍歷到了尾結(jié)點，則把新節(jié)點尾插進(jìn)去
       if ((e = e.next) == null) {
        pred.next = new Node<K,V>(hash, key,
                value, null);
        break;
       }
      }
     }
     //否則判斷是否是樹節(jié)點。這里提一下，TreeBin只是頭結(jié)點對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內(nèi)部也有同步鎖，并不影響
   if (binCount != 0) {
    //如果節(jié)點個數(shù)大于等于 8，則轉(zhuǎn)化為紅黑樹
    if (binCount >= TREEIFY_THRESHOLD)
     treeifyBin(tab, i);
    //把舊節(jié)點值返回
    if (oldVal != null)
     return oldVal;
    break;
   }
  }
 }
 //給元素個數(shù)加 1，并有可能會觸發(fā)擴(kuò)容，比較復(fù)雜，稍后細(xì)講
 addCount(1L, binCount);
 return null;
}

initTable()方法

先看下當(dāng)數(shù)組為空時，是怎么初始化表的。

private final Node<K,V>[] initTable() {
 Node<K,V>[] tab; int sc;
 //循環(huán)判斷表是否為空，直到初始化成功為止。
 while ((tab = table) == null || tab.length == 0) {
  //sizeCtl 這個值有很多情況，默認(rèn)值為0，
  //當(dāng)為 -1 時，說明有其它線程正在對表進(jìn)行初始化操作
  //當(dāng)表初始化成功后，又會把它設(shè)置為擴(kuò)容閾值
  //當(dāng)為一個小于 -1 的負(fù)數(shù)，用來表示當(dāng)前有幾個線程正在幫助擴(kuò)容(后邊細(xì)講)
  if ((sc = sizeCtl) < 0)
   //若 sc 小于0，其實在這里就是-1，因為此時表是空的，不會發(fā)生擴(kuò)容，sc只能為正數(shù)或者-1
   //因此，當(dāng)前線程放棄 CPU 時間片，只是自旋。
   Thread.yield(); // lost initialization race; just spin
  //通過 CAS 把 sc 的值設(shè)置為-1，表明當(dāng)前線程正在進(jìn)行表的初始化，其它失敗的線程就會自旋
  else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
   try {
    //重新檢查一下表是否為空
    if ((tab = table) == null || tab.length == 0) {
     //如果sc大于0，則為sc，否則返回默認(rèn)容量 16。
     //當(dāng)調(diào)用有參構(gòu)造創(chuàng)建 Map 時，sc的值是大于0的。
     int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
     @SuppressWarnings("unchecked")
     //創(chuàng)建數(shù)組
     Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
     table = tab = nt;
     //n減去 1/4 n ，即為 0.75n ，表示擴(kuò)容閾值
     sc = n - (n >>> 2);
    }
   } finally {
    //更新 sizeCtl 為擴(kuò)容閾值
    sizeCtl = sc;
   }
   //若當(dāng)前線程初始化表成功，則跳出循環(huán)。其它自旋的線程因為判斷數(shù)組不為空，也會停止自旋
   break;
  }
 }
 return tab;
}

addCount()方法

若 put 方法元素插入成功之后，則會調(diào)用此方法，傳入?yún)?shù)為 addCount(1L, binCount)。這個方法的目的很簡單，就是把整個 table 的元素個數(shù)加 1 。但是，實現(xiàn)比較難。

我們先思考一下，如果讓我們自己去實現(xiàn)這樣的統(tǒng)計元素個數(shù)，怎么實現(xiàn)？

類比 1.8 的 HashMap ，我們可以搞一個 size 變量來存儲個數(shù)統(tǒng)計。但是，這是在多線程環(huán)境下，需要考慮并發(fā)的問題。因此，可以把 size 設(shè)置為 volatile 的，保證可見性，然后通過 CAS 樂觀鎖來自增 1。

這樣雖然也可以實現(xiàn)。但是，設(shè)想一下現(xiàn)在有非常多的線程，都在同一時間操作這個 size 變量，將會造成特別嚴(yán)重的競爭。所以，基于此，這里做了更好的優(yōu)化。讓這些競爭的線程，分散到不同的對象里邊，單獨操作它自己的數(shù)據(jù)(計數(shù)變量)，用這樣的方式盡量降低競爭。到最后需要統(tǒng)計 size 的時候，再把所有對象里邊的計數(shù)相加就可以了。

上邊提到的 size ，在此用 baseCount 表示。分散到的對象用 CounterCell 表示，對象里邊的計數(shù)變量用 value 表示。注意這里的變量都是 volatile 修飾的。

當(dāng)需要修改元素數(shù)量時，線程會先去 CAS 修改 baseCount 加1，若成功即返回。若失敗，則線程被分配到某個 CounterCell ，然后操作 value 加1。若成功，則返回。否則，給當(dāng)前線程重新分配一個 CounterCell，再嘗試給 value 加1。（這里簡略的說，實際更復(fù)雜）

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

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

//用來存儲線程和線程生成的隨機(jī)數(shù)的對應(yīng)關(guān)系
static final int getProbe() {
 return UNSAFE.getInt(Thread.currentThread(), PROBE);
}

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

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

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

//Integer.numberOfLeadingZeros方法的作用是返回 n 的最高位為1的前面的0的個數(shù)
//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 方法還沒完，別忘了有可能元素個數(shù)加 1 的操作還未成功，就走到 fullAddCount 這個方法了。看方法名，就知道了，全力增加計數(shù)值，一定要成功（奧利給）。這個方法和擴(kuò)容遷移方法是最難的，保持耐心~

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

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

transfer()方法

需要說明的一點是，雖然我們一直在說幫助擴(kuò)容，其實更準(zhǔn)確的說應(yīng)該是幫助遷移元素。因為擴(kuò)容的第一次初始化新表（擴(kuò)容后的新表）這個動作，只能由一個線程完成。其他線程都是在幫助遷移元素到新數(shù)組。

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

為了方便，上邊以原數(shù)組長度 8 為例。在元素遷移的時候，所有線程都遵循從后向前推進(jìn)的規(guī)則，即如圖A線程是第一個進(jìn)來的線程，會從下標(biāo)為7的位置，開始遷移數(shù)據(jù)。

而且當(dāng)前線程遷移時會確定一個范圍，限定它此次遷移的數(shù)據(jù)范圍，如圖 A 線程只能遷移 bound=6到 i=7 這兩個數(shù)據(jù)。

此時，其它線程就不能遷移這部分?jǐn)?shù)據(jù)了，只能繼續(xù)向前推進(jìn)，尋找其它可以遷移的數(shù)據(jù)范圍，且每次推進(jìn)的步長為固定值 stride（此處假設(shè)為2）。如圖中 B線程發(fā)現(xiàn) A 線程正在遷移6,7的數(shù)據(jù)，因此只能向前尋找，然后遷移 bound=4 到 i=5 的這兩個數(shù)據(jù)。

當(dāng)每個線程遷移完成它的范圍內(nèi)數(shù)據(jù)時，都會繼續(xù)向前推進(jìn)。那什么時候是個頭呢？

這就需要維護(hù)一個全局的變量 transferIndex，來表示所有線程總共推進(jìn)到的元素下標(biāo)位置。如圖，線程 A 第一次遷移成功后又向前推進(jìn)，然后遷移2,3 的數(shù)據(jù)。此時，若沒有其他線程在幫助遷移，則 transferIndex 即為2。

剩余部分等待下一個線程來遷移，或者有任何的 A 和B線程已經(jīng)遷移完成，也可以推進(jìn)到這里幫助遷移。直到 transferIndex=0 。（會做一些其他校驗來判斷是否遷移全部完成，看代碼）。

//這個類是一個標(biāo)志，用來代表當(dāng)前桶（數(shù)組中的某個下標(biāo)位置）的元素已經(jīng)全部遷移完成
static final class ForwardingNode<K,V> extends Node<K,V> {
 final Node<K,V>[] nextTable;
 ForwardingNode(Node<K,V>[] tab) {
  //把當(dāng)前桶的頭結(jié)點的 hash 值設(shè)置為 -1，表明已經(jīng)遷移完成，
  //這個節(jié)點中并不存儲有效的數(shù)據(jù)
  super(MOVED, null, null, null);
  this.nextTable = tab;
 }
}

//遷移數(shù)據(jù)
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
 int n = tab.length, stride;
 //根據(jù)當(dāng)前CPU核心數(shù)，確定每次推進(jìn)的步長，最小值為16.（為了方便我們以2為例）
 if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
  stride = MIN_TRANSFER_STRIDE; // subdivide range
 //從 addCount 方法，只會有一個線程跳轉(zhuǎn)到這里，初始化新數(shù)組
 if (nextTab == null) {            // initiating
  try {
   @SuppressWarnings("unchecked")
   //新數(shù)組長度為原數(shù)組的兩倍
   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 指代新數(shù)組
  nextTable = nextTab;
  //這里就把推進(jìn)的下標(biāo)值初始化為原數(shù)組長度（以16為例）
  transferIndex = n;
 }
 //新數(shù)組長度
 int nextn = nextTab.length;
 //創(chuàng)建一個標(biāo)志類
 ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
 //是否向前推進(jìn)的標(biāo)志
 boolean advance = true;
 //是否所有線程都全部遷移完成的標(biāo)志
 boolean finishing = false; // to ensure sweep before committing nextTab
 //i 代表當(dāng)前線程正在遷移的桶的下標(biāo)，bound代表它本次可以遷移的范圍下限
 for (int i = 0, bound = 0;;) {
  Node<K,V> f; int fh;
  //需要向前推進(jìn)
  while (advance) {
   int nextIndex, nextBound;
   //(1) 先看 (3) 。i每次自減 1，直到 bound。若超過bound范圍，或者finishing標(biāo)志為true，則不用向前推進(jìn)。
   //若未全部完成遷移，且 i 并未走到 bound，則跳轉(zhuǎn)到 (7)，處理當(dāng)前桶的元素遷移。
   if (--i >= bound || finishing)
    advance = false;
   //(2) 每次執(zhí)行，都會把 transferIndex 最新的值同步給 nextIndex
   //若 transferIndex小于等于0，則說明原數(shù)組中的每個桶位置，都有線程在處理遷移了，
   //于是，需要跳出while循環(huán)，并把 i設(shè)為 -1，以跳轉(zhuǎn)到④判斷在處理的線程是否已經(jīng)全部完成。
   else if ((nextIndex = transferIndex) <= 0) {
    i = -1;
    advance = false;
   }
   //(3) 第一個線程會先走到這里，確定它的數(shù)據(jù)遷移范圍。(2)處會更新 nextIndex為 transferIndex 的最新值
   //因此第一次 nextIndex=n=16，nextBound代表當(dāng)次遷移的數(shù)據(jù)范圍下限，減去步長即可，
   //所以，第一次時，nextIndex=16，nextBound=16-2=14。后續(xù)，每次都會間隔一個步長。
   else if (U.compareAndSwapInt
      (this, TRANSFERINDEX, nextIndex,
       nextBound = (nextIndex > stride ?
           nextIndex - stride : 0))) {
    //bound代表當(dāng)次數(shù)據(jù)遷移下限
    bound = nextBound;
    //第一次的i為15，因為長度16的數(shù)組，最后一個元素的下標(biāo)為15
    i = nextIndex - 1;
    //表明不需要向前推進(jìn)，只有當(dāng)把當(dāng)前范圍內(nèi)的數(shù)據(jù)全部遷移完成后，才可以向前推進(jìn)
    advance = false;
   }
  }
  //(4)
  if (i < 0 || i >= n || i + n >= nextn) {
   int sc;
   //若全部線程遷移完成
   if (finishing) {
    nextTable = null;
    //更新table為新表
    table = nextTab;
    //擴(kuò)容閾值改為原來數(shù)組長度的 3/2 ，即新長度的 3/4，也就是新數(shù)組長度的0.75倍
    sizeCtl = (n << 1) - (n >>> 1);
    return;
   }
   //到這，說明當(dāng)前線程已經(jīng)完成了自己的所有遷移（無論參與了幾次遷移），
   //則把 sc 減1，表明參與擴(kuò)容的線程數(shù)減少 1。
   if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
    //在 addCount 方法最后，我們強(qiáng)調(diào)，遷移開始時，會設(shè)置 sc=(rs << RESIZE_STAMP_SHIFT) + 2
    //每當(dāng)有一個線程參與遷移，sc 就會加 1，每當(dāng)有一個線程完成遷移，sc 就會減 1。
    //因此，這里就是去校驗當(dāng)前 sc 是否和初始值是否相等。相等，則說明全部線程遷移完成。
    if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
     return;
    //只有此處，才會把finishing 設(shè)置為true。
    finishing = advance = true;
    //這里非常有意思，會把 i 從 -1 修改為16，
    //目的就是，讓 i 再從后向前掃描一遍數(shù)組，檢查是否所有的桶都已被遷移完成，參看 (6)
    i = n; // recheck before commit
   }
  }
  //(5) 若i的位置元素為空，則說明當(dāng)前桶的元素已經(jīng)被遷移完成，就把頭結(jié)點設(shè)置為fwd標(biāo)志。
  else if ((f = tabAt(tab, i)) == null)
   advance = casTabAt(tab, i, null, fwd);
  //(6) 若當(dāng)前桶的頭結(jié)點是 ForwardingNode ，說明遷移完成，則向前推進(jìn) 
  else if ((fh = f.hash) == MOVED)
   advance = true; // already processed
  //(7) 處理當(dāng)前桶的數(shù)據(jù)遷移。
  else {
   synchronized (f) {  //給頭結(jié)點加鎖
    if (tabAt(tab, i) == f) {
     Node<K,V> ln, hn;
     //若hash值大于等于0，則說明是普通鏈表節(jié)點
     if (fh >= 0) {
      int runBit = fh & n;
      //這里是 1.7 的 CHM 的 rehash 方法和 1.8 HashMap的 resize 方法的結(jié)合體。
      //會分成兩條鏈表，一條鏈表和原來的下標(biāo)相同，另一條鏈表是原來的下標(biāo)加數(shù)組長度的位置
      //然后找到 lastRun 節(jié)點，從它到尾結(jié)點整體遷移。
      //lastRun前邊的節(jié)點則單個遷移，但是需要注意的是，這里是頭插法。
      //另外還有一點和1.7不同，1.7 lastRun前邊的節(jié)點是復(fù)制過去的，而這里是直接遷移的，沒有復(fù)制操作。
      //所以，最后會有兩條鏈表，一條鏈表從 lastRun到尾結(jié)點是正序的，而lastRun之前的元素是倒序的，
      //另外一條鏈表，從頭結(jié)點開始就是倒敘的。看下圖。
      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;
     }
     //樹節(jié)點
     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;
     }
    }
   }
  }
 }
}

遷移后的新數(shù)組鏈表方向示意圖，以 runBit =0 為例。

helpTransfer()方法

最后再看 put 方法中的這個方法，就比較簡單了。

final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
 Node<K,V>[] nextTab; int sc;
 //頭結(jié)點為 ForwardingNode ，并且新數(shù)組已經(jīng)初始化
 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) {
   //若校驗標(biāo)識失敗，或者已經(jīng)擴(kuò)容完成，或推進(jìn)下標(biāo)到頭，則退出
   if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
    sc == rs + MAX_RESIZERS || transferIndex <= 0)
    break;
   //當(dāng)前線程需要幫助遷移，sc值加1
   if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
    transfer(tab, nextTab);
    break;
   }
  }
  return nextTab;
 }
 return table;
}

JDK1.8 的 CHM 最主要的邏輯基本上都講完了，其它方法原理類同。1.8 的 ConcurrentHashMap 實現(xiàn)原理還是比較簡單的，但是代碼實現(xiàn)比較復(fù)雜。相對于 1.7 來說，鎖的粒度降低了，效率也提高了。

結(jié)語

特別推薦一個分享架構(gòu)+算法的優(yōu)質(zhì)內(nèi)容，還沒關(guān)注的小伙伴，可以長按關(guān)注一下：
長按訂閱更多精彩▼
如有收獲，點個在看，誠摯感謝