點個贊，看一看，好習慣！本文 GitHub https://github.com/OUYANGSIHAI/JavaInterview 已收錄，這是我花了 3 個月總結的一線大廠 Java 面試總結，本人已拿大廠 offer。
另外，原創文章首發在我的個人博客：blog.ouyangsihai.cn，歡迎訪問。

Java8 ConcurrentHashMap

Java7 中實現的 ConcurrentHashMap 說實話還是比較復雜的，Java8 對 ConcurrentHashMap 進行了比較大的改動。建議讀者可以參考 Java8 中 HashMap 相對于 Java7 HashMap 的改動，對于 ConcurrentHashMap，Java8 也引入了紅黑樹。

說實話，Java8 ConcurrentHashMap 源碼真心不簡單，最難的在于擴容，數據遷移操作不容易看懂。

我們先用一個示意圖來描述下其結構：

結構上和 Java8 的 HashMap 基本上一樣，不過它要保證線程安全性，所以在源碼上確實要復雜一些。
##初始化

// 這構造函數里，什么都不干 public ConcurrentHashMap() { }public ConcurrentHashMap(int initialCapacity) {if (initialCapacity < 0)throw new IllegalArgumentException();int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?MAXIMUM_CAPACITY :tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));this.sizeCtl = cap; }

這個初始化方法有點意思，通過提供初始容量，計算了 sizeCtl，sizeCtl = 【 (1.5 * initialCapacity + 1)，然后向上取最近的 2 的 n 次方】。如 initialCapacity 為 10，那么得到 sizeCtl 為 16，如果 initialCapacity 為 11，得到 sizeCtl 為 32。

sizeCtl 這個屬性使用的場景很多，不過只要跟著文章的思路來，就不會被它搞暈了。

如果你愛折騰，也可以看下另一個有三個參數的構造方法，這里我就不說了，大部分時候，我們會使用無參構造函數進行實例化，我們也按照這個思路來進行源碼分析吧。
##put 過程分析

仔細地一行一行代碼看下去：

