「死磕Java並發」—–J.U.C之Java並發容器：ConcurrentHashMap

原文出自:http://cmsblogs.com

作者：chenssy

此篇博客所有源碼均來自JDK 1.8

HashMap是我們用得非常頻繁的一個集合，但是由於它是非線程安全的，在多線程環境下，put操作是有可能產生死循環的，導致CPU利用率接近100%。為了解決該問題，提供了Hashtable和Collections.synchronizedMap(hashMap)兩種解決方案，但是這兩種方案都是對讀寫加鎖，獨占式，一個線程在讀時其他線程必須等待，吞吐量較低，性能較為低下。故而Doug Lea大神給我們提供了高性能的線程安全HashMap：ConcurrentHashMap。

ConcurrentHashMap的實現

ConcurrentHashMap作為Concurrent一族，其有著高效地並發操作，相比Hashtable的笨重，ConcurrentHashMap則更勝一籌了。 在1.8版本以前，ConcurrentHashMap採用分段鎖的概念，使鎖更加細化，但是1.8已經改變了這種思路，而是利用CAS+Synchronized來保證並發更新的安全，當然底層採用數組+鍊表+紅黑樹的存儲結構。 關於1.7和1.8的區別請參考占小狼博客：談談ConcurrentHashMap1.7和1.8的不同實現:http://www.jianshu.com/p/e694f1e868ec 我們從如下幾個部分全面了解ConcurrentHashMap在1.8中是如何實現的：

1. 重要概念
2. 重要內部類
3. ConcurrentHashMap的初始化
4. put操作
5. get操作
6. size操作
7. 擴容
8. 紅黑樹轉換

重要概念

ConcurrentHashMap定義了如下幾個常量：

上面是ConcurrentHashMap定義的常量，簡單易懂，就不多闡述了。下面介紹ConcurrentHashMap幾個很重要的概念。

1. table：用來存放Node節點數據的，默認為null，默認大小為16的數組，每次擴容時大小總是2的冪次方；
2. nextTable：擴容時新生成的數據，數組為table的兩倍；
3. Node：節點，保存key-value的數據結構；
4. ForwardingNode：一個特殊的Node節點，hash值為-1，其中存儲nextTable的引用。只有table發生擴容的時候，ForwardingNode才會發揮作用，作為一個占位符放在table中表示當前節點為null或則已經被移動
5. sizeCtl：控制標識符，用來控制table初始化和擴容操作的，在不同的地方有不同的用途，其值也不同，所代表的含義也不同

• 負數代表正在進行初始化或擴容操作
• -1代表正在初始化
• -N 表示有N-1個線程正在進行擴容操作
• 正數或0代表hash表還沒有被初始化，這個數值表示初始化或下一次進行擴容的大小

重要內部類

為了實現ConcurrentHashMap，Doug Lea提供了許多內部類來進行輔助實現，如Node，TreeNode,TreeBin等等。下面我們就一起來看看ConcurrentHashMap幾個重要的內部類。

Node

作為ConcurrentHashMap中最核心、最重要的內部類，Node擔負著重要角色：key-value鍵值對。所有插入ConCurrentHashMap的中數據都將會包裝在Node中。定義如下：

在Node內部類中，其屬性value、next都是帶有volatile的。同時其對value的setter方法進行了特殊處理，不允許直接調用其setter方法來修改value的值。最後Node還提供了find方法來賦值map.get()。

TreeNode

我們在學習HashMap的時候就知道，HashMap的核心數據結構就是鍊表。在ConcurrentHashMap中就不一樣了，如果鍊表的數據過長是會轉換為紅黑樹來處理。當它並不是直接轉換，而是將這些鍊表的節點包裝成TreeNode放在TreeBin對象中，然後由TreeBin完成紅黑樹的轉換。所以TreeNode也必須是ConcurrentHashMap的一個核心類，其為樹節點類，定義如下：

源碼展示TreeNode繼承Node，且提供了findTreeNode用來查找查找hash為h，key為k的節點。

TreeBin

