溫馨提示×
您好，登錄后才能下訂單哦！
點(diǎn)擊 登錄注冊(cè) 即表示同意《億速云用戶服務(wù)條款》
您好，登錄后才能下訂單哦！
這篇文章主要介紹“ConcurrentHashMap是怎么實(shí)現(xiàn)線程安全的”的相關(guān)知識(shí)，小編通過實(shí)際案例向大家展示操作過程，操作方法簡單快捷，實(shí)用性強(qiáng)，希望這篇“ConcurrentHashMap是怎么實(shí)現(xiàn)線程安全的”文章能幫助大家解決問題。
我們知道，在日常開發(fā)中使用的 HashMap 是線程不安全的，而線程安全類 HashTable 和 SynchronizedMap 只是簡單的在方法上加鎖實(shí)現(xiàn)了線程安全，效率低下，所以在線程安全的環(huán)境下我們通常會(huì)使用 ConcurrentHashMap，那么 ConcurrentHashMap 又是如何實(shí)現(xiàn)線程安全的呢?
針對(duì)這個(gè)問題，可以從以下幾個(gè)方面來閱讀源碼予以解答
在 JDK 1.8 中，初始化 ConcurrentHashMap 的時(shí)候這個(gè) Node[] 數(shù)組是還未初始化的，會(huì)等到第一次 put() 方法調(diào)用時(shí)才初始化
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();
......
}此時(shí)會(huì)有并發(fā)問題的，如果多個(gè)線程同時(shí)調(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是一個(gè)標(biāo)記位，若為-1，代表有線程在進(jìn)行初始化工作了
if ((sc = sizeCtl) < 0)
// 讓出CPU時(shí)間片
Thread.yield();
// 此時(shí)，代表沒有線程在進(jìn)行初始化工作，CAS操作，將本實(shí)例的sizeCtl變量設(shè)置為-1
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
// 如果CAS操作成功了，代表本線程將負(fù)責(zé)初始化工作
try {
// 再檢查一遍數(shù)組是否為空
if ((tab = table) == null || tab.length == 0) {
// 在初始化ConcurrentHashMap時(shí)，sizeCtl代表數(shù)組大小，默認(rèn)16
// 所以此時(shí)n默認(rè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;
// 通過位運(yùn)算，n減去n二進(jìn)制右移2位，相當(dāng)于乘以0.75
// 例如16經(jīng)過運(yùn)算為12，與乘0.75一樣，只不過位運(yùn)算更快
sc = n - (n >>> 2);
}
} finally {
// 將計(jì)算后的sc（12）直接賦值給sizeCtl，表示達(dá)到12長度就擴(kuò)容
// 由于這里只會(huì)有一個(gè)線程在執(zhí)行，直接賦值即可，沒有線程安全問題，只需要保證可見性
sizeCtl = sc;
}
break;
}
}
return tab;
}table 變量使用了 volatile 來保證每次獲取到的都是最新寫入的值
transient volatile Node<K,V>[] table;
ConcurrentHashMap 源碼中 sizeCtl 變量注釋如下
// 表初始化和調(diào)整控件大小。如果為負(fù)值，則表正在初始化或調(diào)整大?。?1用于初始化，否則-（1+活動(dòng)調(diào)整大小線程的數(shù)量） // 否則，當(dāng)table為null時(shí)，將保留創(chuàng)建時(shí)使用的初始表大小，默認(rèn)值為0。初始化后，保存下一個(gè)要調(diào)整表大小的元素計(jì)數(shù)值 private transient volatile int sizeCtl;
在 ConcurrentHashMap 初始化時(shí)，初始化 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é)
就算有多個(gè)線程同時(shí)進(jìn)行 put 操作，在初始化 Node[] 數(shù)組時(shí)，使用了 CAS 操作來決定到底是哪個(gè)線程有資格進(jìn)行初始化，其他線程只能等待。
用到的并發(fā)技巧如下：
volatile 修飾 sizeCtl 變量：它是一個(gè)標(biāo)記位，用來告訴其他線程這個(gè)坑位有沒有線程在進(jìn)行初始化工作，其線程間的可見性由 volatile 保證
CAS 操作：CAS 操作保證了設(shè)置 sizeCtl 標(biāo)記位的原子性，保證了在多線程同時(shí)進(jìn)行初始化 Node[] 數(shù)組時(shí)，只有一個(gè)線程能成功
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());
// 用來計(jì)算在這個(gè)節(jié)點(diǎn)總共有多少個(gè)元素，用來控制擴(kuò)容或者轉(zhuǎn)換為樹
int binCount = 0;
// 數(shù)組的遍歷，自旋插入結(jié)點(diǎn)，直到成功
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
// 當(dāng)Node[]數(shù)組為空時(shí)，進(jìn)行初始化
if (tab == null || (n = tab.length) == 0)
tab = initTable();
// Unsafe類volatile的方式取出hashCode散列后通過與運(yùn)算得出的Node[]數(shù)組下標(biāo)值對(duì)應(yīng)的Node對(duì)象
// 此時(shí) Node 位置若為 null，則表示還沒有線程在此 Node 位置進(jìn)行插入操作，說明本次操作是第一次
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
// 如果這個(gè)位置沒有元素的話，則通過 CAS 的方式插入數(shù)據(jù)
if (casTabAt(tab, i, null,
// 創(chuàng)建一個(gè) Node 添加到數(shù)組中，null 表示的是下一個(gè)節(jié)點(diǎn)為空
new Node<K,V>(hash, key, value, null)))
// 插入成功，退出循環(huán)
break;
}
// 如果檢測(cè)到某個(gè)節(jié)點(diǎn)的 hash 值是 MOVED，則表示正在進(jìn)行數(shù)組擴(kuò)容
else if ((fh = f.hash) == MOVED)
// 幫助擴(kuò)容
tab = helpTransfer(tab, f);
// 此時(shí)，說明已經(jīng)有線程對(duì)Node[]進(jìn)行了插入操作，后面的插入很有可能會(huì)發(fā)生Hash沖突
else {
V oldVal = null;
// ----------------synchronized----------------
synchronized (f) {
// 二次確認(rèn)此Node對(duì)象還是原來的那一個(gè)
if (tabAt(tab, i) == f) {
// ----------------table[i]是鏈表結(jié)點(diǎn)----------------
if (fh >= 0) {
// 記錄結(jié)點(diǎn)數(shù)，超過閾值后，需要轉(zhuǎn)為紅黑樹，提高查找效率
binCount = 1;
// 遍歷這個(gè)鏈表
for (Node<K,V> e = f;; ++binCount) {
K ek;
// 要存的元素的 hash 值和 key 跟要存儲(chǔ)的位置的節(jié)點(diǎn)的相同的時(shí)候，替換掉該節(jié)點(diǎn)的 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 的時(shí)候，則判斷該節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)是否為空
if ((e = e.next) == null) {
// ----------------“尾插法”插入新結(jié)點(diǎn)----------------
pred.next = new Node<K,V>(hash, key,
value, null);
break;
}
}
}
// ----------------table[i]是紅黑樹結(jié)點(diǎn)----------------
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) {
// 當(dāng)在同一個(gè)節(jié)點(diǎn)的數(shù)目達(dá)到8個(gè)的時(shí)候，則擴(kuò)張數(shù)組或?qū)⒔o節(jié)點(diǎn)的數(shù)據(jù)轉(zhuǎn)為tree
if (binCount >= TREEIFY_THRESHOLD)
// 鏈表 -> 紅黑樹 轉(zhuǎn)換
treeifyBin(tab, i);
// 表明本次put操作只是替換了舊值，不用更改計(jì)數(shù)值
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);// 計(jì)數(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ù)組中的對(duì)象是否是最新的，所以需要 Unsafe 類 volatile 式地拿到最新的 Node
put() 方法的核心思想：由于其減小了鎖的粒度，若 Hash 完美不沖突的情況下，可同時(shí)支持 n 個(gè)線程同時(shí) put 操作，n 為 Node 數(shù)組大小，在默認(rèn)大小 16 下，可以支持最大同時(shí) 16 個(gè)線程無競(jìng)爭同時(shí)操作且線程安全
當(dāng) Hash 沖突嚴(yán)重時(shí)，Node 鏈表越來越長，將導(dǎo)致嚴(yán)重的鎖競(jìng)爭，此時(shí)會(huì)進(jìn)行擴(kuò)容，將 Node 進(jìn)行再散列，下面會(huì)介紹擴(kuò)容的線程安全性。
總結(jié)一下用到的并發(fā)技巧
減小鎖粒度：將 Node 鏈表的頭節(jié)點(diǎn)作為鎖，若在默認(rèn)大小 16 情況下，將有 16 把鎖，大大減小了鎖競(jìng)爭（上下文切換），就像開頭所說，將串行的部分最大化縮小，在理想情況下線程的 put 操作都為并行操作。同時(shí)直接鎖住頭節(jié)點(diǎn)，保證了線程安全
使用了 volatile 修飾 table 變量，并使用 Unsafe 的 getObjectVolatile() 方法拿到最新的 Node
CAS 操作：如果上述拿到的最新的 Node 為 null，則說明還沒有任何線程在此 Node 位置進(jìn)行插入操作，說明本次操作是第一次
synchronized 同步鎖：如果此時(shí)拿到的最新的 Node 不為 null，則說明已經(jīng)有線程在此 Node 位置進(jìn)行了插入操作，此時(shí)就產(chǎn)生了 hash 沖突；此時(shí)的 synchronized 同步鎖就起到了關(guān)鍵作用，防止在多線程的情況下發(fā)生數(shù)據(jù)覆蓋（線程不安全），接著在 synchronized 同步鎖的管理下按照相應(yīng)的規(guī)則執(zhí)行操作
當(dāng) hash 值相同并 key 值也相同時(shí)，則替換掉原 value
否則，將數(shù)據(jù)插入鏈表或紅黑樹相應(yīng)的節(jié)點(diǎn)
對(duì)于 get 操作其實(shí)沒有線程安全的問題，只有可見性的問題，只需要確保 get 的數(shù)據(jù)是線程之間可見的即可
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
// 重新計(jì)算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é)點(diǎn)，table[i]就是待查找的項(xiàng)，直接返回
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
// hash值<0, 說明遇到特殊結(jié)點(diǎn)(非鏈表結(jié)點(diǎn)), 調(diào)用find()方法查找
else if (eh < 0)
// 調(diào)用節(jié)點(diǎn)對(duì)象的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);
}在擴(kuò)容時(shí)，ConcurrentHashMap 支持多線程并發(fā)擴(kuò)容，在擴(kuò)容過程中同時(shí)支持 get 查數(shù)據(jù)，若有線程 put 數(shù)據(jù)，還會(huì)幫助一起擴(kuò)容，這種無阻塞算法，將并行最大化的設(shè)計(jì)，堪稱一絕
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
// stride可理解成“步長”，即“數(shù)據(jù)遷移”時(shí)，每個(gè)線程要負(fù)責(zé)舊table中的多少個(gè)桶，根據(jù)幾核的CPU決定“步長”
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
// 本線程分到的遷移量，假設(shè)為16（默認(rèn)也為16）
stride = MIN_TRANSFER_STRIDE;
// 說明第一次擴(kuò)容
if (nextTab == null) {
try {
@SuppressWarnings("unchecked")
// 創(chuàng)建新table數(shù)組，擴(kuò)大一倍為 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;
// 表示當(dāng)前線程要進(jìn)行數(shù)據(jù)遷移的桶區(qū)間
transferIndex = n;
}
int nextn = nextTab.length;
// 在get或者put時(shí)若遇到此 Node，則可以知道當(dāng)前Node正在遷移
// ForwardingNode結(jié)點(diǎn)，當(dāng)舊table的某個(gè)桶中的所有結(jié)點(diǎn)都遷移完后，用該結(jié)點(diǎn)占據(jù)這個(gè)桶
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
// 標(biāo)識(shí)一個(gè)桶的遷移工作是否完成，advance == true 表示可以進(jìn)行下一個(gè)位置的遷移
boolean advance = true;
// 最后一個(gè)數(shù)據(jù)遷移的線程將該值置為true，并進(jìn)行本輪擴(kuò)容的收尾工作
boolean finishing = false;
// i標(biāo)識(shí)桶索引, bound標(biāo)識(shí)邊界
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;
}
}
// 當(dāng)前是處理最后一個(gè)tranfer任務(wù)的線程或出現(xiàn)擴(kuò)容沖突
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
if (finishing) {// 所有桶遷移均已完成
nextTable = null;
table = nextTab;
sizeCtl = (n << 1) - (n >>> 1);
return;
}
// 擴(kuò)容線程數(shù)減1,表示當(dāng)前線程已完成自己的transfer任務(wù)
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
// 判斷當(dāng)前線程是否是本輪擴(kuò)容中的最后一個(gè)線程，如果不是，則直接退出
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
return;
finishing = advance = true;
/**
* 最后一個(gè)數(shù)據(jù)遷移線程要重新檢查一次舊table中的所有桶，看是否都被正確遷移到新table了：
* ①正常情況下，重新檢查時(shí)，舊table的所有桶都應(yīng)該是ForwardingNode;
* ②特殊情況下，比如擴(kuò)容沖突(多個(gè)線程申請(qǐng)到了同一個(gè)transfer任務(wù))，此時(shí)當(dāng)前線程領(lǐng)取的任務(wù)會(huì)作廢，那么最后檢查時(shí)，
* 還要處理因?yàn)樽鲝U而沒有被遷移的桶，把它們正確遷移到新table中
*/
i = n;
}
}
// 舊桶本身為null，不用遷移，直接嘗試放一個(gè)ForwardingNode
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
// 該舊桶已經(jīng)遷移完成，直接跳過
else if ((fh = f.hash) == MOVED)
advance = true;
// 該舊桶未遷移完成，進(jìn)行數(shù)據(jù)遷移
else {
synchronized (f) {
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;
// 桶的hash>0，說明是鏈表遷移
if (fh >= 0) {
/**
* 下面的過程會(huì)將舊桶中的鏈表分成兩部分：ln鏈和hn鏈
* ln鏈會(huì)插入到新table的槽i中，hn鏈會(huì)插入到新table的槽i+n中
*/
int runBit = fh & n;
// lastRun指向最后一個(gè)相鄰runBit不同的結(jié)點(diǎ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;
}
// 以lastRun所指向的結(jié)點(diǎn)為分界，將鏈表拆成2個(gè)子鏈表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; // 表示當(dāng)前舊桶的結(jié)點(diǎn)已遷移完畢
}
else if (f instanceof TreeBin) {
/**
* 下面的過程會(huì)先以鏈表方式遍歷，復(fù)制所有結(jié)點(diǎn)，然后根據(jù)高低位組裝成兩個(gè)鏈表；
* 然后看下是否需要進(jìn)行紅黑樹轉(zhuǎn)換，最后放到新table對(duì)應(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;
}
}
// 判斷是否需要進(jìn)行 紅黑樹 <-> 鏈表 的轉(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; // 表示當(dāng)前舊桶的結(jié)點(diǎn)已遷移完畢
}
}
}
}
}
}2.4.1. 擴(kuò)容時(shí)的 get 操作
假設(shè) Node下標(biāo)為 16 的 Node 節(jié)點(diǎn)正在遷移擴(kuò)容，突然有一個(gè)線程進(jìn)來調(diào)用 get() 方法，正好 key 又散列到下標(biāo)為 16 的節(jié)點(diǎn)，此時(shí)怎么辦？
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
// 重新計(jì)算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é)點(diǎn)，table[i]就是待查找的項(xiàng)，直接返回
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
// hash值<0, 說明遇到特殊結(jié)點(diǎn)(非鏈表結(jié)點(diǎn)), 調(diào)用find()方法查找
else if (eh < 0)
// 調(diào)用節(jié)點(diǎn)對(duì)象的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() 操作的源碼中，會(huì)判斷 Node 中的 hash 是否小于 0（eh < 0），是否還記得我們的占位 Node，其 hash 為 MOVED，為常量值 -1，所以此時(shí)判斷線程正在遷移，委托給內(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ù)組的引用也是因?yàn)檫@個(gè)，它可以支持在遷移的過程中照樣不阻塞地查找值，可謂是精妙絕倫的設(shè)計(jì)
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
2.4.2. 多線程協(xié)助擴(kuò)容
在 put 操作時(shí)，假設(shè)正在遷移擴(kuò)容，正好有一個(gè)線程進(jìn)來，想要 put 值到遷移的 Node上，怎么辦？
在 put() 方法中調(diào)用了 helpTransfer() 方法
// put() 方法中的代碼片段，幫助擴(kuò)容
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，表示多一個(gè)線程進(jìn)來協(xié)助擴(kuò)容
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
transfer(tab, nextTab);
break;
}
}
return nextTab;
}
return table;
}此方法涉及大量復(fù)雜的位運(yùn)算，這里只是簡單的說幾句，此時(shí) sizeCtl 變量用來表示 ConcurrentHashMap 正在擴(kuò)容，當(dāng)其準(zhǔn)備擴(kuò)容時(shí)，會(huì)將 sizeCtl 設(shè)置為一個(gè)負(fù)數(shù)
2.4.3. 總結(jié)
ConcurrentHashMap 運(yùn)用各類 CAS 操作，將擴(kuò)容操作的并發(fā)性能實(shí)現(xiàn)最大化，在擴(kuò)容過程中，
就算有線程調(diào)用 get 查詢方法，也可以安全的查詢數(shù)據(jù)
若有線程進(jìn)行 put 操作，還會(huì)協(xié)助擴(kuò)容
利用 sizeCtl 標(biāo)記位和各種 volatile 變量進(jìn)行 CAS 操作達(dá)到多線程之間的通信、協(xié)助，在遷移擴(kuò)容過程中只鎖一個(gè) Node 節(jié)點(diǎn)，即保證了線程安全，又提高了并發(fā)性能
在 put 值時(shí)，發(fā)現(xiàn) Node 為占位 Node（ForwardingNode）時(shí)，會(huì)協(xié)助擴(kuò)容
// 在 put() 方法中的代碼片段 else if ((fh = f.hash) == MOVED) tab = helpTransfer(tab, f);
在 put 值時(shí)，檢測(cè)到單鏈表長度大于 8 時(shí)
final V putVal(K key, V value, boolean onlyIfAbsent) {
......
if (binCount != 0) {
// TREEIFY_THRESHOLD=8，當(dāng)鏈表長度大于8時(shí)
if (binCount >= TREEIFY_THRESHOLD)
// 調(diào)用treeifyBin方法
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
......
}treeifyBin() 方法會(huì)將單鏈表轉(zhuǎn)換為紅黑樹，增加查找效率，但在這之前，會(huì)檢查數(shù)組長度，若小于 64，則會(huì)優(yōu)先做擴(kuò)容操作
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，則先擴(kuò)容
if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
// 擴(kuò)容
tryPresize(n << 1);
else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
synchronized (b) {
// 轉(zhuǎn)換為紅黑樹的操作
......
}
}
}
}在每次 put 值之后，都會(huì)調(diào)用 addCount() 方法，檢測(cè) Node[] 數(shù)組大小是否達(dá)到閾值
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)計(jì)元素個(gè)數(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;
// 元素個(gè)數(shù)達(dá)到閾值，進(jìn)行擴(kuò)容
while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
(n = tab.length) < MAXIMUM_CAPACITY) {
int rs = resizeStamp(n);
// 發(fā)現(xiàn)sizeCtl為負(fù)數(shù)，證明有線程正在遷移擴(kuò)容
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);
}
// 不為負(fù)數(shù)，則為第一個(gè)遷移的線程
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
transfer(tab, null);
s = sumCount();
}
}
}關(guān)于“ConcurrentHashMap是怎么實(shí)現(xiàn)線程安全的”的內(nèi)容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識(shí)，可以關(guān)注億速云行業(yè)資訊頻道，小編每天都會(huì)為大家更新不同的知識(shí)點(diǎn)。
免責(zé)聲明：本站發(fā)布的內(nèi)容（圖片、視頻和文字）以原創(chuàng)、轉(zhuǎn)載和分享為主，文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如果涉及侵權(quán)請(qǐng)聯(lián)系站長郵箱：is@yisu.com進(jìn)行舉報(bào)，并提供相關(guān)證據(jù)，一經(jīng)查實(shí)，將立刻刪除涉嫌侵權(quán)內(nèi)容。
億速云公眾號(hào)
手機(jī)網(wǎng)站二維碼