<em>Mac</em>Book项目 2009年学校开始实施<em>Mac</em>Book项目,所有师生配备一本<em>Mac</em>Book,并同步更新了校园无线网络。学校每周进行电脑技术更新,每月发送技术支持资料,极大改变了教学及学习方式。因此2011
2021-06-01 09:32:01
ConcurrentHashMap是如何實現執行緒安全的你知道嗎
2022-10-14 14:00:49
1. 前言
我們知道，在日常開發中使用的 HashMap 是執行緒不安全的，而執行緒安全類 HashTable 和 SynchronizedMap 只是簡單的在方法上加鎖實現了執行緒安全，效率低下，所以線上程安全的環境下我們通常會使用 ConcurrentHashMap，那麼 ConcurrentHashMap 又是如何實現執行緒安全的呢?
2. ConcurrentHashMap 是如何實現執行緒安全的
針對這個問題，可以從以下幾個方面來閱讀原始碼予以解答
2.1. 初始化資料結構時的執行緒安全
在 JDK 1.8 中，初始化 ConcurrentHashMap 的時候這個 Node[] 陣列是還未初始化的，會等到第一次 put() 方法呼叫時才初始化
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode());
int binCount = 0;
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
// 判斷Node陣列為空
if (tab == null || (n = tab.length) == 0)
// 初始化Node陣列
tab = initTable();
......
}此時會有並行問題的，如果多個執行緒同時呼叫 initTable() 初始化 Node[] 陣列怎麼辦？
看看 Doug Lea 大師是如何處理的
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
// 每次迴圈都獲取最新的Node[]陣列參照
while ((tab = table) == null || tab.length == 0) {
// sizeCtl是一個標記位，若為-1，代表有執行緒在進行初始化工作了
if ((sc = sizeCtl) < 0)
// 讓出CPU時間片
Thread.yield();
// 此時，代表沒有執行緒在進行初始化工作，CAS操作，將本範例的sizeCtl變數設定為-1
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
// 如果CAS操作成功了，代表本執行緒將負責初始化工作
try {
// 再檢查一遍陣列是否為空
if ((tab = table) == null || tab.length == 0) {
// 在初始化ConcurrentHashMap時，sizeCtl代表陣列大小，預設16
// 所以此時n預設為16
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
// 將其賦值給table變數
table = tab = nt;
// 通過位運算，n減去n二進位制右移2位，相當於乘以0.75
// 例如16經過運算為12，與乘0.75一樣，只不過位運算更快
sc = n - (n >>> 2);
}
} finally {
// 將計算後的sc（12）直接賦值給sizeCtl，表示達到12長度就擴容
// 由於這裡只會有一個執行緒在執行，直接賦值即可，沒有執行緒安全問題，只需要保證可見性
sizeCtl = sc;
}
break;
}
}
return tab;
}
table 變數使用了 volatile 來保證每次獲取到的都是最新寫入的值
transient volatile Node<K,V>[] table;
ConcurrentHashMap 原始碼中 sizeCtl 變數註釋如下
// 表初始化和調整控制元件大小。如果為負值，則表正在初始化或調整大小：-1用於初始化，否則-（1+活動調整大小執行緒的數量） // 否則，當table為null時，將保留建立時使用的初始表大小，預設值為0。初始化後，儲存下一個要調整表大小的元素計數值 private transient volatile int sizeCtl;
在 ConcurrentHashMap 初始化時，初始化 sizeCtl
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;
}
public ConcurrentHashMap(Map<? extends K, ? extends V> m) {
this.sizeCtl = DEFAULT_CAPACITY;
putAll(m);
}
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {
if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
if (initialCapacity < concurrencyLevel) // Use at least as many bins
initialCapacity = concurrencyLevel; // as estimated threads
long size = (long)(1.0 + (long)initialCapacity / loadFactor);
int cap = (size >= (long)MAXIMUM_CAPACITY) ?
MAXIMUM_CAPACITY : tableSizeFor((int)size);
this.sizeCtl = cap;
}2.1.1. 總結
就算有多個執行緒同時進行 put 操作，在初始化 Node[] 陣列時，使用了 CAS 操作來決定到底是哪個執行緒有資格進行初始化，其他執行緒只能等待。
用到的並行技巧如下：
- volatile 修飾 sizeCtl 變數：它是一個標記位，用來告訴其他執行緒這個坑位有沒有執行緒在進行初始化工作，其執行緒間的可見性由 volatile 保證
- CAS 操作：CAS 操作保證了設定 sizeCtl 標記位的原子性，保證了在多執行緒同時進行初始化 Node[] 陣列時，只有一個執行緒能成功
2.2. put 操作時的執行緒安全
public V put(K key, V value) {
return putVal(key, value, false);
}
final V putVal(K key, V value, boolean onlyIfAbsent) {
// K,V 都不能為空
if (key == null || value == null) throw new NullPointerException();
// 取得 key 的 hash 值
int hash = spread(key.hashCode());
// 用來計算在這個節點總共有多少個元素，用來控制擴容或者轉換為樹
int binCount = 0;
// 陣列的遍歷，自旋插入結點，直到成功
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
// 當Node[]陣列為空時，進行初始化
if (tab == null || (n = tab.length) == 0)
tab = initTable();
// Unsafe類volatile的方式取出hashCode雜湊後通過與運算得出的Node[]陣列下標值對應的Node物件
// 此時 Node 位置若為 null，則表示還沒有執行緒在此 Node 位置進行插入操作，說明本次操作是第一次
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
// 如果這個位置沒有元素的話，則通過 CAS 的方式插入資料
if (casTabAt(tab, i, null,
// 建立一個 Node 新增到陣列中，null 表示的是下一個節點為空
new Node<K,V>(hash, key, value, null)))
// 插入成功，退出迴圈
break;
}
// 如果檢測到某個節點的 hash 值是 MOVED，則表示正在進行陣列擴容
else if ((fh = f.hash) == MOVED)
// 幫助擴容
tab = helpTransfer(tab, f);
// 此時，說明已經有執行緒對Node[]進行了插入操作，後面的插入很有可能會發生Hash衝突
else {
V oldVal = null;
// ----------------synchronized----------------
synchronized (f) {
// 二次確認此Node物件還是原來的那一個
if (tabAt(tab, i) == f) {
// ----------------table[i]是連結串列結點----------------
if (fh >= 0) {
// 記錄結點數，超過閾值後，需要轉為紅黑樹，提高查詢效率
binCount = 1;
// 遍歷這個連結串列
for (Node<K,V> e = f;; ++binCount) {
K ek;
// 要存的元素的 hash 值和 key 跟要儲存的位置的節點的相同的時候，替換掉該節點的 value 即可
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;
// 如果不是同樣的 hash，同樣的 key 的時候，則判斷該節點的下一個節點是否為空
if ((e = e.next) == null) {
// ----------------「尾插法」插入新結點----------------
pred.next = new Node<K,V>(hash, key,
value, null);
break;
}
}
}
// ----------------table[i]是紅黑樹結點----------------
else if (f instanceof TreeBin) {
Node<K,V> p;
binCount = 2;
// 呼叫putTreeVal方法，將該元素新增到樹中去
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
if (binCount != 0) {
// 當在同一個節點的數目達到8個的時候，則擴張陣列或將給節點的資料轉為tree
if (binCount >= TREEIFY_THRESHOLD)
// 連結串列 -> 紅黑樹 轉換
treeifyBin(tab, i);
// 表明本次put操作只是替換了舊值，不用更改計數值
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);// 計數值加1
return null;
}
值得關注的是 tabAt(tab, i) 方法，其使用 Unsafe 類 volatile 的操作 volatile 式地檢視值，保證每次獲取到的值都是 最新 的
static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {
return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
}雖然上面的 table 變數加了 volatile，但也只能保證其參照的可見性，並不能確保其陣列中的物件是否是最新的，所以需要 Unsafe 類 volatile 式地拿到最新的 Node
2.2.1. 總結
put() 方法的核心思想：由於其減小了鎖的粒度，若 Hash 完美不衝突的情況下，可同時支援 n 個執行緒同時 put 操作，n 為 Node 陣列大小，在預設大小 16 下，可以支援最大同時 16 個執行緒無競爭同時操作且執行緒安全
當 Hash 衝突嚴重時，Node 連結串列越來越長，將導致嚴重的鎖競爭，此時會進行擴容，將 Node 進行再雜湊，下面會介紹擴容的執行緒安全性。
總結一下用到的並行技巧
- 減小鎖粒度：將 Node 連結串列的頭節點作為鎖，若在預設大小 16 情況下，將有 16 把鎖，大大減小了鎖競爭（上下文切換），就像開頭所說，將序列的部分最大化縮小，在理想情況下執行緒的 put 操作都為並行操作。同時直接鎖住頭節點，保證了執行緒安全
- 使用了 volatile 修飾 table 變數，並使用 Unsafe 的 getObjectVolatile() 方法拿到最新的 Node
- CAS 操作：如果上述拿到的最新的 Node 為 null，則說明還沒有任何執行緒在此 Node 位置進行插入操作，說明本次操作是第一次
- synchronized 同步鎖：如果此時拿到的最新的 Node 不為 null，則說明已經有執行緒在此 Node 位置進行了插入操作，此時就產生了 hash 衝突；此時的 synchronized 同步鎖就起到了關鍵作用，防止在多執行緒的情況下發生資料覆蓋（執行緒不安全），接著在 synchronized 同步鎖的管理下按照相應的規則執行操作
當 hash 值相同並 key 值也相同時，則替換掉原 value
否則，將資料插入連結串列或紅黑樹相應的節點
2.3. get 操作時的執行緒安全
對於 get 操作其實沒有執行緒安全的問題，只有可見性的問題，只需要確保 get 的資料是執行緒之間可見的即可
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
// 重新計算key的hash值
int h = spread(key.hashCode());
// table不能為null，且table[i]不能為空
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
// 檢查頭結點，table[i]就是待查詢的項，直接返回
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
// hash值<0, 說明遇到特殊結點(非連結串列結點), 呼叫find()方法查詢
else if (eh < 0)
// 呼叫節點物件的find方法查詢值
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 操作中除了增加了遷移的判斷以外，基本與 HashMap 的 get 操作無異，這裡不多贅述，值得一提的是這裡使用了 tabAt() 方法 Unsafe 類 volatile 的方式去獲取 Node[] 陣列中的 Node，保證獲得到的 Node 是最新的
static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {
return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
}2.4. 擴容操作時的執行緒安全
在擴容時，ConcurrentHashMap 支援多執行緒並行擴容，在擴容過程中同時支援 get 查資料，若有執行緒 put 資料，還會幫助一起擴容，這種無阻塞演演算法，將並行最大化的設計，堪稱一絕
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
// stride可理解成「步長」，即「資料遷移」時，每個執行緒要負責舊table中的多少個桶，根據幾核的CPU決定「步長」
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
// 本執行緒分到的遷移量，假設為16（預設也為16）
stride = MIN_TRANSFER_STRIDE;
// 說明第一次擴容
if (nextTab == null) {
try {
@SuppressWarnings("unchecked")
// 建立新table陣列，擴大一倍為 32,n還為16
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
nextTab = nt;
} catch (Throwable ex) {
// 將表示容量的sizeCtl 設定為最大值，然後返回
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;
// 表示當前執行緒要進行資料遷移的桶區間
transferIndex = n;
}
int nextn = nextTab.length;
// 在get或者put時若遇到此 Node，則可以知道當前Node正在遷移
// ForwardingNode結點，當舊table的某個桶中的所有結點都遷移完後，用該結點佔據這個桶
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
// 標識一個桶的遷移工作是否完成，advance == true 表示可以進行下一個位置的遷移
boolean advance = true;
// 最後一個資料遷移的執行緒將該值置為true，並進行本輪擴容的收尾工作
boolean finishing = false;
// i標識桶索引, bound標識邊界
for (int i = 0, bound = 0;;) {
Node<K,V> 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;
}
}
// 當前是處理最後一個tranfer任務的執行緒或出現擴容衝突
if (i < 0 || i >= n || i + n >= nextn) {
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;
}
}
// 舊桶本身為null，不用遷移，直接嘗試放一個ForwardingNode
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
// 該舊桶已經遷移完成，直接跳過
else if ((fh = f.hash) == MOVED)
advance = true;
// 該舊桶未遷移完成，進行資料遷移
else {
synchronized (f) {
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;
// 桶的hash>0，說明是連結串列遷移
if (fh >= 0) {
/**
* 下面的過程會將舊桶中的連結串列分成兩部分：ln鏈和hn鏈
* ln鏈會插入到新table的槽i中，hn鏈會插入到新table的槽i+n中
*/
int runBit = fh & n;
// lastRun指向最後一個相鄰runBit不同的結點
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;
}
// 以lastRun所指向的結點為分界，將連結串列拆成2個子連結串列ln、hn
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); // ln連結串列存入新桶的索引i位置
setTabAt(nextTab, i + n, hn); // hn連結串列存入新桶的索引i+n位置
setTabAt(tab, i, fwd); // 設定ForwardingNode佔位
advance = true; // 表示當前舊桶的結點已遷移完畢
}
else if (f instanceof TreeBin) {
/**
* 下面的過程會先以連結串列方式遍歷，複製所有結點，然後根據高低位組裝成兩個連結串列；
* 然後看下是否需要進行紅黑樹轉換，最後放到新table對應的桶中
*/
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); // 設定ForwardingNode佔位
advance = true; // 表示當前舊桶的結點已遷移完畢
}
}
}
}
}
}
2.4.1. 擴容時的 get 操作
假設 Node下標為 16 的 Node 節點正在遷移擴容，突然有一個執行緒進來呼叫 get() 方法，正好 key 又雜湊到下標為 16 的節點，此時怎麼辦？
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
// 重新計算key的hash值
int h = spread(key.hashCode());
// table不能為null，且table[i]不能為空
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
// 檢查頭結點，table[i]就是待查詢的項，直接返回
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
// hash值<0, 說明遇到特殊結點(非連結串列結點), 呼叫find()方法查詢
else if (eh < 0)
// 呼叫節點物件的find方法查詢值
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() 操作的原始碼中，會判斷 Node 中的 hash 是否小於 0（eh < 0），是否還記得我們的佔位 Node，其 hash 為 MOVED，為常數值 -1，所以此時判斷執行緒正在遷移，委託給內部類 ForwardingNode 佔位 Node 去查詢值
// //內部類 ForwardingNode 中 find() 方法
Node<K,V> find(int h, Object k) {
// 這裡的查詢，是去新Node陣列中查詢的
outer: for (Node<K,V>[] tab = nextTable;;) {
Node<K,V> e; int n;
if (k == null || tab == null || (n = tab.length) == 0 ||
(e = tabAt(tab, (n - 1) & h)) == null)
return null;
for (;;) {
int eh; K ek;
if ((eh = e.hash) == h &&
((ek = e.key) == k || (ek != null && k.equals(ek))))
return e;
if (eh < 0) {
if (e instanceof ForwardingNode) {
tab = ((ForwardingNode<K,V>)e).nextTable;
continue outer;
}
else
return e.find(h, k);
}
if ((e = e.next) == null)
return null;
}
}
}
到這裡之所以佔位 Node 需要儲存新 Node[] 陣列的參照也是因為這個，它可以支援在遷移的過程中照樣不阻塞地查詢值，可謂是精妙絕倫的設計
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
2.4.2. 多執行緒協助擴容
在 put 操作時，假設正在遷移擴容，正好有一個執行緒進來，想要 put 值到遷移的 Node上，怎麼辦？
在 put() 方法中呼叫了 helpTransfer() 方法
// put() 方法中的程式碼片段，幫助擴容
tab = helpTransfer(tab, f);
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
Node<K,V>[] nextTab; int sc;
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;
// sizeCtl加 1，表示多一個執行緒進來協助擴容
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
transfer(tab, nextTab);
break;
}
}
return nextTab;
}
return table;
}
此方法涉及大量複雜的位運算，這裡只是簡單的說幾句，此時 sizeCtl 變數用來表示 ConcurrentHashMap 正在擴容，當其準備擴容時，會將 sizeCtl 設定為一個負數
2.4.3. 總結
ConcurrentHashMap 運用各類 CAS 操作，將擴容操作的並行效能實現最大化，在擴容過程中，
- 就算有執行緒呼叫 get 查詢方法，也可以安全的查詢資料
- 若有執行緒進行 put 操作，還會協助擴容
利用 sizeCtl 標記位和各種 volatile 變數進行 CAS 操作達到多執行緒之間的通訊、協助，在遷移擴容過程中只鎖一個 Node 節點，即保證了執行緒安全，又提高了並行效能
3. 什麼情況下 ConcurrentHashMap 會進行擴容操作
在 put 值時，發現 Node 為佔位 Node（ForwardingNode）時，會協助擴容
// 在 put() 方法中的程式碼片段 else if ((fh = f.hash) == MOVED) tab = helpTransfer(tab, f);
在 put 值時，檢測到單連結串列長度大於 8 時
final V putVal(K key, V value, boolean onlyIfAbsent) {
......
if (binCount != 0) {
// TREEIFY_THRESHOLD=8，當連結串列長度大於8時
if (binCount >= TREEIFY_THRESHOLD)
// 呼叫treeifyBin方法
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
......
}treeifyBin() 方法會將單連結串列轉換為紅黑樹，增加查詢效率，但在這之前，會檢查陣列長度，若小於 64，則會優先做擴容操作
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，則先擴容
if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
// 擴容
tryPresize(n << 1);
else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
synchronized (b) {
// 轉換為紅黑樹的操作
......
}
}
}
}在每次 put 值之後，都會呼叫 addCount() 方法，檢測 Node[] 陣列大小是否達到閾值
private final void addCount(long x, int check) {
CounterCell[] as; long b, s;
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))) {
fullAddCount(x, uncontended);
return;
}
if (check <= 1)
return;
s = sumCount();
}
if (check >= 0) {
Node<K,V>[] tab, nt; int n, sc;
// 元素個數達到閾值，進行擴容
while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
(n = tab.length) < MAXIMUM_CAPACITY) {
int rs = resizeStamp(n);
// 發現sizeCtl為負數，證明有執行緒正在遷移擴容
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);
}
// 不為負數，則為第一個遷移的執行緒
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
transfer(tab, null);
s = sumCount();
}
}
}
以上為個人經驗，希望能給大家一個參考，也希望大家多多支援it145.com。
相關文章
-
黑科技的輕簡出行，告別「電量焦慮症」：Anker氮化鎵超能充系列
综合看Anker超能充系列的性价比很高，并且与不仅和iPhone12/苹果<em>Mac</em>Book很配，而且适合多设备充电需求的日常使用或差旅场景，不管是安卓还是Switch同样也能用得上它，希望这次分享能给准备购入充电器的小伙伴们有所
2021-06-01 09:31:42
-
吳亦凡廠牌首秀，L4WUDU居然忘詞了，Rapper和明星比還是有差距！
除了L4WUDU与吴亦凡已经多次共事，成为了明面上的厂牌成员，吴亦凡还曾带领20XXCLUB全队参加2020年的一场音乐节，这也是20XXCLUB首次全员合照，王嗣尧Turbo、陈彦希Regi、<em>Mac</em> Ova Seas、林渝植等人全部出场。然而让
2021-06-01 09:31:34
-
IPFS、Chia、Bzz和ICP挖礦該怎麼選？哪一個更有優勢？
目前应用IPFS的机构：1 谷歌<em>浏览器</em>支持IPFS分布式协议 2 万维网 (历史档案博物馆)数据库 3 火狐<em>浏览器</em>支持 IPFS分布式协议 4 EOS 等数字货币数据存储 5 美国国会图书馆，历史资料永久保存在 IPFS 6 加
2021-06-01 09:31:24
-
有哪些事是買了雪佛蘭後才知道的？美事偷著樂，開拓者車主隨聊
开拓者的车机是兼容苹果和<em>安卓</em>，虽然我不怎么用，但确实兼顾了我家人的很多需求：副驾的门板还配有解锁开关，有的时候老婆开车，下车的时候偶尔会忘记解锁，我在副驾驶可以自己开门：第二排设计很好，不仅配置了一个很大的
2021-06-01 09:30:48
-
iPhone12在618最新定價，跌價幅度超過1400元，還等iPhone13嗎？
不仅是<em>安卓</em>手机，苹果手机的降价力度也是前所未有了，iPhone12也“跳水价”了，发布价是6799元，如今已经跌至5308元，降价幅度超过1400元，最新定价确认了。iPhone12是苹果首款5G手机，同时也是全球首款5nm芯片的智能机，它
2021-06-01 09:30:45