該類並不負責key-value的鍵值對包裝，它用於在鍊表轉換為紅黑樹時包裝TreeNode節點，也就是說ConcurrentHashMap紅黑樹存放是TreeBin，不是TreeNode。該類封裝了一系列的方法，包括putTreeVal、lookRoot、UNlookRoot、remove、balanceInsetion、balanceDeletion。由於TreeBin的代碼太長我們這裡只展示構造方法（構造方法就是構造紅黑樹的過程）：

通過構造方法是不是發現了部分端倪，構造方法就是在構造一個紅黑樹的過程。

ForwardingNode

這是一個真正的輔助類，該類僅僅只存活在ConcurrentHashMap擴容操作時。只是一個標誌節點，並且指向nextTable，它提供find方法而已。該類也是集成Node節點，其hash為-1，key、value、next均為null。如下：

構造函數

ConcurrentHashMap提供了一系列的構造函數用於創建ConcurrentHashMap對象：

初始化： initTable()

ConcurrentHashMap的初始化主要由initTable()方法實現，在上面的構造函數中我們可以看到，其實ConcurrentHashMap在構造函數中並沒有做什麼事，僅僅只是設置了一些參數而已。其真正的初始化是發生在插入的時候，例如put、merge、compute、computeIfAbsent、computeIfPresent操作時。其方法定義如下：

初始化方法initTable()的關鍵就在於sizeCtl，該值默認為0，如果在構造函數時有參數傳入該值則為2的冪次方。該值如果 < 0，表示有其他線程正在初始化，則必須暫停該線程。如果線程獲得了初始化的權限則先將sizeCtl設置為-1，防止有其他線程進入，最後將sizeCtl設置0.75 * n，表示擴容的閾值。

put操作

ConcurrentHashMap最常用的put、get操作，ConcurrentHashMap的put操作與HashMap並沒有多大區別，其核心思想依然是根據hash值計算節點插入在table的位置，如果該位置為空，則直接插入，否則插入到鍊表或者樹中。但是ConcurrentHashMap會涉及到多線程情況就會複雜很多。我們先看原始碼，然後根據原始碼一步一步分析：

 public V put(K key, V value) {
 return putVal(key, value, false);
 }
 final V putVal(K key, V value, boolean onlyIfAbsent) {
 //key、value均不能為null
 if (key == null || value == null) throw new NullPointerException();
 //計算hash值
 int hash = spread(key.hashCode());
 int binCount = 0;
 for (Node[] tab = table;;) {
 Node f; int n, i, fh;
 // table為null，進行初始化工作
 if (tab == null || (n = tab.length) == 0)
 tab = initTable();
 //如果i位置沒有節點，則直接插入，不需要加鎖
 else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
 if (casTabAt(tab, i, null,
 new Node(hash, key, value, null)))
 break; // no lock when adding to empty bin
 }
 // 有線程正在進行擴容操作，則先幫助擴容
 else if ((fh = f.hash) == MOVED)
 tab = helpTransfer(tab, f);
 else {
 V oldVal = null;
 //對該節點進行加鎖處理（hash值相同的鍊表的頭節點），對性能有點兒影響
 synchronized (f) {
 if (tabAt(tab, i) == f) {
 //fh > 0 表示為鍊表，將該節點插入到鍊表尾部
 if (fh >= 0) {
 binCount = 1;
 for (Node e = f;; ++binCount) {
 K ek;
 //hash 和 key 都一樣，替換value
 if (e.hash == hash &&
 ((ek = e.key) == key ||
 (ek != null && key.equals(ek)))) {
 oldVal = e.val;
 //putIfAbsent()
 if (!onlyIfAbsent)
 e.val = value;
 break;
 }
 Node pred = e;
 //鍊表尾部 直接插入
 if ((e = e.next) == null) {
 pred.next = new Node(hash, key,
 value, null);
 break;
 }
 }
 }
 //樹節點，按照樹的插入操作進行插入
 else if (f instanceof TreeBin) {
 Node p;
 binCount = 2;
 if ((p = ((TreeBin)f).putTreeVal(hash, key,
 value)) != null) {
 oldVal = p.val;
 if (!onlyIfAbsent)
 p.val = value;
 }
 }
 }
 }
 if (binCount != 0) {
 // 如果鍊表長度已經達到臨界值8 就需要把鍊表轉換為樹結構
 if (binCount >= TREEIFY_THRESHOLD)
 treeifyBin(tab, i);
 if (oldVal != null)
 return oldVal;
 break;
 }
 }
 }
 //size + 1 
 addCount(1L, binCount);
 return null;
 }

