ConcurrentHashMap是如何實現(xiàn)線程安全的你知道嗎
更新時間：2022年10月12日 16:12:32 作者：桐花思雨
這篇文章主要介紹了ConcurrentHashMap是如何實現(xiàn)線程安全的你知道嗎，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教
1. 前言
我們知道，在日常開發(fā)中使用的 HashMap 是線程不安全的，而線程安全類 HashTable 和 SynchronizedMap 只是簡單的在方法上加鎖實現(xiàn)了線程安全，效率低下，所以在線程安全的環(huán)境下我們通常會使用 ConcurrentHashMap，那么 ConcurrentHashMap 又是如何實現(xiàn)線程安全的呢?
2. ConcurrentHashMap 是如何實現(xiàn)線程安全的
針對這個問題，可以從以下幾個方面來閱讀源碼予以解答
2.1. 初始化數(shù)據(jù)結(jié)構(gòu)時的線程安全
在 JDK 1.8 中，初始化 ConcurrentHashMap 的時候這個 Node[] 數(shù)組是還未初始化的，會等到第一次 put() 方法調(diào)用時才初始化
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數(shù)組為空
if (tab == null || (n = tab.length) == 0)
// 初始化Node數(shù)組
tab = initTable();
......
}此時會有并發(fā)問題的，如果多個線程同時調(diào)用 initTable() 初始化 Node[] 數(shù)組怎么辦？
看看 Doug Lea 大師是如何處理的
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
// 每次循環(huán)都獲取最新的Node[]數(shù)組引用
while ((tab = table) == null || tab.length == 0) {
// sizeCtl是一個標記位，若為-1，代表有線程在進行初始化工作了
if ((sc = sizeCtl) < 0)
// 讓出CPU時間片
Thread.yield();
// 此時，代表沒有線程在進行初始化工作，CAS操作，將本實例的sizeCtl變量設(shè)置為-1
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
// 如果CAS操作成功了，代表本線程將負責初始化工作
try {
// 再檢查一遍數(shù)組是否為空
if ((tab = table) == null || tab.length == 0) {
// 在初始化ConcurrentHashMap時，sizeCtl代表數(shù)組大小，默認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經(jīng)過運算為12，與乘0.75一樣，只不過位運算更快
sc = n - (n >>> 2);
}
} finally {
// 將計算后的sc（12）直接賦值給sizeCtl，表示達到12長度就擴容
// 由于這里只會有一個線程在執(zhí)行，直接賦值即可，沒有線程安全問題，只需要保證可見性
sizeCtl = sc;
}
break;
}
}
return tab;
}
table 變量使用了 volatile 來保證每次獲取到的都是最新寫入的值
transient volatile Node<K,V>[] table;
ConcurrentHashMap 源碼中 sizeCtl 變量注釋如下
// 表初始化和調(diào)整控件大小。如果為負值，則表正在初始化或調(diào)整大小：-1用于初始化，否則-（1+活動調(diào)整大小線程的數(shù)量） // 否則，當table為null時，將保留創(chuàng)建時使用的初始表大小，默認值為0。初始化后，保存下一個要調(diào)整表大小的元素計數(shù)值 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. 總結(jié)
就算有多個線程同時進行 put 操作，在初始化 Node[] 數(shù)組時，使用了 CAS 操作來決定到底是哪個線程有資格進行初始化，其他線程只能等待。
用到的并發(fā)技巧如下：
- volatile 修飾 sizeCtl 變量：它是一個標記位，用來告訴其他線程這個坑位有沒有線程在進行初始化工作，其線程間的可見性由 volatile 保證
- CAS 操作：CAS 操作保證了設(shè)置 sizeCtl 標記位的原子性，保證了在多線程同時進行初始化 Node[] 數(shù)組時，只有一個線程能成功
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());
// 用來計算在這個節(jié)點總共有多少個元素，用來控制擴容或者轉(zhuǎn)換為樹
int binCount = 0;
// 數(shù)組的遍歷，自旋插入結(jié)點，直到成功
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
// 當Node[]數(shù)組為空時，進行初始化
if (tab == null || (n = tab.length) == 0)
tab = initTable();
// Unsafe類volatile的方式取出hashCode散列后通過與運算得出的Node[]數(shù)組下標值對應(yīng)的Node對象
// 此時 Node 位置若為 null，則表示還沒有線程在此 Node 位置進行插入操作，說明本次操作是第一次
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
// 如果這個位置沒有元素的話，則通過 CAS 的方式插入數(shù)據(jù)
if (casTabAt(tab, i, null,
// 創(chuàng)建一個 Node 添加到數(shù)組中，null 表示的是下一個節(jié)點為空
new Node<K,V>(hash, key, value, null)))
// 插入成功，退出循環(huán)
break;
}
// 如果檢測到某個節(jié)點的 hash 值是 MOVED，則表示正在進行數(shù)組擴容
else if ((fh = f.hash) == MOVED)
// 幫助擴容
tab = helpTransfer(tab, f);
// 此時，說明已經(jīng)有線程對Node[]進行了插入操作，后面的插入很有可能會發(fā)生Hash沖突
else {
V oldVal = null;
// ----------------synchronized----------------
synchronized (f) {
// 二次確認此Node對象還是原來的那一個
if (tabAt(tab, i) == f) {
// ----------------table[i]是鏈表結(jié)點----------------
if (fh >= 0) {
// 記錄結(jié)點數(shù)，超過閾值后，需要轉(zhuǎn)為紅黑樹，提高查找效率
binCount = 1;
// 遍歷這個鏈表
for (Node<K,V> e = f;; ++binCount) {
K ek;
// 要存的元素的 hash 值和 key 跟要存儲的位置的節(jié)點的相同的時候，替換掉該節(jié)點的 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 的時候，則判斷該節(jié)點的下一個節(jié)點是否為空
if ((e = e.next) == null) {
// ----------------“尾插法”插入新結(jié)點----------------
pred.next = new Node<K,V>(hash, key,
value, null);
break;
}
}
}
// ----------------table[i]是紅黑樹結(jié)點----------------
else if (f instanceof TreeBin) {
Node<K,V> p;
binCount = 2;
// 調(diào)用putTreeVal方法，將該元素添加到樹中去
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
if (binCount != 0) {
// 當在同一個節(jié)點的數(shù)目達到8個的時候，則擴張數(shù)組或?qū)⒔o節(jié)點的數(shù)據(jù)轉(zhuǎn)為tree
if (binCount >= TREEIFY_THRESHOLD)
// 鏈表 -> 紅黑樹 轉(zhuǎn)換
treeifyBin(tab, i);
// 表明本次put操作只是替換了舊值，不用更改計數(shù)值
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);// 計數(shù)值加1
return null;
}
值得關(guān)注的是 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，但也只能保證其引用的可見性，并不能確保其數(shù)組中的對象是否是最新的，所以需要 Unsafe 類 volatile 式地拿到最新的 Node
2.2.1. 總結(jié)
put() 方法的核心思想：由于其減小了鎖的粒度，若 Hash 完美不沖突的情況下，可同時支持 n 個線程同時 put 操作，n 為 Node 數(shù)組大小，在默認大小 16 下，可以支持最大同時 16 個線程無競爭同時操作且線程安全
當 Hash 沖突嚴重時，Node 鏈表越來越長，將導(dǎo)致嚴重的鎖競爭，此時會進行擴容，將 Node 進行再散列，下面會介紹擴容的線程安全性。
總結(jié)一下用到的并發(fā)技巧
- 減小鎖粒度：將 Node 鏈表的頭節(jié)點作為鎖，若在默認大小 16 情況下，將有 16 把鎖，大大減小了鎖競爭（上下文切換），就像開頭所說，將串行的部分最大化縮小，在理想情況下線程的 put 操作都為并行操作。同時直接鎖住頭節(jié)點，保證了線程安全
- 使用了 volatile 修飾 table 變量，并使用 Unsafe 的 getObjectVolatile() 方法拿到最新的 Node
- CAS 操作：如果上述拿到的最新的 Node 為 null，則說明還沒有任何線程在此 Node 位置進行插入操作，說明本次操作是第一次
- synchronized 同步鎖：如果此時拿到的最新的 Node 不為 null，則說明已經(jīng)有線程在此 Node 位置進行了插入操作，此時就產(chǎn)生了 hash 沖突；此時的 synchronized 同步鎖就起到了關(guān)鍵作用，防止在多線程的情況下發(fā)生數(shù)據(jù)覆蓋（線程不安全），接著在 synchronized 同步鎖的管理下按照相應(yīng)的規(guī)則執(zhí)行操作
當 hash 值相同并 key 值也相同時，則替換掉原 value
否則，將數(shù)據(jù)插入鏈表或紅黑樹相應(yīng)的節(jié)點
2.3. get 操作時的線程安全
對于 get 操作其實沒有線程安全的問題，只有可見性的問題，只需要確保 get 的數(shù)據(jù)是線程之間可見的即可
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) {
// 檢查頭結(jié)點，table[i]就是待查找的項，直接返回
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
// hash值<0, 說明遇到特殊結(jié)點(非鏈表結(jié)點), 調(diào)用find()方法查找
else if (eh < 0)
// 調(diào)用節(jié)點對象的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[] 數(shù)組中的 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 支持多線程并發(fā)擴容，在擴容過程中同時支持 get 查數(shù)據(jù)，若有線程 put 數(shù)據(jù)，還會幫助一起擴容，這種無阻塞算法，將并行最大化的設(shè)計，堪稱一絕
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
// stride可理解成“步長”，即“數(shù)據(jù)遷移”時，每個線程要負責舊table中的多少個桶，根據(jù)幾核的CPU決定“步長”
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
// 本線程分到的遷移量，假設(shè)為16（默認也為16）
stride = MIN_TRANSFER_STRIDE;
// 說明第一次擴容
if (nextTab == null) {
try {
@SuppressWarnings("unchecked")
// 創(chuàng)建新table數(shù)組，擴大一倍為 32,n還為16
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
nextTab = nt;
} catch (Throwable ex) {
// 將表示容量的sizeCtl 設(shè)置為最大值，然后返回
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;
// 表示當前線程要進行數(shù)據(jù)遷移的桶區(qū)間
transferIndex = n;
}
int nextn = nextTab.length;
// 在get或者put時若遇到此 Node，則可以知道當前Node正在遷移
// ForwardingNode結(jié)點，當舊table的某個桶中的所有結(jié)點都遷移完后，用該結(jié)點占據(jù)這個桶
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
// 標識一個桶的遷移工作是否完成，advance == true 表示可以進行下一個位置的遷移
boolean advance = true;
// 最后一個數(shù)據(jù)遷移的線程將該值置為true，并進行本輪擴容的收尾工作
boolean finishing = false;
// i標識桶索引, bound標識邊界
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
// 每一次自旋前的預(yù)處理，主要是為了定位本輪處理的桶區(qū)間
// 正常情況下，預(yù)處理完成后：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任務(wù)的線程或出現(xiàn)擴容沖突
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
if (finishing) {// 所有桶遷移均已完成
nextTable = null;
table = nextTab;
sizeCtl = (n << 1) - (n >>> 1);
return;
}
// 擴容線程數(shù)減1,表示當前線程已完成自己的transfer任務(wù)
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
// 判斷當前線程是否是本輪擴容中的最后一個線程，如果不是，則直接退出
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
return;
finishing = advance = true;
/**
* 最后一個數(shù)據(jù)遷移線程要重新檢查一次舊table中的所有桶，看是否都被正確遷移到新table了：
* ①正常情況下，重新檢查時，舊table的所有桶都應(yīng)該是ForwardingNode;
* ②特殊情況下，比如擴容沖突(多個線程申請到了同一個transfer任務(wù))，此時當前線程領(lǐng)取的任務(wù)會作廢，那么最后檢查時，
* 還要處理因為作廢而沒有被遷移的桶，把它們正確遷移到新table中
*/
i = n;
}
}
// 舊桶本身為null，不用遷移，直接嘗試放一個ForwardingNode
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
// 該舊桶已經(jīng)遷移完成，直接跳過
else if ((fh = f.hash) == MOVED)
advance = true;
// 該舊桶未遷移完成，進行數(shù)據(jù)遷移
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不同的結(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;
}
// 以lastRun所指向的結(jié)點為分界，將鏈表拆成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); // 設(shè)置ForwardingNode占位
advance = true; // 表示當前舊桶的結(jié)點已遷移完畢
}
else if (f instanceof TreeBin) {
/**
* 下面的過程會先以鏈表方式遍歷，復(fù)制所有結(jié)點，然后根據(jù)高低位組裝成兩個鏈表；
* 然后看下是否需要進行紅黑樹轉(zhuǎn)換，最后放到新table對應(yīng)的桶中
*/
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;
}
}
// 判斷是否需要進行 紅黑樹 <-> 鏈表 的轉(zhuǎn)換
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); // 設(shè)置ForwardingNode占位
advance = true; // 表示當前舊桶的結(jié)點已遷移完畢
}
}
}
}
}
}
2.4.1. 擴容時的 get 操作
假設(shè) Node下標為 16 的 Node 節(jié)點正在遷移擴容，突然有一個線程進來調(diào)用 get() 方法，正好 key 又散列到下標為 16 的節(jié)點，此時怎么辦？
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) {
// 檢查頭結(jié)點，table[i]就是待查找的項，直接返回
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
// hash值<0, 說明遇到特殊結(jié)點(非鏈表結(jié)點), 調(diào)用find()方法查找
else if (eh < 0)
// 調(diào)用節(jié)點對象的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，所以此時判斷線程正在遷移，委托給內(nèi)部類 ForwardingNode 占位 Node 去查找值
// //內(nèi)部類 ForwardingNode 中 find() 方法
Node<K,V> find(int h, Object k) {
// 這里的查找，是去新Node數(shù)組中查找的
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[] 數(shù)組的引用也是因為這個，它可以支持在遷移的過程中照樣不阻塞地查找值，可謂是精妙絕倫的設(shè)計
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
2.4.2. 多線程協(xié)助擴容
在 put 操作時，假設(shè)正在遷移擴容，正好有一個線程進來，想要 put 值到遷移的 Node上，怎么辦？
在 put() 方法中調(diào)用了 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，表示多一個線程進來協(xié)助擴容
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
transfer(tab, nextTab);
break;
}
}
return nextTab;
}
return table;
}
此方法涉及大量復(fù)雜的位運算，這里只是簡單的說幾句，此時 sizeCtl 變量用來表示 ConcurrentHashMap 正在擴容，當其準備擴容時，會將 sizeCtl 設(shè)置為一個負數(shù)
2.4.3. 總結(jié)
ConcurrentHashMap 運用各類 CAS 操作，將擴容操作的并發(fā)性能實現(xiàn)最大化，在擴容過程中，
- 就算有線程調(diào)用 get 查詢方法，也可以安全的查詢數(shù)據(jù)
- 若有線程進行 put 操作，還會協(xié)助擴容
利用 sizeCtl 標記位和各種 volatile 變量進行 CAS 操作達到多線程之間的通信、協(xié)助，在遷移擴容過程中只鎖一個 Node 節(jié)點，即保證了線程安全，又提高了并發(fā)性能
3. 什么情況下 ConcurrentHashMap 會進行擴容操作
在 put 值時，發(fā)現(xiàn) Node 為占位 Node（ForwardingNode）時，會協(xié)助擴容
// 在 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)
// 調(diào)用treeifyBin方法
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
......
}treeifyBin() 方法會將單鏈表轉(zhuǎn)換為紅黑樹，增加查找效率，但在這之前，會檢查數(shù)組長度，若小于 64，則會優(yōu)先做擴容操作
private final void treeifyBin(Node<K,V>[] tab, int index) {
Node<K,V> b; int n, sc;
if (tab != null) {
// MIN_TREEIFY_CAPACITY = 64，若數(shù)組長度小于64，則先擴容
if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
// 擴容
tryPresize(n << 1);
else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
synchronized (b) {
// 轉(zhuǎn)換為紅黑樹的操作
......
}
}
}
}在每次 put 值之后，都會調(diào)用 addCount() 方法，檢測 Node[] 數(shù)組大小是否達到閾值
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)) {
// 統(tǒng)計元素個數(shù)的操作
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;
// 元素個數(shù)達到閾值，進行擴容
while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
(n = tab.length) < MAXIMUM_CAPACITY) {
int rs = resizeStamp(n);
// 發(fā)現(xiàn)sizeCtl為負數(shù)，證明有線程正在遷移擴容
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);
}
// 不為負數(shù)，則為第一個遷移的線程
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
transfer(tab, null);
s = sumCount();
}
}
}
以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。
相關(guān)文章
這篇文章主要介紹了通過實例了解Spring中@Profile的作用,文中通過示例代碼介紹的非常詳細，對大家的學(xué)習或者工作具有一定的參考學(xué)習價值,需要的朋友可以參考下2019-11-11
這篇文章主要介紹了java 查找字符串所在的位置代碼，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-09-09
Java中的ReentrantReadWriteLock實現(xiàn)原理詳解
這篇文章主要介紹了Java中的ReentrantReadWriteLock實現(xiàn)原理詳解,讀寫鎖實現(xiàn)了接口ReadWriteLock,適合于讀多寫少的情況,支持公平鎖和非公平鎖,支持可沖入(進入讀鎖后可再進入讀鎖,進入寫鎖后可再進入寫鎖和讀鎖),需要的朋友可以參考下2024-01-01
這篇文章主要介紹了Java類成員訪問權(quán)限控制知識總結(jié),文中通過示例代碼介紹的非常詳細，對大家的學(xué)習或者工作具有一定的參考學(xué)習價值,需要的朋友可以參考下2020-04-04
這篇文章主要介紹了Hadoop的安裝與環(huán)境搭建教程圖解,本文圖文并茂給大家介紹的非常詳細，具有一定的參考借鑒價值，需要的朋友可以參考下2019-06-06
java input 調(diào)用手機相機和本地照片上傳圖片到服務(wù)器然后壓縮的方法
今天小編就為大家分享一篇java input 實現(xiàn)調(diào)用手機相機和本地照片上傳圖片到服務(wù)器然后壓縮的方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2018-08-08
這篇文章主要介紹了Java中的HashMap內(nèi)存泄漏問題詳解,WeakHashMap中的key是弱引用,如果再使用之后沒有及時remove掉這個key,那么當GC時key就可能會被回收,導(dǎo)致key對應(yīng)的value對象占用的內(nèi)存無法回收進而導(dǎo)致內(nèi)存泄漏,需要的朋友可以參考下2023-09-09
在我們的工作中,常常要對一個字符串進行一些操作,這里提供一些常用的方法,常常需要這些方法進行組合處理字符串,這篇文章主要給大家介紹了關(guān)于JAVA?String類中的一些常用方法,需要的朋友可以參考下2023-10-10
這篇文章主要介紹了flink?RichFunction之坑及解決方案，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-12-12
基于JavaMail的Java實現(xiàn)簡單郵件發(fā)送功能
這篇文章主要為大家詳細介紹了基于JavaMail的Java實現(xiàn)簡單郵件發(fā)送功能，具有一定的參考價值，感興趣的小伙伴們可以參考一下2017-09-09