摘要:一擴容的基本思路中,最復雜的部分就是擴容數據遷移,涉及多線程的合作和。單線程注意這兩種情況都是調用了方法,通過第二個入參進行區分表示擴容后的新數組,如果為,表示首次發起擴容。第二種情況下,是通過和移位運算來保證僅有一個線程能發起擴容。
本文首發于一世流云專欄:https://segmentfault.com/blog...
通過上一篇文章——ConcurrentHashMap原理(1),相信讀者對ConcurrentHashMap的基本原理有了一個初步認識,但是上一篇中還有一個遺留問題沒有討論到,那就是ConcurrentHashMap的擴容和數據遷移。本文中,我們將會對這兩個問題進行討論。
一、擴容的基本思路JDK1.8中,ConcurrentHashMap最復雜的部分就是擴容/數據遷移,涉及多線程的合作和rehash。我們先來考慮下一般情況下,如何對一個Hash表進行擴容。
擴容思路Hash表的擴容,一般都包含兩個步驟:
①table數組的擴容
table數組的擴容,一般就是新建一個2倍大小的槽數組,這個過程通過由一個單線程完成,且不允許出現并發。
②數據遷移
所謂數據遷移,就是把舊table中的各個槽中的結點重新分配到新table中。比如,單線程情況下,可以遍歷原來的table,然后put到新table中。
這一過程通常涉及到槽中key的rehash,因為key映射到桶的位置與table的大小有關,新table的大小變了,key映射的位置一般也會變化。
ConcurrentHashMap在處理rehash的時候,并不會重新計算每個key的hash值,而是利用了一種很巧妙的方法。我們在上一篇說過,ConcurrentHashMap內部的table數組的大小必須為2的冪次,原因是讓key均勻分布,減少沖突,這只是其中一個原因。另一個原因就是:
當table數組的大小為2的冪次時,通過key.hash & table.length-1這種方式計算出的索引i,當table擴容后(2倍),新的索引要么在原來的位置i,要么是i+n。
我們來看個例子:
上圖中:
擴容前,table數組大小為16,key1和key2映射到同一個索引5;
擴容后,table數組的大小變成 2*16=32 ,key1的索引不變,key2的索引變成 5+16=21。
而且還有一個特點,擴容后key對應的索引如果發生了變化,那么其變化后的索引最高位一定是1(見擴容后key2的最高位)。
這種處理方式非常利于擴容時多個線程同時進行的數據遷移操作,因為舊table的各個桶中的結點遷移不會互相影響,所以就可以用“分治”的方式,將整個table數組劃分為很多部分,每一部分包含一定區間的桶,每個數據遷移線程處理各自區間中的結點,對多線程同時進行數據遷移非常有利,后面我們會詳細介紹。擴容時機
我們再來看下,ConcurrentHashMap何時會發生擴容。
在上篇文章中,我們提到過,當往Map中插入結點時,如果鏈表的結點數目超過一定閾值,就會觸發鏈表 -> 紅黑樹的轉換:
if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); // 鏈表 -> 紅黑樹 轉換
現在,我們來分析下treeifyBin這個紅黑樹化的操作:
/** * 嘗試進行 鏈表 -> 紅黑樹 的轉換. */ private final void treeifyBin(Node[] tab, int index) { Node b; int n, sc; if (tab != null) { // CASE 1: table的容量 < MIN_TREEIFY_CAPACITY(64)時,直接進行table擴容,不進行紅黑樹轉換 if ((n = tab.length) < MIN_TREEIFY_CAPACITY) tryPresize(n << 1); // CASE 2: table的容量 ≥ MIN_TREEIFY_CAPACITY(64)時,進行鏈表 -> 紅黑樹的轉換 else if ((b = tabAt(tab, index)) != null && b.hash >= 0) { synchronized (b) { if (tabAt(tab, index) == b) { TreeNode hd = null, tl = null; // 遍歷鏈表,建立紅黑樹 for (Node e = b; e != null; e = e.next) { TreeNode p = new TreeNode (e.hash, e.key, e.val, null, null); if ((p.prev = tl) == null) hd = p; else tl.next = p; tl = p; } // 以TreeBin類型包裝,并鏈接到table[index]中 setTabAt(tab, index, new TreeBin (hd)); } } } } }
上述第一個分支中,還會再對table數組的長度進行一次判斷:
如果table長度小于閾值MIN_TREEIFY_CAPACITY——默認64,則會調用tryPresize方法把數組長度擴大到原來的兩倍。
從代碼也可以看到,鏈表 -> 紅黑樹這一轉換并不是一定會進行的,table長度較小時,CurrentHashMap會首先選擇擴容,而非立即轉換成紅黑樹。
來看下tryPresize方法如何執行擴容:
/** * 嘗試對table數組進行擴容. * * @param 待擴容的大小 */ private final void tryPresize(int size) { // 視情況將size調整為2的冪次 int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY : tableSizeFor(size + (size >>> 1) + 1); int sc; while ((sc = sizeCtl) >= 0) { Node[] tab = table; int n; //CASE 1: table還未初始化,則先進行初始化 if (tab == null || (n = tab.length) == 0) { n = (sc > c) ? sc : c; if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { try { if (table == tab) { Node [] nt = (Node []) new Node, ?>[n]; table = nt; sc = n - (n >>> 2); } } finally { sizeCtl = sc; } } } // CASE2: c <= sc說明已經被擴容過了;n >= MAXIMUM_CAPACITY說明table數組已達到最大容量 else if (c <= sc || n >= MAXIMUM_CAPACITY) break; // CASE3: 進行table擴容 else if (tab == table) { int rs = resizeStamp(n); // 根據容量n生成一個隨機數,唯一標識本次擴容操作 if (sc < 0) { // sc < 0 表明此時有別的線程正在進行擴容 Node [] nt; // 如果當前線程無法協助進行數據轉移, 則退出 if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || sc == rs + MAX_RESIZERS || (nt = nextTable) == null || transferIndex <= 0) break; // 協助數據轉移, 把正在執行transfer任務的線程數加1 if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) transfer(tab, nt); } // sc置為負數, 當前線程自身成為第一個執行transfer(數據轉移)的線程 // 這個CAS操作可以保證,僅有一個線程會執行擴容 else if (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2)) transfer(tab, null); } } }
前兩個分支沒什么好說的,看下注釋很容易理解,關鍵看第三個分支 —— CASE3:進行table擴容。
CASE3其實分為兩種情況:
已經有其它線程正在執行擴容了,則當前線程會嘗試協助“數據遷移”;(多線程并發)
沒有其它線程正在執行擴容,則當前線程自身發起擴容。(單線程)
注意:這兩種情況都是調用了transfer方法,通過第二個入參nextTab進行區分(nextTab表示擴容后的新table數組,如果為null,表示首次發起擴容)。二、擴容的原理
第二種情況下,是通過CAS和移位運算來保證僅有一個線程能發起擴容。
我們來看下transfer方法,這個方法可以被多個線程同時調用,也是“數據遷移”的核心操作方法:
/** * 數據轉移和擴容. * 每個調用tranfer的線程會對當前舊table中[transferIndex-stride, transferIndex-1]位置的結點進行遷移 * * @param tab 舊table數組 * @param nextTab 新table數組 */ private final void transfer(Node[] tab, Node [] nextTab) { int n = tab.length, stride; // stride可理解成“步長”,即數據遷移時,每個線程要負責舊table中的多少個桶 if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE) stride = MIN_TRANSFER_STRIDE; if (nextTab == null) { // 首次擴容 try { // 創建新table數組 Node [] nt = (Node []) new Node, ?>[n << 1]; nextTab = nt; } catch (Throwable ex) { // 處理內存溢出(OOME)的情況 sizeCtl = Integer.MAX_VALUE; return; } nextTable = nextTab; transferIndex = n; // [transferIndex-stride, transferIndex-1]表示當前線程要進行數據遷移的桶區間 } int nextn = nextTab.length; // ForwardingNode結點,當舊table的某個桶中的所有結點都遷移完后,用該結點占據這個桶 ForwardingNode fwd = new ForwardingNode (nextTab); // 標識一個桶的遷移工作是否完成,advance == true 表示可以進行下一個位置的遷移 boolean advance = true; // 最后一個數據遷移的線程將該值置為true,并進行本輪擴容的收尾工作 boolean finishing = false; // i標識桶索引, bound標識邊界 for (int i = 0, bound = 0; ; ) { Node f; int fh; // 每一次自旋前的預處理,主要是定位本輪處理的桶區間 // 正常情況下,預處理完成后:i == transferIndex-1,bound == transferIndex-stride while (advance) { int nextIndex, nextBound; if (--i >= bound || finishing) advance = false; else if ((nextIndex = transferIndex) <= 0) { i = -1; advance = false; } 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) { // CASE1:當前是處理最后一個tranfer任務的線程或出現擴容沖突 int sc; if (finishing) { // 所有桶遷移均已完成 nextTable = null; table = nextTab; sizeCtl = (n << 1) - (n >>> 1); return; } // 擴容線程數減1,表示當前線程已完成自己的transfer任務 if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) { // 判斷當前線程是否是本輪擴容中的最后一個線程,如果不是,則直接退出 if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT) return; finishing = advance = true; /** * 最后一個數據遷移線程要重新檢查一次舊table中的所有桶,看是否都被正確遷移到新table了: * ①正常情況下,重新檢查時,舊table的所有桶都應該是ForwardingNode; * ②特殊情況下,比如擴容沖突(多個線程申請到了同一個transfer任務),此時當前線程領取的任務會作廢,那么最后檢查時, * 還要處理因為作廢而沒有被遷移的桶,把它們正確遷移到新table中 */ i = n; // recheck before commit } } else if ((f = tabAt(tab, i)) == null) // CASE2:舊桶本身為null,不用遷移,直接嘗試放一個ForwardingNode advance = casTabAt(tab, i, null, fwd); else if ((fh = f.hash) == MOVED) // CASE3:該舊桶已經遷移完成,直接跳過 advance = true; else { // CASE4:該舊桶未遷移完成,進行數據遷移 synchronized (f) { if (tabAt(tab, i) == f) { Node ln, hn; if (fh >= 0) { // CASE4.1:桶的hash>0,說明是鏈表遷移 /** * 下面的過程會將舊桶中的鏈表分成兩部分:ln鏈和hn鏈 * ln鏈會插入到新table的槽i中,hn鏈會插入到新table的槽i+n中 */ int runBit = fh & n; // 由于n是2的冪次,所以runBit要么是0,要么高位是1 Node lastRun = f; // lastRun指向最后一個相鄰runBit不同的結點 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; } // 以lastRun所指向的結點為分界,將鏈表拆成2個子鏈表ln、hn 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); } setTabAt(nextTab, i, ln); // ln鏈表存入新桶的索引i位置 setTabAt(nextTab, i + n, hn); // hn鏈表存入新桶的索引i+n位置 setTabAt(tab, i, fwd); // 設置ForwardingNode占位 advance = true; // 表示當前舊桶的結點已遷移完畢 } else if (f instanceof TreeBin) { // CASE4.2:紅黑樹遷移 /** * 下面的過程會先以鏈表方式遍歷,復制所有結點,然后根據高低位組裝成兩個鏈表; * 然后看下是否需要進行紅黑樹轉換,最后放到新table對應的桶中 */ 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; } } // 判斷是否需要進行 紅黑樹 <-> 鏈表 的轉換 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); // 設置ForwardingNode占位 advance = true; // 表示當前舊桶的結點已遷移完畢 } } } } } }
tranfer方法的開頭,會計算出一個stride變量的值,這個stride其實就是每個線程處理的桶區間,也就是步長:
// stride可理解成“步長”,即數據遷移時,每個線程要負責舊table中的多少個桶 if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE) stride = MIN_TRANSFER_STRIDE;
首次擴容時,會將table數組變成原來的2倍:
if (nextTab == null) { // 首次擴容 try { // 創建新table數組 Node[] nt = (Node []) new Node, ?>[n << 1]; nextTab = nt; } catch (Throwable ex) { // 處理內存溢出(OOME)的情況 sizeCtl = Integer.MAX_VALUE; return; } nextTable = nextTab; transferIndex = n; // [transferIndex-stride, transferIndex-1]表示當前線程要進行數據遷移的桶區間 }
注意上面的transferIndex變量,這是一個字段,table[transferIndex-stride, transferIndex-1]就是當前線程要進行數據遷移的桶區間:
/** * 擴容時需要用到的一個下標變量. */ private transient volatile int transferIndex;
整個transfer方法幾乎都在一個自旋操作中完成,從右往左開始進行數據遷移,transfer的退出點是當某個線程處理完最后的table區段——table[0,stride-1]。
transfer方法主要包含4個分支,即對4種不同情況進行處理,我們按照難易程度來解釋下各個分支所做的事情:
CASE2:桶table[i]為空當舊table的桶table[i] == null,說明原來這個桶就沒有數據,那就直接嘗試放置一個ForwardingNode,表示這個桶已經處理完成。
else if ((f = tabAt(tab, i)) == null) // CASE2:舊桶本身為null,不用遷移,直接嘗試放一個ForwardingNode advance = casTabAt(tab, i, null, fwd);
注:ForwardingNode我們在上一篇提到過,主要做占用位,多線程進行數據遷移時,其它線程看到這個桶中是ForwardingNode結點,就知道有線程已經在數據遷移了。CASE3:桶table[i]已遷移完成
另外,當最后一個線程完成遷移任務后,會遍歷所有桶,看看是否都是ForwardingNode,如果是,那么說明整個擴容/數據遷移的過程就完成了。
沒什么好說的,就是桶已經用ForwardingNode結點占用了,表示該桶的數據都遷移完了。
else if ((fh = f.hash) == MOVED) // CASE3:該舊桶已經遷移完成,直接跳過 advance = true;CASE4:桶table[i]未遷移完成
如果舊桶的數據未遷移完成,就要進行遷移,這里根據桶中結點的類型分為:鏈表遷移、紅黑樹遷移。
①鏈表遷移
鏈表遷移的過程如下,首先會遍歷一遍原鏈表,找到最后一個相鄰runBit不同的結點。
runbit是根據key.hash和舊table長度n進行與運算得到的值,由于table的長度為2的冪次,所以runbit只可能為0或最高位為1
然后,會進行第二次鏈表遍歷,按照第一次遍歷找到的結點為界,將原鏈表分成2個子鏈表,再鏈接到新table的槽中。可以看到,新table的索引要么是i,要么是i+n,這里就利用了上一節說的ConcurrentHashMap的rehash特點。
if (fh >= 0) { // CASE4.1:桶的hash>0,說明是鏈表遷移 /** * 下面的過程會將舊桶中的鏈表分成兩部分:ln鏈和hn鏈 * ln鏈會插入到新table的槽i中,hn鏈會插入到新table的槽i+n中 */ int runBit = fh & n; // 由于n是2的冪次,所以runBit要么是0,要么高位是1 NodelastRun = f; // lastRun指向最后一個相鄰runBit不同的結點 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; } // 以lastRun所指向的結點為分界,將鏈表拆成2個子鏈表ln、hn 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); } setTabAt(nextTab, i, ln); // ln鏈表存入新桶的索引i位置 setTabAt(nextTab, i + n, hn); // hn鏈表存入新桶的索引i+n位置 setTabAt(tab, i, fwd); // 設置ForwardingNode占位 advance = true; // 表示當前舊桶的結點已遷移完畢 }
②紅黑樹遷移
紅黑樹的遷移按照鏈表遍歷的方式進行,當鏈表結點超過/小于閾值時,涉及紅黑樹<->鏈表的相互轉換:
else if (f instanceof TreeBin) { // CASE4.2:紅黑樹遷移 /** * 下面的過程會先以鏈表方式遍歷,復制所有結點,然后根據高低位組裝成兩個鏈表; * 然后看下是否需要進行紅黑樹轉換,最后放到新table對應的桶中 */ TreeBinCASE1:當前是最后一個遷移任務或出現擴容沖突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; } } // 判斷是否需要進行 紅黑樹 <-> 鏈表 的轉換 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); // 設置ForwardingNode占位 advance = true; // 表示當前舊桶的結點已遷移完畢 }
我們剛才說了,調用transfer的線程會自動領用某個區段的桶,進行數據遷移操作,當區段的初始索引i變成負數的時候,說明當前線程處理的其實就是最后剩下的桶,并且處理完了。
所以首先會更新sizeCtl變量,將擴容線程數減1,然后會做一些收尾工作:
設置table指向擴容后的新數組,遍歷一遍舊數組,確保每個桶的數據都遷移完成——被ForwardingNode占用。
另外,可能在擴容過程中,出現擴容沖突的情況,比如多個線程領用了同一區段的桶,這時任何一個線程都不能進行數據遷移。
if (i < 0 || i >= n || i + n >= nextn) { // CASE1:當前是處理最后一個tranfer任務的線程或出現擴容沖突 int sc; if (finishing) { // 所有桶遷移均已完成 nextTable = null; table = nextTab; sizeCtl = (n << 1) - (n >>> 1); return; } // 擴容線程數減1,表示當前線程已完成自己的transfer任務 if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) { // 判斷當前線程是否是本輪擴容中的最后一個線程,如果不是,則直接退出 if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT) return; finishing = advance = true; /** * 最后一個數據遷移線程要重新檢查一次舊table中的所有桶,看是否都被正確遷移到新table了: * ①正常情況下,重新檢查時,舊table的所有桶都應該是ForwardingNode; * ②特殊情況下,比如擴容沖突(多個線程申請到了同一個transfer任務),此時當前線程領取的任務會作廢,那么最后檢查時, * 還要處理因為作廢而沒有被遷移的桶,把它們正確遷移到新table中 */ i = n; // recheck before commit } }
文章版權歸作者所有,未經允許請勿轉載,若此文章存在違規行為,您可以聯系管理員刪除。
轉載請注明本文地址:http://specialneedsforspecialkids.com/yun/76838.html
摘要:需要注意的是所鏈接的是一顆紅黑樹,紅黑樹的結點用表示,所以中實際上一共有五種不同類型的結點。時不再延續,轉而直接對每個桶加鎖,并用紅黑樹鏈接沖突結點。 showImg(https://segmentfault.com/img/bVbfTCY?w=1920&h=1080); 本文首發于一世流云專欄:https://segmentfault.com/blog... 一、Concurren...
摘要:整個包,按照功能可以大致劃分如下鎖框架原子類框架同步器框架集合框架執行器框架本系列將按上述順序分析,分析所基于的源碼為。后,根據一系列常見的多線程設計模式,設計了并發包,其中包下提供了一系列基礎的鎖工具,用以對等進行補充增強。 showImg(https://segmentfault.com/img/remote/1460000016012623); 本文首發于一世流云專欄:https...
摘要:我們來看下的類繼承圖可以看到,實現了接口,在多線程進階二五之框架中,我們提到過實現了接口,以提供和排序相關的功能,維持元素的有序性,所以就是一種為并發環境設計的有序工具類。唯一的區別是針對的僅僅是鍵值,針對鍵值對進行操作。 showImg(https://segmentfault.com/img/bVbggic?w=960&h=600); 本文首發于一世流云專欄:https://seg...
摘要:僅僅當有多個線程同時進行寫操作時,才會進行同步。可以看到,上述方法返回一個迭代器對象,的迭代是在舊數組上進行的,當創建迭代器的那一刻就確定了,所以迭代過程中不會拋出并發修改異常。另外,迭代器對象也不支持修改方法,全部會拋出異常。 showImg(https://segmentfault.com/img/bVbggij?w=960&h=600); 本文首發于一世流云專欄:https://...
摘要:我們之前已經介紹過了,底層基于跳表實現,其操作平均時間復雜度均為。事實上,內部引用了一個對象,以組合方式,委托對象實現了所有功能。線程安全內存的使用較多迭代是對快照進行的,不會拋出,且迭代過程中不支持修改操作。 showImg(https://segmentfault.com/img/bVbggjf?w=600&h=377); 本文首發于一世流云專欄:https://segmentfa...
閱讀 3969·2021-11-23 10:09
閱讀 1338·2021-11-23 09:51
閱讀 2939·2021-11-23 09:51
閱讀 1585·2021-09-07 09:59
閱讀 2354·2019-08-30 15:55
閱讀 2292·2019-08-30 15:55
閱讀 2949·2019-08-30 15:52
閱讀 2560·2019-08-26 17:04