按照上面的源碼，我們可以確定put整個流程如下：

• 判空；ConcurrentHashMap的key、value都不允許為null
• 計算hash。利用方法計算hash值。

 static final int spread(int h) {
 return (h ^ (h >>> 16)) & HASH_BITS;
 }

• 遍歷table，進行節點插入操作，過程如下：
• 如果table為空，則表示ConcurrentHashMap還沒有初始化，則進行初始化操作：initTable()
• 根據hash值獲取節點的位置i，若該位置為空，則直接插入，這個過程是不需要加鎖的。計算f位置：i=(n - 1) & hash
• 如果檢測到fh = f.hash == -1，則f是ForwardingNode節點，表示有其他線程正在進行擴容操作，則幫助線程一起進行擴容操作
• 如果f.hash >= 0 表示是鍊表結構，則遍歷鍊表，如果存在當前key節點則替換value，否則插入到鍊表尾部。如果f是TreeBin類型節點，則按照紅黑樹的方法更新或者增加節點
• 若鍊表長度 > TREEIFY_THRESHOLD(默認是8)，則將鍊表轉換為紅黑樹結構
• 調用addCount方法，ConcurrentHashMap的size + 1

這裡整個put操作已經完成。

get操作

ConcurrentHashMap的get操作還是挺簡單的，無非就是通過hash來找key相同的節點而已，當然需要區分鍊表和樹形兩種情況。

 public V get(Object key) {
 Node[] tab; Node e, p; int n, eh; K ek;
 // 計算hash
 int h = spread(key.hashCode());
 if ((tab = table) != null && (n = tab.length) > 0 &&
 (e = tabAt(tab, (n - 1) & h)) != null) {
 // 搜索到的節點key與傳入的key相同且不為null,直接返回這個節點
 if ((eh = e.hash) == h) {
 if ((ek = e.key) == key || (ek != null && key.equals(ek)))
 return e.val;
 }
 // 樹
 else if (eh < 0)
 return (p = e.find(h, key)) != null ? p.val : null;
 // 鍊表，遍歷
 while ((e = e.next) != null) {
 if (e.hash == h &&
 ((ek = e.key) == key || (ek != null && key.equals(ek))))
 return e.val;
 }
 }
 return null;
 }

get操作的整個邏輯非常清楚：

• 計算hash值
• 判斷table是否為空，如果為空，直接返回null
• 根據hash值獲取table中的Node節點（tabAt(tab, (n - 1) & h)），然後根據鍊表或者樹形方式找到相對應的節點，返回其value值。

size 操作

ConcurrentHashMap的size()方法我們雖然用得不是很多，但是我們還是很有必要去了解的。ConcurrentHashMap的size()方法返回的是一個不精確的值，因為在進行統計的時候有其他線程正在進行插入和刪除操作。當然為了這個不精確的值，ConcurrentHashMap也是操碎了心。 為了更好地統計size，ConcurrentHashMap提供了baseCount、counterCells兩個輔助變量和一個CounterCell輔助內部類。

 @sun.misc.Contended static final class CounterCell {
 volatile long value;
 CounterCell(long x) { value = x; }
 }
 //ConcurrentHashMap中元素個數,但返回的不一定是當前Map的真實元素個數。基於CAS無鎖更新
 private transient volatile long baseCount;
 private transient volatile CounterCell[] counterCells;