public V put(K key, V value) {return putVal(key, value, false); }final V putVal(K key, V value, boolean onlyIfAbsent) {if (key == null || value == null) throw new NullPointerException();// 得到 hash 值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();// 找該 hash 值對應的數組下標，得到第一個節點 felse if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {// 如果數組該位置為空，// 用一次 CAS 操作將這個新值放入其中即可，這個 put 操作差不多就結束了，可以拉到最后面了// 如果 CAS 失敗，那就是有并發操作，進到下一個循環就好了if (casTabAt(tab, i, null,new Node<K,V>(hash, key, value, null)))break; // no lock when adding to empty bin}// hash 居然可以等于 MOVED，這個需要到后面才能看明白，不過從名字上也能猜到，肯定是因為在擴容else if ((fh = f.hash) == MOVED)// 幫助數據遷移，這個等到看完數據遷移部分的介紹后，再理解這個就很簡單了tab = helpTransfer(tab, f);else { // 到這里就是說，f 是該位置的頭結點，而且不為空V oldVal = null;// 獲取數組該位置的頭結點的監視器鎖synchronized (f) {if (tabAt(tab, i) == f) {if (fh >= 0) { // 頭結點的 hash 值大于 0，說明是鏈表// 用于累加，記錄鏈表的長度binCount = 1;// 遍歷鏈表for (Node<K,V> e = f;; ++binCount) {K ek;// 如果發現了"相等"的 key，判斷是否要進行值覆蓋，然后也就可以 break 了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;}}}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;}}}}// binCount != 0 說明上面在做鏈表操作if (binCount != 0) {// 判斷是否要將鏈表轉換為紅黑樹，臨界值和 HashMap 一樣，也是 8if (binCount >= TREEIFY_THRESHOLD)// 這個方法和 HashMap 中稍微有一點點不同，那就是它不是一定會進行紅黑樹轉換，// 如果當前數組的長度小于 64，那么會選擇進行數組擴容，而不是轉換為紅黑樹// 具體源碼我們就不看了，擴容部分后面說treeifyBin(tab, i);if (oldVal != null)return oldVal;break;}}}// addCount(1L, binCount);return null; }

put 的主流程看完了，但是至少留下了幾個問題，第一個是初始化，第二個是擴容，第三個是幫助數據遷移，這些我們都會在后面進行一一介紹。
##初始化數組：initTable

這個比較簡單，主要就是初始化一個合適大小的數組，然后會設置 sizeCtl。

初始化方法中的并發問題是通過對 sizeCtl 進行一個 CAS 操作來控制的。

private final Node<K,V>[] initTable() {Node<K,V>[] tab; int sc;while ((tab = table) == null || tab.length == 0) {// 初始化的"功勞"被其他線程"搶去"了if ((sc = sizeCtl) < 0)Thread.yield(); // lost initialization race; just spin// CAS 一下，將 sizeCtl 設置為 -1，代表搶到了鎖else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {try {if ((tab = table) == null || tab.length == 0) {// DEFAULT_CAPACITY 默認初始容量是 16int n = (sc > 0) ? sc : DEFAULT_CAPACITY;// 初始化數組，長度為 16 或初始化時提供的長度Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];// 將這個數組賦值給 table，table 是 volatile 的table = tab = nt;// 如果 n 為 16 的話，那么這里 sc = 12// 其實就是 0.75 * nsc = n - (n >>> 2);}} finally {// 設置 sizeCtl 為 sc，我們就當是 12 吧sizeCtl = sc;}break;}}return tab; }

##鏈表轉紅黑樹: treeifyBin

前面我們在 put 源碼分析也說過，treeifyBin 不一定就會進行紅黑樹轉換，也可能是僅僅做數組擴容。我們還是進行源碼分析吧。

private final void treeifyBin(Node<K,V>[] tab, int index) {Node<K,V> b; int n, sc;if (tab != null) {// MIN_TREEIFY_CAPACITY 為 64// 所以，如果數組長度小于 64 的時候，其實也就是 32 或者 16 或者更小的時候，會進行數組擴容if ((n = tab.length) < MIN_TREEIFY_CAPACITY)// 后面我們再詳細分析這個方法tryPresize(n << 1);// b 是頭結點else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {// 加鎖synchronized (b) {if (tabAt(tab, index) == b) {// 下面就是遍歷鏈表，建立一顆紅黑樹TreeNode<K,V> hd = null, tl = null;for (Node<K,V> e = b; e != null; e = e.next) {TreeNode<K,V> p =new TreeNode<K,V>(e.hash, e.key, e.val,null, null);if ((p.prev = tl) == null)hd = p;elsetl.next = p;tl = p;}// 將紅黑樹設置到數組相應位置中setTabAt(tab, index, new TreeBin<K,V>(hd));}}}} }

##擴容：tryPresize

如果說 Java8 ConcurrentHashMap 的源碼不簡單，那么說的就是擴容操作和遷移操作。

這個方法要完完全全看懂還需要看之后的 transfer 方法，讀者應該提前知道這點。

這里的擴容也是做翻倍擴容的，擴容后數組容量為原來的 2 倍。

// 首先要說明的是，方法參數 size 傳進來的時候就已經翻了倍了 private final void tryPresize(int size) {// c：size 的 1.5 倍，再加 1，再往上取最近的 2 的 n 次方。int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :tableSizeFor(size + (size >>> 1) + 1);int sc;while ((sc = sizeCtl) >= 0) {Node<K,V>[] tab = table; int n;// 這個 if 分支和之前說的初始化數組的代碼基本上是一樣的，在這里，我們可以不用管這塊代碼if (tab == null || (n = tab.length) == 0) {n = (sc > c) ? sc : c;if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {try {if (table == tab) {@SuppressWarnings("unchecked")Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];table = nt;sc = n - (n >>> 2); // 0.75 * n}} finally {sizeCtl = sc;}}}else if (c <= sc || n >= MAXIMUM_CAPACITY)break;else if (tab == table) {// 我沒看懂 rs 的真正含義是什么，不過也關系不大int rs = resizeStamp(n);if (sc < 0) {Node<K,V>[] nt;if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||transferIndex <= 0)break;// 2. 用 CAS 將 sizeCtl 加 1，然后執行 transfer 方法// 此時 nextTab 不為 nullif (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))transfer(tab, nt);}// 1. 將 sizeCtl 設置為 (rs << RESIZE_STAMP_SHIFT) + 2)// 我是沒看懂這個值真正的意義是什么？不過可以計算出來的是，結果是一個比較大的負數// 調用 transfer 方法，此時 nextTab 參數為 nullelse if (U.compareAndSwapInt(this, SIZECTL, sc,(rs << RESIZE_STAMP_SHIFT) + 2))transfer(tab, null);}} }

這個方法的核心在于 sizeCtl 值的操作，首先將其設置為一個負數，然后執行 transfer(tab, null)，再下一個循環將 sizeCtl 加 1，并執行 transfer(tab, nt)，之后可能是繼續 sizeCtl 加 1，并執行 transfer(tab, nt)。

所以，可能的操作就是執行 1 次 transfer(tab, null) + 多次 transfer(tab, nt)，這里怎么結束循環的需要看完 transfer 源碼才清楚。
##數據遷移：transfer

下面這個方法很點長，將原來的 tab 數組的元素遷移到新的 nextTab 數組中。

雖然我們之前說的 tryPresize 方法中多次調用 transfer 不涉及多線程，但是這個 transfer 方法可以在其他地方被調用，典型地，我們之前在說 put 方法的時候就說過了，請往上看 put 方法，是不是有個地方調用了 helpTransfer 方法，helpTransfer 方法會調用 transfer 方法的。

此方法支持多線程執行，外圍調用此方法的時候，會保證第一個發起數據遷移的線程，nextTab 參數為 null，之后再調用此方法的時候，nextTab 不會為 null。

閱讀源碼之前，先要理解并發操作的機制。原數組長度為 n，所以我們有 n 個遷移任務，讓每個線程每次負責一個小任務是最簡單的，每做完一個任務再檢測是否有其他沒做完的任務，幫助遷移就可以了，而 Doug Lea 使用了一個 stride，簡單理解就是步長，每個線程每次負責遷移其中的一部分，如每次遷移 16 個小任務。所以，我們就需要一個全局的調度者來安排哪個線程執行哪幾個任務，這個就是屬性 transferIndex 的作用。

第一個發起數據遷移的線程會將 transferIndex 指向原數組最后的位置，然后從后往前的 stride 個任務屬于第一個線程，然后將 transferIndex 指向新的位置，再往前的 stride 個任務屬于第二個線程，依此類推。當然，這里說的第二個線程不是真的一定指代了第二個線程，也可以是同一個線程，這個讀者應該能理解吧。其實就是將一個大的遷移任務分為了一個個任務包。

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {int n = tab.length, stride;// stride 在單核下直接等于 n，多核模式下為 (n>>>3)/NCPU，最小值是 16// stride 可以理解為”步長“，有 n 個位置是需要進行遷移的，// 將這 n 個任務分為多個任務包，每個任務包有 stride 個任務if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)stride = MIN_TRANSFER_STRIDE; // subdivide range// 如果 nextTab 為 null，先進行一次初始化// 前面我們說了，外圍會保證第一個發起遷移的線程調用此方法時，參數 nextTab 為 null// 之后參與遷移的線程調用此方法時，nextTab 不會為 nullif (nextTab == null) {try {// 容量翻倍Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];nextTab = nt;} catch (Throwable ex) { // try to cope with OOMEsizeCtl = Integer.MAX_VALUE;return;}// nextTable 是 ConcurrentHashMap 中的屬性nextTable = nextTab;// transferIndex 也是 ConcurrentHashMap 的屬性，用于控制遷移的位置transferIndex = n;}int nextn = nextTab.length;// ForwardingNode 翻譯過來就是正在被遷移的 Node// 這個構造方法會生成一個Node，key、value 和 next 都為 null，關鍵是 hash 為 MOVED// 后面我們會看到，原數組中位置 i 處的節點完成遷移工作后，// 就會將位置 i 處設置為這個 ForwardingNode，用來告訴其他線程該位置已經處理過了// 所以它其實相當于是一個標志。ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);// advance 指的是做完了一個位置的遷移工作，可以準備做下一個位置的了boolean advance = true;boolean finishing = false; // to ensure sweep before committing nextTab/** 下面這個 for 循環，最難理解的在前面，而要看懂它們，應該先看懂后面的，然后再倒回來看* */// i 是位置索引，bound 是邊界，注意是從后往前for (int i = 0, bound = 0;;) {Node<K,V> f; int fh;// 下面這個 while 真的是不好理解// advance 為 true 表示可以進行下一個位置的遷移了// 簡單理解結局：i 指向了 transferIndex，bound 指向了 transferIndex-stridewhile (advance) {int nextIndex, nextBound;if (--i >= bound || finishing)advance = false;// 將 transferIndex 值賦給 nextIndex// 這里 transferIndex 一旦小于等于 0，說明原數組的所有位置都有相應的線程去處理了else if ((nextIndex = transferIndex) <= 0) {i = -1;advance = false;}else if (U.compareAndSwapInt(this, TRANSFERINDEX, nextIndex,nextBound = (nextIndex > stride ?nextIndex - stride : 0))) {// 看括號中的代碼，nextBound 是這次遷移任務的邊界，注意，是從后往前bound = nextBound;i = nextIndex - 1;advance = false;}}if (i < 0 || i >= n || i + n >= nextn) {int sc;if (finishing) {// 所有的遷移操作已經完成nextTable = null;// 將新的 nextTab 賦值給 table 屬性，完成遷移table = nextTab;// 重新計算 sizeCtl：n 是原數組長度，所以 sizeCtl 得出的值將是新數組長度的 0.75 倍sizeCtl = (n << 1) - (n >>> 1);return;}// 之前我們說過，sizeCtl 在遷移前會設置為 (rs << RESIZE_STAMP_SHIFT) + 2// 然后，每有一個線程參與遷移就會將 sizeCtl 加 1，// 這里使用 CAS 操作對 sizeCtl 進行減 1，代表做完了屬于自己的任務if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {// 任務結束，方法退出if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)return;// 到這里，說明 (sc - 2) == resizeStamp(n) << RESIZE_STAMP_SHIFT，// 也就是說，所有的遷移任務都做完了，也就會進入到上面的 if(finishing){} 分支了finishing = advance = true;i = n; // recheck before commit}}// 如果位置 i 處是空的，沒有任何節點，那么放入剛剛初始化的 ForwardingNode ”空節點“else if ((f = tabAt(tab, i)) == null)advance = casTabAt(tab, i, null, fwd);// 該位置處是一個 ForwardingNode，代表該位置已經遷移過了else if ((fh = f.hash) == MOVED)advance = true; // already processedelse {// 對數組該位置處的結點加鎖，開始處理數組該位置處的遷移工作synchronized (f) {if (tabAt(tab, i) == f) {Node<K,V> ln, hn;// 頭結點的 hash 大于 0，說明是鏈表的 Node 節點if (fh >= 0) {// 下面這一塊和 Java7 中的 ConcurrentHashMap 遷移是差不多的，// 需要將鏈表一分為二，// 找到原鏈表中的 lastRun，然后 lastRun 及其之后的節點是一起進行遷移的// lastRun 之前的節點需要進行克隆，然后分到兩個鏈表中int runBit = fh & n;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);elsehn = new Node<K,V>(ph, pk, pv, hn);}// 其中的一個鏈表放在新數組的位置 isetTabAt(nextTab, i, ln);// 另一個鏈表放在新數組的位置 i+nsetTabAt(nextTab, i + n, hn);// 將原數組該位置處設置為 fwd，代表該位置已經處理完畢，// 其他線程一旦看到該位置的 hash 值為 MOVED，就不會進行遷移了setTabAt(tab, i, fwd);// advance 設置為 true，代表該位置已經遷移完畢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;elseloTail.next = p;loTail = p;++lc;}else {if ((p.prev = hiTail) == null)hi = p;elsehiTail.next = p;hiTail = p;++hc;}}// 如果一分為二后，節點數少于 8，那么將紅黑樹轉換回鏈表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;// 將 ln 放置在新數組的位置 isetTabAt(nextTab, i, ln);// 將 hn 放置在新數組的位置 i+nsetTabAt(nextTab, i + n, hn);// 將原數組該位置處設置為 fwd，代表該位置已經處理完畢，// 其他線程一旦看到該位置的 hash 值為 MOVED，就不會進行遷移了setTabAt(tab, i, fwd);// advance 設置為 true，代表該位置已經遷移完畢advance = true;}}}}} }

說到底，transfer 這個方法并沒有實現所有的遷移任務，每次調用這個方法只實現了 transferIndex 往前 stride 個位置的遷移工作，其他的需要由外圍來控制。

這個時候，再回去仔細看 tryPresize 方法可能就會更加清晰一些了。
##get 過程分析

get 方法從來都是最簡單的，這里也不例外：

1、計算 hash 值
2、根據 hash 值找到數組對應位置: (n - 1) & h
3、根據該位置處結點性質進行相應查找

• 如果該位置為 null，那么直接返回 null 就可以了
• 如果該位置處的節點剛好就是我們需要的，返回該節點的值即可
• 如果該位置節點的 hash 值小于 0，說明正在擴容，或者是紅黑樹，后面我們再介紹 find 方法
• 如果以上 3 條都不滿足，那就是鏈表，進行遍歷比對即可

public V get(Object key) {Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;int h = spread(key.hashCode());if ((tab = table) != null && (n = tab.length) > 0 &&(e = tabAt(tab, (n - 1) & h)) != null) {// 判斷頭結點是否就是我們需要的節點if ((eh = e.hash) == h) {if ((ek = e.key) == key || (ek != null && key.equals(ek)))return e.val;}// 如果頭結點的 hash 小于 0，說明 正在擴容，或者該位置是紅黑樹else if (eh < 0)// 參考 ForwardingNode.find(int h, Object k) 和 TreeBin.find(int h, Object k)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; }

簡單說一句，此方法的大部分內容都很簡單，只有正好碰到擴容的情況，ForwardingNode.find(int h, Object k) 稍微復雜一些，不過在了解了數據遷移的過程后，這個也就不難了，所以限于篇幅這里也不展開說了。
###總結

其實也不是很難嘛，雖然沒有像之前的 AQS 和線程池一樣一行一行源碼進行分析，但還是把所有初學者可能會糊涂的地方都進行了深入的介紹，只要是稍微有點基礎的讀者，應該是很容易就能看懂 HashMap 和 ConcurrentHashMap 源碼了。

看源碼不算是目的吧，深入地了解 Doug Lea 的設計思路，我覺得還挺有趣的，大師就是大師，代碼寫得真的是好啊。

我發現很多人都以為我寫博客主要是源碼分析，說真的，我對于源碼分析沒有那么大熱情，主要都是為了用源碼說事罷了，可能之后的文章還是會有比較多的源碼分析成分，大家該怎么看就怎么看吧。

出處：https://www.javadoop.com/post/hashmap#Java7%20ConcurrentHashMap

點個贊，看一看，好習慣！本文 GitHub https://github.com/OUYANGSIHAI/JavaInterview 已收錄，這是我花了 3 個月總結的一線大廠 Java 面試總結，本人已拿大廠 offer。
另外，原創文章首發在我的個人博客：blog.ouyangsihai.cn，歡迎訪問。

最后，再分享我歷時三個月總結的 Java 面試 + Java 后端技術學習指南，這是本人這幾年及春招的總結，已經拿到了大廠 offer，整理成了一本電子書，拿去不謝，目錄如下：

現在免費分享大家，在下面我的公眾號 程序員的技術圈子 回復 面試 即可獲取。

有收獲？希望老鐵們來個三連擊，給更多的人看到這篇文章

1、老鐵們，關注我的原創微信公眾號「程序員的技術圈子」，專注于 Java、數據結構和算法、微服務、中間件等技術分享，保證你看完有所收獲。

2、給俺點個贊唄，可以讓更多的人看到這篇文章，順便激勵下我繼續寫作，嘻嘻。

3、另外，原創文章首發在我的個人博客：blog.ouyangsihai.cn，歡迎訪問。

點贊是對我最大的鼓勵
↓↓↓↓↓↓

總結

以上是生活随笔為你收集整理的Java8 ConcurrentHashMap详解的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。