這裡我們需要清楚CounterCell 的定義 size()方法定義如下：

 public int size() {
 long n = sumCount();
 return ((n < 0L) ? 0 :
 (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
 (int)n);
 }

內部調用sunmCount()：

 final long sumCount() {
 CounterCell[] as = counterCells; CounterCell a;
 long sum = baseCount;
 if (as != null) {
 for (int i = 0; i < as.length; ++i) {
 //遍歷，所有counter求和
 if ((a = as[i]) != null)
 sum += a.value; 
 }
 }
 return sum;
 }

sumCount()就是疊代counterCells來統計sum的過程。我們知道put操作時，肯定會影響size()，我們就來看看CouncurrentHashMap是如何為了這個不和諧的size()操碎了心。 在put()方法最後會調用addCount()方法，該方法主要做兩件事，一件更新baseCount的值，第二件檢測是否進行擴容，我們只看更新baseCount部分：

 private final void addCount(long x, int check) {
 CounterCell[] as; long b, s;
 // s = b + x，完成baseCount++操作；
 if ((as = counterCells) != null ||
 !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
 CounterCell a; long v; int m;
 boolean uncontended = true;
 if (as == null || (m = as.length - 1) < 0 ||
 (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
 !(uncontended =
 U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
 // 多線程CAS發生失敗時執行
 fullAddCount(x, uncontended);
 return;
 }
 if (check <= 1)
 return;
 s = sumCount();
 }
 // 檢查是否進行擴容
 }

x == 1，如果counterCells == null，則U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)，如果並發競爭比較大可能會導致改過程失敗，如果失敗則最終會調用fullAddCount()方法。其實為了提高高並發的時候baseCount可見性的失敗問題，又避免一直重試，JDK 8 引入了類Striped64,其中LongAdder和DoubleAdder都是基於該類實現的，而CounterCell也是基於Striped64實現的。如果counterCells ！= null，且uncontended = U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x)也失敗了，同樣會調用fullAddCount()方法，最後調用sumCount()計算s。 其實在1.8中，它不推薦size()方法，而是推崇mappingCount()方法，該方法的定義和size()方法基本一致：

 public long mappingCount() {
 long n = sumCount();
 return (n < 0L) ? 0L : n; // ignore transient negative values
 }

擴容操作

當ConcurrentHashMap中table元素個數達到了容量閾值（sizeCtl）時，則需要進行擴容操作。在put操作時最後一個會調用addCount(long x, int check)，該方法主要做兩個工作：1.更新baseCount；2.檢測是否需要擴容操作。如下：

 private final void addCount(long x, int check) {
 CounterCell[] as; long b, s;
 // 更新baseCount
 //check >= 0 :則需要進行擴容操作
 if (check >= 0) {
 Node[] tab, nt; int n, sc;
 while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
 (n = tab.length) < MAXIMUM_CAPACITY) {
 int rs = resizeStamp(n);
 if (sc < 0) {
 if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
 sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
 transferIndex <= 0)
 break;
 if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
 transfer(tab, nt);
 }
 //當前線程是唯一的或是第一個發起擴容的線程 此時nextTable=null
 else if (U.compareAndSwapInt(this, SIZECTL, sc,
 (rs << RESIZE_STAMP_SHIFT) + 2))
 transfer(tab, null);
 s = sumCount();
 }
 }
 }

transfer()方法為ConcurrentHashMap擴容操作的核心方法。由於ConcurrentHashMap支持多線程擴容，而且也沒有進行加鎖，所以實現會變得有點兒複雜。整個擴容操作分為兩步：

1. 構建一個nextTable，其大小為原來大小的兩倍，這個步驟是在單線程環境下完成的
2. 將原來table裡面的內容複製到nextTable中，這個步驟是允許多線程操作的，所以性能得到提升，減少了擴容的時間消耗

我們先來看看原始碼，然後再一步一步分析：

 private final void transfer(Node[] tab, Node[] nextTab) {
 int n = tab.length, stride;
 // 每核處理的量小於16，則強制賦值16
 if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
 stride = MIN_TRANSFER_STRIDE; // subdivide range
 if (nextTab == null) { // initiating
 try {
 @SuppressWarnings("unchecked")
 Node[] nt = (Node[])new Node[n << 1]; //構建一個nextTable對象，其容量為原來容量的兩倍
 nextTab = nt;
 } catch (Throwable ex) { // try to cope with OOME
 sizeCtl = Integer.MAX_VALUE;
 return;
 }
 nextTable = nextTab;
 transferIndex = n;
 }
 int nextn = nextTab.length;
 // 連接點指針，用於標誌位（fwd的hash值為-1，fwd.nextTable=nextTab）
 ForwardingNode fwd = new ForwardingNode(nextTab);
 // 當advance == true時，表明該節點已經處理過了
 boolean advance = true;
 boolean finishing = false; // to ensure sweep before committing nextTab
 for (int i = 0, bound = 0;;) {
 Node f; int fh;
 // 控制 --i ,遍歷原hash表中的節點
 while (advance) {
 int nextIndex, nextBound;
 if (--i >= bound || finishing)
 advance = false;
 else if ((nextIndex = transferIndex) <= 0) {
 i = -1;
 advance = false;
 }
 // 用CAS計算得到的transferIndex
 else if (U.compareAndSwapInt
 (this, TRANSFERINDEX, nextIndex,
 nextBound = (nextIndex > stride ?
 nextIndex - stride : 0))) {
 bound = nextBound;
 i = nextIndex - 1;
 advance = false;
 }
 }
 if (i < 0 || i >= n || i + n >= nextn) {
 int sc;
 // 已經完成所有節點複製了
 if (finishing) {
 nextTable = null;
 table = nextTab; // table 指向nextTable
 sizeCtl = (n << 1) - (n >>> 1); // sizeCtl閾值為原來的1.5倍
 return; // 跳出死循環，
 }
 // CAS 更擴容閾值，在這裡面sizectl值減一，說明新加入一個線程參與到擴容操作
 if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
 if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
 return;
 finishing = advance = true;
 i = n; // recheck before commit
 }
 }
 // 遍歷的節點為null，則放入到ForwardingNode 指針節點
 else if ((f = tabAt(tab, i)) == null)
 advance = casTabAt(tab, i, null, fwd);
 // f.hash == -1 表示遍歷到了ForwardingNode節點，意味著該節點已經處理過了
 // 這裡是控制並發擴容的核心
 else if ((fh = f.hash) == MOVED)
 advance = true; // already processed
 else {
 // 節點加鎖
 synchronized (f) {
 // 節點複製工作
 if (tabAt(tab, i) == f) {
 Node ln, hn;
 // fh >= 0 ,表示為鍊表節點
 if (fh >= 0) {
 // 構造兩個鍊表 一個是原鍊表 另一個是原鍊表的反序排列
 int runBit = fh & n;
 Node lastRun = f;
 for (Node 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 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(ph, pk, pv, ln);
 else
 hn = new Node(ph, pk, pv, hn);
 }
 // 在nextTable i 位置處插上鍊表
 setTabAt(nextTab, i, ln);
 // 在nextTable i + n 位置處插上鍊表
 setTabAt(nextTab, i + n, hn);
 // 在table i 位置處插上ForwardingNode 表示該節點已經處理過了
 setTabAt(tab, i, fwd);
 // advance = true 可以執行--i動作，遍歷節點
 advance = true;
 }
 // 如果是TreeBin，則按照紅黑樹進行處理，處理邏輯與上面一致
 else if (f instanceof TreeBin) {
 TreeBin t = (TreeBin)f;
 TreeNode lo = null, loTail = null;
 TreeNode hi = null, hiTail = null;
 int lc = 0, hc = 0;
 for (Node e = t.first; e != null; e = e.next) {
 int h = e.hash;
 TreeNode p = new TreeNode
 (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;
 }
 }
 // 擴容後樹節點個數若<=6，將樹轉鍊表
 ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
 (hc != 0) ? new TreeBin(lo) : t;
 hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
 (lc != 0) ? new TreeBin(hi) : t;
 setTabAt(nextTab, i, ln);
 setTabAt(nextTab, i + n, hn);
 setTabAt(tab, i, fwd);
 advance = true;
 }
 }
 }
 }
 }
 }

上面的源碼有點兒長，稍微複雜了一些，在這裡我們拋棄它多線程環境，我們從單線程角度來看：

1. 為每個內核分任務，並保證其不小於16
2. 檢查nextTable是否為null，如果是，則初始化nextTable，使其容量為table的兩倍
3. 死循環遍歷節點，知道finished：節點從table複製到nextTable中，支持並發，請思路如下：

• 如果節點 f 為null，則插入ForwardingNode（採用Unsafe.compareAndSwapObjectf方法實現），這個是觸發並發擴容的關鍵
• 如果f為鍊表的頭節點（fh >= 0）,則先構造一個反序鍊表，然後把他們分別放在nextTable的i和i + n位置，並將ForwardingNode 插入原節點位置，代表已經處理過了
• 如果f為TreeBin節點，同樣也是構造一個反序 ，同時需要判斷是否需要進行unTreeify()操作，並把處理的結果分別插入到nextTable的i 和i+nw位置，並插入ForwardingNode 節點

1. 所有節點複製完成後，則將table指向nextTable，同時更新sizeCtl = nextTable的0.75倍，完成擴容過程

在多線程環境下，ConcurrentHashMap用兩點來保證正確性：ForwardingNode和synchronized。當一個線程遍歷到的節點如果是ForwardingNode，則繼續往後遍歷，如果不是，則將該節點加鎖，防止其他線程進入，完成後設置ForwardingNode節點，以便要其他線程可以看到該節點已經處理過了，如此交叉進行，高效而又安全。 下圖是擴容的過程（來自：http://blog.csdn.net/u010723709/article/details/48007881）：

在put操作時如果發現fh.hash = -1，則表示正在進行擴容操作，則當前線程會協助進行擴容操作。

 else if ((fh = f.hash) == MOVED)
 tab = helpTransfer(tab, f);

helpTransfer()方法為協助擴容方法，當調用該方法的時候，nextTable一定已經創建了，所以該方法主要則是進行複製工作。如下：

 final Node[] helpTransfer(Node[] tab, Node f) {
 Node[] nextTab; int sc;
 if (tab != null && (f instanceof ForwardingNode) &&
 (nextTab = ((ForwardingNode)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;
 if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
 transfer(tab, nextTab);
 break;
 }
 }
 return nextTab;
 }
 return table;
 }

轉換紅黑樹

在put操作是，如果發現鍊表結構中的元素超過了TREEIFY_THRESHOLD（默認為8），則會把鍊表轉換為紅黑樹，已便於提高查詢效率。如下：

if (binCount >= TREEIFY_THRESHOLD)
 treeifyBin(tab, i);

調用treeifyBin方法用與將鍊表轉換為紅黑樹。

private final void treeifyBin(Node[] tab, int index) {
 Node b; int n, sc;
 if (tab != null) {
 if ((n = tab.length) < MIN_TREEIFY_CAPACITY)//如果table.length<64 就擴大一倍 返回
 tryPresize(n << 1);
 else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
 synchronized (b) {
 if (tabAt(tab, index) == b) {
 TreeNode hd = null, tl = null;
 //構造了一個TreeBin對象 把所有Node節點包裝成TreeNode放進去
 for (Node e = b; e != null; e = e.next) {
 TreeNode p =
 new TreeNode(e.hash, e.key, e.val,
 null, null);//這裡只是利用了TreeNode封裝 而沒有利用TreeNode的next域和parent域
 if ((p.prev = tl) == null)
 hd = p;
 else
 tl.next = p;
 tl = p;
 }
 //在原來index的位置 用TreeBin替換掉原來的Node對象
 setTabAt(tab, index, new TreeBin(hd));
 }
 }
 }
 }
 }

從上面源碼可以看出，構建紅黑樹的過程是同步的，進入同步後過程如下：

1. 根據table中index位置Node鍊表，重新生成一個hd為頭結點的TreeNode
2. 根據hd頭結點，生成TreeBin樹結構，並用TreeBin替換掉原來的Node對象。

整個紅黑樹的構建過程有點兒複雜，關於ConcurrentHashMap 紅黑樹的構建過程，我們後續分析。 【注】：ConcurrentHashMap的擴容和鍊表轉紅黑樹稍微複雜點，後續另起博文分析