緩存之美：從根上理解 ConcurrentHashMap
本文將詳細(xì)介紹 ConcurrentHashMap 構(gòu)造方法、添加值方法和擴容操作等源碼實現(xiàn)。ConcurrentHashMap 是線程安全的哈希表，此哈希表的設(shè)計主要目的是在最小化更新操作對哈希表的占用，以保持并發(fā)可讀性，次要目的是保持空間消耗與 HashMap 相同或更好，并支持利用多線程在空表上高效地插入初始值。在 Java 8 及之后的版本，使用 CAS 操作、 synchronized 關(guān)鍵字、合適的 自旋重試 和 volatile關(guān)鍵字（保證可見性和禁止指令重排）來保證并發(fā)安全，并對節(jié)點進(jìn)行了優(yōu)化：采用了鏈表和紅黑樹的實現(xiàn)，在鏈表節(jié)點數(shù)量大于等于 8 且數(shù)組（在后文中會稱每個元素位置為桶）大小大于等于 64 時會轉(zhuǎn)變?yōu)榧t黑樹，在擴容邏輯中，當(dāng)樹節(jié)點小于等于 6 時又會轉(zhuǎn)換成鏈表（刪除邏輯中鏈表轉(zhuǎn)換紅黑樹的邏輯并不嚴(yán)格按照大小為 6 的閾值），優(yōu)化空間利用并提高查詢效率。它的默認(rèn)大小為 16，負(fù)載因子為 0.75F，負(fù)載因子不支持指定其他值，這是與 HashMap 的不同點，在講解構(gòu)造方法的源碼時，會提到這一點，大家需要留意，接下來步入正文：
構(gòu)造方法
首先我們來看它的構(gòu)造方法，重點關(guān)注注釋信息：
public class ConcurrentHashMap extends AbstractMap implements ConcurrentMap, Serializable {
private transient volatile int sizeCtl;
private static final int MAXIMUM_CAPACITY = 1 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
// 如果指定的并發(fā)線程數(shù)大于初始化容量，那么以并發(fā)線程數(shù)為準(zhǔn)
if (initialCapacity < concurrencyLevel)
initialCapacity = concurrencyLevel;
long size = (long) (1.0 + (long) initialCapacity / loadFactor);
int cap = (size >= (long) MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : tableSizeFor((int) size);
this.sizeCtl = cap;
}
// 向上取整 2 的n次冪
private static final int tableSizeFor(int c) {
// Integer.numberOfLeadingZeros(c - 1) 用于計算 c-1 的二進(jìn)制表示中最高位 1 之前有多少個 0
// -1 的二級制表示為 11111111111111111111111111111111（32個1），無符號右移則會得到某 2的n次冪-1 的結(jié)果
int n = -1 >>> Integer.numberOfLeadingZeros(c - 1);
// 限制最大值的同時，結(jié)果永遠(yuǎn)為 2 的 n 次冪
return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;
}
}
負(fù)載因子 loadFactor 作為局部變量計算完 size 后，并沒有被記錄下來，后續(xù)有關(guān)該值的邏輯，如擴容閾值的計算均使用了默認(rèn)值 0.75F。這么做的原因在源碼 JavaDoc 中有詳細(xì)的解釋：
理想情況下，容器中的節(jié)點遵循泊松分布，一個桶中有 k 個元素的概率分布如下：
0：0.60653066
1：0.30326533
2：0.07581633
3：0.01263606
4：0.00157952
5：0.00015795
6：0.00001316
7：0.00000094
8：0.00000006
更多：不到千萬分之一
在隨機散列下，兩個線程訪問不同元素的鎖爭用概率約為 1 / (8 * 元素數(shù)量)。
在負(fù)載因子為 0.75F 時，能夠較好的防止多個元素發(fā)生碰撞，在隨機散列的情況下，多線程發(fā)生鎖爭搶的概率較低。
ConcurrentHashMap#tableSizeFor 方法計算結(jié)果會將數(shù)組大小固定為 2 的 n 次冪，這樣做是為了 提高性能和簡化實現(xiàn)，以下為詳細(xì)解釋：
位運算優(yōu)化：當(dāng)哈希表的大小是 2 的 n 次冪時，可以使用位運算來代替取模運算（%），從而提高哈希表操作的性能。比如計算某元素在數(shù)組中的位置，index = hash % table.length 可以簡化為 index = hash & (table.length - 1)，位運算 & 通常比取模運算更快
哈希分布均勻性：2 的 n 次冪減 1 的 2 進(jìn)制表示中低位均為 1，哈希值與它進(jìn)行位與計算可直接獲取索引值，這樣可以減少哈希沖突的概率，使分布更加均勻
簡化擴容邏輯：在擴容時，直接指定新表的大小是舊表的兩倍（也是 2 的 n 次冪），元素的重新分配變得更加簡便，要么元素的位置要么保持不變，要么移動到新位置 index + oldCapacity，這種移動邏輯可以通過簡單的位運算實現(xiàn)
put 方法
put 方法是核心方法，需要重點關(guān)注添加值時使用到的 CAS + synchronized 的同步機制。更新元素計數(shù)的 addCount 方法采用了非常巧妙的實現(xiàn)，后文中我們會詳細(xì)介紹。除此之外，擴容操作也會專門進(jìn)行說明，它協(xié)調(diào)多線程共同完成擴容的解決方案也很值得學(xué)習(xí)，這些內(nèi)容掌握之后，ConcurrentHashMap 中也再沒有更難的內(nèi)容。我們以 put 為切入點，重點關(guān)注注釋信息：
public class ConcurrentHashMap extends AbstractMap implements ConcurrentMap, Serializable {
static final int TREEIFY_THRESHOLD = 8;
// 最高位為 0，其余位為 1
static final int HASH_BITS = 0x7fffffff;
private static final int DEFAULT_CAPACITY = 16;
transient volatile Node[] table;
private transient volatile int sizeCtl;
public V put(K key, V value) {
return putVal(key, value, false);
}
final V putVal(K key, V value, boolean onlyIfAbsent) {
// key 和 value 均不能為 null
if (key == null || value == null) throw new NullPointerException();
// hash 擾動，用于使 hash 結(jié)果均勻分布
int hash = spread(key.hashCode());
int binCount = 0;
for (Node[] tab = table;;) {
Node f; int n, i, fh; K fk; V fv;
// 懶加載實現(xiàn)初始化
if (tab == null || (n = tab.length) == 0)
tab = initTable();
// 如果元素 hash （數(shù)組大小 - 1 & hash）到桶的位置沒有元素（為 null）
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
// 通過 CAS 操作直接將元素放到該位置
if (casTabAt(tab, i, null, new Node(hash, key, value)))
break;
}
// 擴容相關(guān)邏輯，后續(xù)再講解
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
// onlyIfAbsent 是入?yún)?，默認(rèn)為 false，true 表示鍵不存在時才插入新值，存在相同鍵值不能覆蓋；false 為一直覆蓋邏輯
// 該邏輯滿足在此特定條件下，避免獲取鎖，從而提高性能
else if (onlyIfAbsent && fh == hash && ((fk = f.key) == key || (fk != null && key.equals(fk))) && (fv = f.val) != null)
return fv;
// 執(zhí)行到此處意味著該元素 hash 到的桶位置存在元素，需要追加到此處的鏈表或紅黑樹上，f 為該桶位置的第一個元素
else {
V oldVal = null;
// 鎖住第一個元素
synchronized (f) {
// 校驗此處元素并沒有發(fā)生變更（類似單例模式的雙重檢測鎖機制）
if (tabAt(tab, i) == f) {
// 頭節(jié)點的 hash 值大于 0，說明是鏈表
if (fh >= 0) {
binCount = 1;
// 注意這里會對鏈表中元素數(shù)量進(jìn)行累加
for (Node e = f;; ++binCount) {
K ek;
// 如果發(fā)現(xiàn)了 key 值相同的元素，根據(jù) onlyIfAbsent 字段判斷是否需要覆蓋
if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) {
oldVal = e.val;
if (!onlyIfAbsent)
e.val = value;
break;
}
// 直到找到鏈表的尾巴節(jié)點，進(jìn)行添加（尾插法）
Node pred = e;
if ((e = e.next) == null) {
pred.next = new Node(hash, key, value);
break;
}
}
}
// 如果該元素是紅黑樹
else if (f instanceof TreeBin) {
Node p;
binCount = 2;
// 調(diào)用紅黑樹添加元素的方法
if ((p = ((TreeBin)f).putTreeVal(hash, key, value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
// ReservationNode 為 computeIfAbsent 和 compute 方法中使用的占位節(jié)點，同樣也是為了保證并發(fā)環(huán)境下的正確性和一致性
else if (f instanceof ReservationNode)
throw new IllegalStateException("Recursive update");
}
}
if (binCount != 0) {
// 如果鏈表中元素數(shù)量超過了樹化閾值，則將鏈表轉(zhuǎn)換為紅黑樹
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
// 更新元素計數(shù)
addCount(1L, binCount);
return null;
}
// 擾動函數(shù)，用于計算哈希值以減少碰撞的可能性
static final int spread(int h) {
// 右移 16 位后，使該 hash 值的高 16 位和低 16 位“混合”，使 hash 值均勻分布，位與操作則是限制哈希值的范圍，并保證它為非負(fù)數(shù)
return (h ^ (h >>> 16)) & HASH_BITS;
}
private final Node[] initTable() {
Node[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
// 通過 CAS 操作將 sizeCtl 賦值為 -1，成功后 sizeCtl 為 -1 表示正在有線程在對它進(jìn)行初始化
// 如果此時再有其他線程來操作，CAS 操作會失敗，會在 while 循環(huán)中自旋等待直到完成初始化
if ((sc = sizeCtl) < 0)
Thread.yield();
else if (U.compareAndSetInt(this, SIZECTL, sc, -1)) {
try {
if ((tab = table) == null || tab.length == 0) {
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
Node[] nt = (Node[])new Node[n];
table = tab = nt;
// 計算擴容閾值，相當(dāng)于原值的 0.75F，構(gòu)造函數(shù)中指定的負(fù)載因子不會生效，均采用默認(rèn) 0.75F 來計算
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}
}
putTreeVal方法為向紅黑樹中添加元素的方法，如果想了解紅黑樹可以參考這篇文章深入理解經(jīng)典紅黑樹，在此處就不再解釋相關(guān)邏輯了。
在這段源碼邏輯中，我能能發(fā)現(xiàn)一些具有“隱藏”性的賦值操作，比如在如下邏輯中，變量 n 的賦值：
public class ConcurrentHashMap extends AbstractMap implements ConcurrentMap, Serializable {
// ...
final V putVal(K key, V value, boolean onlyIfAbsent) {
// ...
for (Node[] tab = table;;) {
Node f; int n, i, fh; K fk; V fv;
if (tab == null || (n = tab.length) == 0)
tab = initTable();
// ...
}
}
}
它是在 if 條件 判斷中完成賦值的，這樣寫代碼確實相對精簡一些，但是也僅限于在此（或技術(shù)組件中），我覺得如果在業(yè)務(wù)代碼中這樣寫，可讀性就比較差了。
put 方法向數(shù)組中每個桶添加第一個元素時，都會使用 CAS 操作。為了節(jié)省空間，并沒有為每個桶都創(chuàng)建一個鎖，而是將每個桶中第一個元素作為鎖，當(dāng)發(fā)生哈希碰撞時，依賴 synchronized 完成鎖定。這里值得關(guān)注的是：在獲取到該元素的鎖后，又重復(fù)判斷了鏈表頭節(jié)點是否仍然為該元素（雙重檢測），因為該元素可能被其他線程操作刪除，在接下來的源碼中還能看到很多在執(zhí)行完同步操作后重新再判斷是否符合條件的邏輯，這也是在提醒我們：寫線程同步相關(guān)代碼時有時需要再校驗。當(dāng)某個桶中第一個元素被鎖定時，其他操作該桶元素的線程操作會被阻塞，如果 equals 方法耗時較長，可能會影響性能，但實際上，這種情況并不常見。
addCount 更新元素計數(shù)方法
接下來我們看一下 addCount 更新元素數(shù)量的方法，該方法實現(xiàn)的元素計數(shù)非常有意思：在更新元素數(shù)量時未發(fā)生沖突則始終使用 baseCount 來表示哈希表中元素適量，一旦 CAS 更新數(shù)量失敗，它便會創(chuàng)建一個 CounterCell[] counterCells 來協(xié)助統(tǒng)計元素數(shù)量，總數(shù)量為 baseCount 和 CounterCell[] counterCells 中計數(shù)值的累加。并且判斷哈希表是否需要擴容也是在這里完成的。本節(jié)方法非常重要，需要大家重點關(guān)注：
public class ConcurrentHashMap extends AbstractMap implements ConcurrentMap, Serializable {
private static final int RESIZE_STAMP_BITS = 16;
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;
private static final int MAXIMUM_CAPACITY = 1 [] table;
// 擴容操作后要使用的數(shù)組
private transient volatile Node[] nextTable;
private transient volatile CounterCell[] counterCells;
private static final long BASECOUNT;
private transient volatile long baseCount;
// counterCells 在創(chuàng)建元素或者在擴容的標(biāo)志，使用 CAS 操作更新
transient volatile int cellsBusy;
// CPU 核數(shù)
static final int NCPU = Runtime.getRuntime().availableProcessors();
final V putVal(K key, V value, boolean onlyIfAbsent) {
// ...
// 增加計數(shù)
addCount(1L, binCount);
return null;
}
private final void addCount(long x, int check) {
CounterCell[] cs; long b, s;
// 更新元素計數(shù)，注意這里有 cas 操作更新 baseCount 的值，如果 CAS 更新失敗或 counterCells 已經(jīng)被初始化，會進(jìn)入 if 條件的執(zhí)行邏輯
if ((cs = counterCells) != null || !U.compareAndSetLong(this, BASECOUNT, b = baseCount, s = b + x)) {
CounterCell c; long v; int m;
boolean uncontended = true;
// 負(fù)責(zé) counterCells 的初始化和擴展
if (cs == null || (m = cs.length - 1) < 0 || (c = cs[ThreadLocalRandom.getProbe() & m]) == null
|| !(uncontended = U.compareAndSetLong(c, CELLVALUE, v = c.value, v + x))) {
fullAddCount(x, uncontended);
return;
}
if (check <= 1)
return;
// 計算元素總和
s = sumCount();
}
if (check >= 0) {
Node[] tab, nt; int n, sc;
// 增加元素后元素數(shù)量大于當(dāng)前 sizeCtl 大小 且 table 已被初始化 且未超過最大容量
while (s >= (long)(sc = sizeCtl) && (tab = table) != null && (n = tab.length) < MAXIMUM_CAPACITY) {
// n 為 table 的長度（length），以 n 為參數(shù)計算 resizeStamp（擴容戳），唯一標(biāo)識，用來協(xié)調(diào)多個線程同時操作 transfer
// 而再將其左移 16 位，會得到一個較大的負(fù)數(shù)，eg: -2145779712，這樣其他線程只能調(diào)用到 sc < 0 的條件分支了
int rs = resizeStamp(n) = 0 表示沒有擴容操作在執(zhí)行，CAS 操作將 sizeCtl 更新為 rs + 2，表示啟動擴容操作，此時 sc 已經(jīng)為一個負(fù)數(shù)了
else if (U.compareAndSetInt(this, SIZECTL, sc, rs + 2))
// 執(zhí)行擴容操作，第二個入?yún)?Node[] nextTab 為 null，只有一個線程能夠啟動擴容是為了 nextTab 只能被初始化一遍
transfer(tab, null);
// 計算元素總和
s = sumCount();
}
}
}
// 與 longAdder 中實現(xiàn)原理一致
private final void fullAddCount(long x, boolean wasUncontended) {
int h;
// 如果線程探針值為 0 證明它還沒被初始化，那么對它初始化
if ((h = ThreadLocalRandom.getProbe()) == 0) {
ThreadLocalRandom.localInit();
h = ThreadLocalRandom.getProbe();
wasUncontended = true;
}
// 沖突標(biāo)志，用于標(biāo)記創(chuàng)建計數(shù)元素時發(fā)生碰撞；槽位引用發(fā)生變化；槽位已滿；槽位中計數(shù)元素發(fā)生變更
boolean collide = false;
// 自旋操作嘗試更新，直到成功為止
for (;;) {
CounterCell[] cs; CounterCell c; int n; long v;
// 如果 counterCells 數(shù)組已經(jīng)初始化，并且長度大于 0
if ((cs = counterCells) != null && (n = cs.length) > 0) {
// 如果槽位為空
if ((c = cs[(n - 1) & h]) == null) {
// 沒有線程在創(chuàng)建計數(shù)元素
if (cellsBusy == 0) {
// 創(chuàng)建計數(shù)元素，并記錄計數(shù)值 x
CounterCell r = new CounterCell(x);
// 將 cellsBusy 更新為 1，標(biāo)志該線程正在創(chuàng)建計數(shù)元素
if (cellsBusy == 0 && U.compareAndSetInt(this, CELLSBUSY, 0, 1)) {
boolean created = false;
try {
CounterCell[] rs; int m, j;
// 雙重校驗該槽位未創(chuàng)建計數(shù)元素
if ((rs = counterCells) != null && (m = rs.length) > 0 && rs[j = (m - 1) & h] == null) {
// 為槽位賦值
rs[j] = r;
created = true;
}
} finally {
// 創(chuàng)建完成更新 cellsBusy 為 0，允許其他線程繼續(xù)操作
cellsBusy = 0;
}
// 創(chuàng)建完成，結(jié)束循環(huán)
if (created)
break;
continue;
}
}
// 變更沖突標(biāo)志
collide = false;
}
// 標(biāo)記 wasUncontended 為 true 后重試
else if (!wasUncontended)
wasUncontended = true;
// 成功更新計算元素值，退出循環(huán)
else if (U.compareAndSetLong(c, CELLVALUE, v = c.value, v + x))
break;
// 槽位引用已經(jīng)發(fā)生變化或槽位已滿
else if (counterCells != cs || n >= NCPU)
// 變更沖突標(biāo)志
collide = false;
// 如果已經(jīng)發(fā)生過沖突，標(biāo)記為 true
else if (!collide)
collide = true;
// 如果 counterCells 需要擴容
else if (cellsBusy == 0 && U.compareAndSetInt(this, CELLSBUSY, 0, 1)) {
try {
// 校驗 counterCells 未發(fā)生變更
if (counterCells == cs)
counterCells = Arrays.copyOf(cs, n
在元素計數(shù) fullAddCount 方法中，會為新增的元素數(shù)量 x 在 CounterCell[] 中找到一個槽位記錄或累加，在進(jìn)行該操作時使用了自旋重試保證執(zhí)行成功。元素計數(shù)累加完成后，會對是否需要擴容進(jìn)行判斷，如果元素數(shù)量超過 sizeCtl 則會進(jìn)行擴容操作，在擴容開始時，會將 sizeCtl 賦值為負(fù)數(shù)，這樣在執(zhí)行擴容方法時，只有一個線程能調(diào)用 transfer(tab, null); 方法，保證擴容后哈希表 nextTable 僅被初始化一次，完成多線程擴容的協(xié)調(diào)。這里弄明白之后，接下來重點看擴容方法 transfer。
transfer 擴容方法
擴容方法 transfer 允許多線程協(xié)同擴容，實現(xiàn)協(xié)同擴容的方法很巧妙，它定義了全局變量 transferIndex 用于記錄當(dāng)前擴容操作的進(jìn)度（為 0 時表示快完成或已經(jīng)完成，初始值為哈希表長度），規(guī)定每個線程的處理步長 stride 最小值為 16，如果 transferIndex 大于步長值 stride 時，其他線程調(diào)用該方法時會被分配擴容任務(wù)協(xié)助完成擴容，這樣每個線程便被分配了一段處理范圍，線程與線程間擴容互不影響，提高了擴容效率。以下為源碼，其中已經(jīng)注明了詳細(xì)注釋，大家需要根據(jù)源碼和注釋信息理解該過程：
public class ConcurrentHashMap extends AbstractMap implements ConcurrentMap, Serializable {
private static final int RESIZE_STAMP_BITS = 16;
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;
private static final int MIN_TRANSFER_STRIDE = 16;
// 轉(zhuǎn)發(fā)節(jié)點 forwarding nodes 的 hash 值 -1
static final int MOVED = -1;
static final int NCPU = Runtime.getRuntime().availableProcessors();
static final int UNTREEIFY_THRESHOLD = 6;
private transient volatile Node[] nextTable;
private transient volatile int transferIndex;
private transient volatile int sizeCtl;
/**
* 擴容哈希表，將舊表 tab 中元素轉(zhuǎn)移到 nextTab 中
*
* @param tab 舊表
* @param nextTab 新表
*/
private final void transfer(Node[] tab, Node[] nextTab) {
// n 為當(dāng)前哈希表長度；stride 為步長，即每個線程負(fù)責(zé)處理的桶數(shù)
int n = tab.length, stride;
// 根據(jù) CPU 核數(shù)，分配任務(wù)給多個線程，每個線程最小的處理步長為 16
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE;
// 當(dāng) nextTab == null 時，表示第一個啟動擴容的線程執(zhí)行，這段邏輯只會被執(zhí)行一次
if (nextTab == null) {
try {
// 右移 1 位，容量擴大兩倍
Node[] nt = (Node[]) new Node[n fwd = new ForwardingNode(nextTab);
// 前進(jìn)標(biāo)志，用于控制是否需要檢查 去處理下一個桶 或 是否需要被分配下一組數(shù)據(jù)的執(zhí)行任務(wù)
boolean advance = true;
// 即將完成標(biāo)志，如果該值為 true 再重新檢查一遍哈希表元素即完成擴容操作
boolean finishing = false;
for (int i = 0, bound = 0; ; ) {
// 桶中第一個節(jié)點
Node f;
// 第一個節(jié)點的 hash 值
int fh;
while (advance) {
int nextIndex, nextBound;
// 在這里更新當(dāng)前線程的處理索引 --i，如果 i 仍然在范圍內(nèi) 或 處于正在完成階段
if (--i >= bound || finishing)
// 結(jié)束循環(huán)去執(zhí)行下面的節(jié)點轉(zhuǎn)移操作
advance = false;
// transferIndex 小于等于 0 說明所有桶的處理任務(wù)已經(jīng)被分配完畢了，新線程無需再協(xié)助擴容了
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
// 結(jié)束循環(huán)去執(zhí)行下面的節(jié)點轉(zhuǎn)移操作
advance = false;
}
// 線程第一次循環(huán)都會來執(zhí)行這段邏輯來分配要處理的數(shù)據(jù)范圍
// CAS 操作更新 transferIndex 為 transferIndex - stride，該值為下一個線程的處理范圍的右邊界 nextBound
else if (U.compareAndSetInt(this, TRANSFERINDEX, nextIndex, nextBound = (nextIndex > stride ? nextIndex - stride : 0))) {
// 當(dāng)前線程處理范圍的左邊界 bound
bound = nextBound;
// 此時 nextIndex 為 CAS 更新前 transferIndex 大小，減一即表示有效索引
// 所以線程處理數(shù)據(jù)的有效范圍是 [bound, i] 的閉區(qū)間，從 i 倒序處理
i = nextIndex - 1;
// 結(jié)束循環(huán)去執(zhí)行下面的節(jié)點轉(zhuǎn)移邏輯
advance = false;
}
}
// 如果 i 超出有效范圍，檢測是否需要結(jié)束擴容
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
// 擴容已完成
if (finishing) {
nextTable = null;
table = nextTab;
// 0.75 倍容量大小，下次擴容的閾值
sizeCtl = (n >> 1);
return;
}
// CAS 操作更新 sizeCtl 為 sizeCtl - 1 表示執(zhí)行擴容的其中之一線程已經(jīng)操作完成了
if (U.compareAndSetInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
// 這里就要和 addCount 方法中執(zhí)行擴容更新 sizeCtl 的方法聯(lián)系起來
// sizeCtl 在執(zhí)行擴容時先被賦值為 resizeStamp(n) ln, hn;
// 檢查當(dāng)前節(jié)點的 hash 值是否為正整數(shù)，如果是則表示它是鏈表節(jié)點
if (fh >= 0) {
// 計算當(dāng)前節(jié)點 hash 值與舊表長度 n 位與運算結(jié)果
int runBit = fh & n;
// 先初始化 lastRun 為當(dāng)前節(jié)點，并遍歷鏈表，根據(jù)位與運算結(jié)果找到鏈表的最后一個節(jié)點由 lastRun 引用
Node lastRun = f;
for (Node p = f.next; p != null; p = p.next) {
int b = p.hash & n;
if (b != runBit) {
runBit = b;
lastRun = p;
}
}
// 根據(jù)位與運算結(jié)果，來決定這個節(jié)點是被劃分到新表的低位還是高位
if (runBit == 0) {
ln = lastRun;
hn = null;
} else {
hn = lastRun;
ln = null;
}
// 遍歷該桶中鏈表的所有節(jié)點
for (Node p = f; p != lastRun; p = p.next) {
int ph = p.hash;
K pk = p.key;
V pv = p.val;
// 根據(jù) hash 值位與運算結(jié)果，采用頭插法不斷向兩個高位和低位鏈表中添加節(jié)點
if ((ph & n) == 0)
ln = new Node(ph, pk, pv, ln);
else
hn = new Node(ph, pk, pv, hn);
}
// 低位鏈表賦值到新表中
setTabAt(nextTab, i, ln);
// 高位鏈表賦值到新表中
setTabAt(nextTab, i + n, hn);
// 將舊表的 i 索引處元素更新為轉(zhuǎn)發(fā)節(jié)點，表示已處理
setTabAt(tab, i, fwd);
// 更新為 true，以便在上文的 while 循環(huán)中判斷是否要被分配下一批任務(wù)或處理下一個桶
advance = true;
}
// 如果該節(jié)點為紅黑樹節(jié)點
else if (f instanceof TreeBin) {
TreeBin t = (TreeBin) f;
// 定義兩組變量分別記錄低位紅黑樹頭尾節(jié)點和高位紅黑樹頭尾節(jié)點
TreeNode lo = null, loTail = null;
TreeNode hi = null, hiTail = null;
// 分別記錄低位、高位節(jié)點數(shù)量
int lc = 0, hc = 0;
// 與鏈表操作相似，都是根據(jù) hash 值位與運算的結(jié)果來確定新節(jié)點的位置
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;
}
}
// 根據(jù)計數(shù)值判斷是否要將紅黑樹轉(zhuǎn)換為鏈表，注意這里紅黑樹轉(zhuǎn)換成鏈表的閾值為 6
ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
(hc != 0) ? new TreeBin(lo) : t;
hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
(lc != 0) ? new TreeBin(hi) : t;
// 將節(jié)點封裝到新哈希表中
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
// 更新該節(jié)點為轉(zhuǎn)發(fā)節(jié)點
setTabAt(tab, i, fwd);
// 更新為 true，以便在上文的 while 循環(huán)中判斷是否要被分配下一批任務(wù)或處理下一個桶
advance = true;
}
}
}
}
}
}
// 轉(zhuǎn)發(fā)節(jié)點，哈希值默認(rèn)為 -1
static final class ForwardingNode extends Node {
final Node[] nextTable;
ForwardingNode(Node[] tab) {
super(MOVED, null, null);
this.nextTable = tab;
}
// ...
}
}
在我們分析完上述源碼后再來看 helpTransfer 方法就非常容易了，該方法用于其他線程協(xié)助完成哈希表的擴容操作，本質(zhì)上還是會調(diào)用 transfer 方法：
public class ConcurrentHashMap extends AbstractMap implements ConcurrentMap, Serializable {
// ...
final Node[] helpTransfer(Node[] tab, Node f) {
Node[] nextTab;
int sc;
// 當(dāng)前哈希表已經(jīng)被初始化；當(dāng)前節(jié)點為轉(zhuǎn)發(fā)節(jié)點，表示擴容進(jìn)行中；新哈希表也完成了初始化
if (tab != null && (f instanceof ForwardingNode) && (nextTab = ((ForwardingNode) f).nextTable) != null) {
// 擴容戳，我們在討論 addCount() 方法時提到過
int rs = resizeStamp(tab.length)
到目前為止，相對復(fù)雜的邏輯已經(jīng)全部講解完了，接下來的內(nèi)容在理解了上述內(nèi)容后再看會非常簡單，所以如果沒理解上方的源碼內(nèi)容需要再去熟悉熟悉。
treeifyBin 樹化方法
treeifyBin 方法用于將鏈表轉(zhuǎn)換成紅黑樹，以提高查詢效率，邏輯非常簡單，關(guān)注注釋信息即可，如下：
public class ConcurrentHashMap extends AbstractMap implements ConcurrentMap, Serializable {
// 鏈表樹化的最小限制
static final int MIN_TREEIFY_CAPACITY = 64;
private final void treeifyBin(Node[] tab, int index) {
Node b;
int n;
if (tab != null) {
// 檢查哈希表長度是否小于 64，如果小于的話執(zhí)行的是擴容操作
if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
tryPresize(n = 0) {
// 同步鎖住該節(jié)點
synchronized (b) {
// 鎖住后校驗該節(jié)點沒有別其他線程修改
if (tabAt(tab, index) == b) {
// 將鏈表轉(zhuǎn)換為紅黑樹
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;
}
// 將原表中原表節(jié)點替換為紅黑樹節(jié)點
setTabAt(tab, index, new TreeBin(hd));
}
}
}
}
}
private final void tryPresize(int size) {
// 計算目標(biāo)容量大小，如果超過最大容量的一半，直接賦值為最大容量，否則計算出合適容量（tableSizeFor 方法已在上文提到過）
int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY : tableSizeFor(size + (size >>> 1) + 1);
int sc;
while ((sc = sizeCtl) >= 0) {
Node[] tab = table;
int n;
// 哈希表為空或者長度為 0，表示哈希表還未完成初始化
if (tab == null || (n = tab.length) == 0) {
// Math.max(sc, c);
n = (sc > c) ? sc : c;
// 更新 sizeCtl 為 -1，表示哈希表正在初始化
if (U.compareAndSetInt(this, SIZECTL, sc, -1)) {
try {
if (table == tab) {
Node[] nt = (Node[]) new Node[n];
table = nt;
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
}
} else if (c <= sc || n >= MAXIMUM_CAPACITY)
break;
// 嘗試擴容
else if (tab == table) {
// 計算擴容戳
int rs = resizeStamp(n);
// 這段邏輯就與 addCount 中第一個操作擴容的線程邏輯一致了
if (U.compareAndSetInt(this, SIZECTL, sc, (rs
get 方法
get 方法非常簡單，如下：
public class ConcurrentHashMap extends AbstractMap implements ConcurrentMap, Serializable {
public V get(Object key) {
Node[] tab;
Node e, p;
int n, eh;
K ek;
// 計算出經(jīng)過擾動的 hash 值
int h = spread(key.hashCode());
// 哈希表已經(jīng)完成初始化且該索引處元素不為 null
if ((tab = table) != null && (n = tab.length) > 0 && (e = tabAt(tab, (n - 1) & h)) != null) {
// 哈希值相等，判斷第一個節(jié)點是不是想要的節(jié)點
if ((eh = e.hash) == h) {
// key 相等則返回對應(yīng)的值
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
// 當(dāng)節(jié)點哈希值小于 0 時，表示兩種情況：要么為紅黑樹節(jié)點，要么正在擴容
} else if (eh < 0)
return (p = e.find(h, key)) != null ? p.val : null;
// 遍歷鏈表嘗試找到要匹配的節(jié)點
while ((e = e.next) != null) {
if (e.hash == h && ((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
}
}
在 eh < 0 的條件下，表示兩種情況：要么為紅黑樹節(jié)點，要么正在擴容（ForwardingNode），前者就不在這里贅述了。這里我們需要看一下在擴容時，哈希表是如何尋找對應(yīng)節(jié)點的，如下為 ForwardingNode 中查找對應(yīng)節(jié)點值的源碼：
static final class ForwardingNode extends Node {
final Node[] nextTable;
ForwardingNode(Node[] tab) {
super(MOVED, null, null);
this.nextTable = tab;
}
Node find(int h, Object k) {
// 注意這里的 outer 標(biāo)志，如果在尋找節(jié)點中發(fā)現(xiàn) e 為轉(zhuǎn)發(fā)節(jié)點，那么需要再去新的被擴容后的表中尋找對應(yīng)節(jié)點
outer: for (Node[] tab = nextTable;;) {
Node e; int n;
// 要找的對象為 null 或者 tab 未初始化 或者 桶對應(yīng)的位置為 null
if (k == null || tab == null || (n = tab.length) == 0 || (e = tabAt(tab, (n - 1) & h)) == null)
return null;
for (;;) {
int eh; K ek;
// 找到了對應(yīng)的節(jié)點
if ((eh = e.hash) == h && ((ek = e.key) == k || (ek != null && k.equals(ek))))
return e;
if (eh < 0) {
// 發(fā)現(xiàn)了轉(zhuǎn)發(fā)節(jié)點，說明該節(jié)點已經(jīng)再次被轉(zhuǎn)移到了新的哈希表中，需要去新的哈希表尋找
if (e instanceof ForwardingNode) {
tab = ((ForwardingNode)e).nextTable;
continue outer;
}
else
// 調(diào)用 node 的 find 方法
return e.find(h, k);
}
// 找完了所有節(jié)點仍然沒找到
if ((e = e.next) == null)
return null;
}
}
}
static class Node implements Map.Entry {
// 簡單地遍歷查找
Node find(int h, Object k) {
Node e = this;
if (k != null) {
do {
K ek;
if (e.hash == h &&
((ek = e.key) == k || (ek != null && k.equals(ek))))
return e;
} while ((e = e.next) != null);
}
return null;
}
}
}
由以上源碼可知，在尋找某節(jié)點時如果發(fā)現(xiàn)了 轉(zhuǎn)發(fā)節(jié)點，那么證明該節(jié)點已經(jīng)被轉(zhuǎn)移到新的哈希表中，那么需要去新的哈希表中尋找。
remove 方法
remove 方法也非常簡單，當(dāng)某節(jié)點被刪除時需要更新計數(shù)，addCount 我們在上文也介紹過，就不在贅述了。在紅黑樹中移除節(jié)點（removeTreeNode 方法）時，可能會調(diào)用到 untreeify 方法，這里的將紅黑樹轉(zhuǎn)換成鏈表的邏輯與在 transfer 中根據(jù)節(jié)點數(shù)量小于等于閾值 6 的轉(zhuǎn)換判斷邏輯不同，removeTreeNode 決定將鏈表轉(zhuǎn)換成紅黑樹時判斷的依據(jù)是紅黑樹是否太?。╰oo small）：root == null || r.right == null || (rl = r.left) == null || rl.left == null，感興趣的大家可以去源碼中了解一下，在這里就不再貼源碼了。以下為刪除方法主要邏輯，重點關(guān)注注釋信息：
public class ConcurrentHashMap extends AbstractMap implements ConcurrentMap, Serializable {
public V remove(Object key) {
return replaceNode(key, null, null);
}
final V replaceNode(Object key, V value, Object cv) {
// 計算 key 對應(yīng)的 hash 值
int hash = spread(key.hashCode());
for (Node[] tab = table;;) {
Node f; int n, i, fh;
// 如果未完成初始化或要刪除的節(jié)點不存在
if (tab == null || (n = tab.length) == 0 || (f = tabAt(tab, i = (n - 1) & hash)) == null)
break;
// 節(jié)點為轉(zhuǎn)移節(jié)點，協(xié)助擴容
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
// 處理對應(yīng)節(jié)點
else {
V oldVal = null;
// 某節(jié)點被修改或刪除，標(biāo)記為 true
boolean validated = false;
// 先加鎖
synchronized (f) {
// 加鎖完成后重復(fù)校驗該節(jié)點是否被修改過
if (tabAt(tab, i) == f) {
// 處理鏈表節(jié)點
if (fh >= 0) {
validated = true;
for (Node e = f, pred = null;;) {
K ek;
// 如果匹配到了對應(yīng)的節(jié)點
if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) {
V ev = e.val;
if (cv == null || cv == ev || (ev != null && cv.equals(ev))) {
// 記錄原節(jié)點值
oldVal = ev;
// value 不為 null 才更新為新節(jié)點值
if (value != null)
e.val = value;
// 刪除中間節(jié)點或尾節(jié)點
else if (pred != null)
pred.next = e.next;
else
// 刪除頭節(jié)點
setTabAt(tab, i, e.next);
}
break;
}
// 向后遍歷，變更引用
pred = e;
if ((e = e.next) == null)
break;
}
}
// 處理紅黑樹節(jié)點
else if (f instanceof TreeBin) {
validated = true;
TreeBin t = (TreeBin)f;
TreeNode r, p;
if ((r = t.root) != null && (p = r.findTreeNode(hash, key, null)) != null) {
V pv = p.val;
if (cv == null || cv == pv || (pv != null && cv.equals(pv))) {
oldVal = pv;
if (value != null)
p.val = value;
else if (t.removeTreeNode(p))
setTabAt(tab, i, untreeify(t.first));
}
}
}
else if (f instanceof ReservationNode)
throw new IllegalStateException("Recursive update");
}
}
if (validated) {
// 原值不為空則返回該值
if (oldVal != null) {
// 節(jié)點被刪除則更新計數(shù)
if (value == null)
addCount(-1L, -1);
return oldVal;
}
break;
}
}
}
return null;
}
}
computeIfAbsent 方法
先前我們提到過 ReservationNode 用于占位，那么我們就以 computeIfAbsent 方法為例，來簡單看一下它是怎么來占位的，其中大部分邏輯我們在上文中已經(jīng)介紹過，主要關(guān)注它的不同點：
public class ConcurrentHashMap extends AbstractMap implements ConcurrentMap, Serializable {
// 該方法用于某個鍵未被添加到哈希表中時，使用給定的函數(shù)計算值并將其添加到哈希表中
public V computeIfAbsent(K key, Function mappingFunction) {
if (key == null || mappingFunction == null)
throw new NullPointerException();
int h = spread(key.hashCode());
V val = null;
int binCount = 0;
for (Node[] tab = table; ; ) {
Node f;
int n, i, fh;
K fk;
V fv;
// 初始化操作
if (tab == null || (n = tab.length) == 0)
tab = initTable();
// 未在桶中匹配到元素
else if ((f = tabAt(tab, i = (n - 1) & h)) == null) {
// 創(chuàng)建占位符節(jié)點，并加鎖
Node r = new ReservationNode();
synchronized (r) {
// CAS 操作將桶中的 null 更新成占位節(jié)點
if (casTabAt(tab, i, null, r)) {
binCount = 1;
Node node = null;
try {
// 計算 key 對應(yīng)的 value
if ((val = mappingFunction.apply(key)) != null)
// value 不為 null 創(chuàng)建鏈表節(jié)點
node = new Node(h, key, val);
} finally {
// 將鏈表節(jié)點添加到桶中
setTabAt(tab, i, node);
}
}
}
if (binCount != 0)
break;
}
// 協(xié)助擴容
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
// 如果桶中的第一個節(jié)點 key 能和現(xiàn)在這個 key 完成匹配，那么直接返回它的值
else if (fh == h && ((fk = f.key) == key || (fk != null && key.equals(fk))) && (fv = f.val) != null)
return fv;
// 否則，遍歷尋找
else {
boolean added = false;
synchronized (f) {
if (tabAt(tab, i) == f) {
if (fh >= 0) {
binCount = 1;
for (Node e = f; ; ++binCount) {
K ek;
// 匹配到對應(yīng)節(jié)點，為 val 賦值并結(jié)束循環(huán)
if (e.hash == h && ((ek = e.key) == key || (ek != null && key.equals(ek)))) {
val = e.val;
break;
}
Node pred = e;
// 沒找到這個節(jié)點創(chuàng)建一個
if ((e = e.next) == null) {
if ((val = mappingFunction.apply(key)) != null) {
if (pred.next != null)
throw new IllegalStateException("Recursive update");
added = true;
pred.next = new Node(h, key, val);
}
break;
}
}
} else if (f instanceof TreeBin) {
binCount = 2;
TreeBin t = (TreeBin) f;
TreeNode r, p;
if ((r = t.root) != null &&
(p = r.findTreeNode(h, key, null)) != null)
val = p.val;
else if ((val = mappingFunction.apply(key)) != null) {
added = true;
t.putTreeVal(h, key, val);
}
} else if (f instanceof ReservationNode)
throw new IllegalStateException("Recursive update");
}
}
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (!added)
return val;
break;
}
}
}
if (val != null)
addCount(1L, binCount);
return val;
}
}
到這里 ConcurrentHashMap 中的源碼大部分已經(jīng)介紹完了，接下來我們簡單談一些有意思的問題。
key 和 value 不能為 null 的妙用
ConcurrentHashMap 與 HashMap 不同，它是不允許 key 和 value 為 null 的，這是為什么呢？根據(jù)源碼分析，我覺得主要原因是為了 簡化并發(fā)邏輯，提高處理效率，這樣當(dāng)遇到某桶中元素為 null 時便能判定此處無元素并不需要為 null 做特殊處理。當(dāng)然這樣做也能 避免歧義，比如我們在使用 get 方法獲取某 key 的值時，為 null 就表示該鍵值對不存在，而不會發(fā)生認(rèn)為這個 key 存在但 value 為 null 的情況。
ConcurrentHashMap 對代碼的編排
ConcurrentHashMap 為了增加可讀性，它首先排列了主要的靜態(tài)常量聲明和內(nèi)部常用的靜態(tài)方法，再聲明了字段，然后才是主要的公有方法，最后是通用的擴容方法、樹的定義、迭代器和批量操作方法。之后我們在寫代碼時也可以參考它對字段方法的排列。
That's all
審核編輯 黃宇
聲明：本文內(nèi)容及配圖由入駐作者撰寫或者入駐合作網(wǎng)站授權(quán)轉(zhuǎn)載。文章觀點僅代表作者本人，不代表電子發(fā)燒友網(wǎng)立場。文章及其配圖僅供工程師學(xué)習(xí)之用，如有內(nèi)容侵權(quán)或者其他違規(guī)問題，請聯(lián)系本站處理。
舉報投訴
-
哈希表
+關(guān)注
關(guān)注
0文章
10瀏覽量
4991 -
hashmap
+關(guān)注
關(guān)注
0文章
15瀏覽量
2482
發(fā)布評論請先 登錄
相關(guān)推薦
熱點推薦
串口DMA發(fā)送有緩存嗎？
串口DMA發(fā)送有緩存嗎， 我是從ringbuffer取出來，放到申請的緩存里，啟動串口DMA發(fā)送，然后就釋放了。暫時沒發(fā)現(xiàn)什么問題。
用的drv_usart.c是這個版本
發(fā)表于 10-10 06:14
緩存之美：萬文詳解 Caffeine 實現(xiàn)原理（上）
文章將采用“總-分-總”的結(jié)構(gòu)對配置固定大小元素驅(qū)逐策略的 Caffeine 緩存進(jìn)行介紹，首先會講解它的實現(xiàn)原理，在大家對它有一個概念之后再深入具體源碼的細(xì)節(jié)之中，理解它的設(shè)計理念，從中能學(xué)習(xí)到
harmony-utils之LRUCacheUtil，LRUCache緩存工具類
harmony-utils之LRUCacheUtil，LRUCache緩存工具類 harmony-utils 簡介與說明 harmony-utils 一款功能豐富且極易上手的HarmonyOS工具庫
高性能緩存設(shè)計：如何解決緩存偽共享問題
緩存行，引發(fā)無效化風(fēng)暴，使看似無關(guān)的變量操作拖慢整體效率。本文從緩存結(jié)構(gòu)原理出發(fā)，通過實驗代碼復(fù)現(xiàn)偽共享問題（耗時從3709ms優(yōu)化至473ms），解析其底層機制；同時深入剖析高性能緩存
由 Mybatis 源碼暢談軟件設(shè)計（八）：從根上理解 Mybatis 二級緩存
1. 驗證二級緩存 在上一篇帖子中的 User 和 Department 實體類依然要用，這里就不再贅述了，要啟用二級緩存，需要在 Mapper.xml 文件中指定 cache 標(biāo)簽，如下
MCU緩存設(shè)計
從Flash或外部存儲器讀取的指令，減少CPU因等待指令加載而停滯，適用于實時性要求高的場景（如中斷服務(wù)程序）。 D-Cache?：緩存從Flash、SRAM或外部存儲器讀取的數(shù)據(jù)，加速變量與堆棧的讀寫操作。 TCM（緊耦合內(nèi)存
Nginx緩存配置詳解
Nginx 是一個功能強大的 Web 服務(wù)器和反向代理服務(wù)器，它可以用于實現(xiàn)靜態(tài)內(nèi)容的緩存，緩存可以分為客戶端緩存和服務(wù)端緩存。
緩存對大數(shù)據(jù)處理的影響分析
緩存對大數(shù)據(jù)處理的影響顯著且重要，主要體現(xiàn)在以下幾個方面： 一、提高數(shù)據(jù)訪問速度 在大數(shù)據(jù)環(huán)境中，數(shù)據(jù)存儲通常采用分布式存儲系統(tǒng)，數(shù)據(jù)量龐大，直接從存儲系統(tǒng)中讀取數(shù)據(jù)會存在較高的延遲。而通過緩存技術(shù)
HTTP緩存頭的使用 本地緩存與遠(yuǎn)程緩存的區(qū)別
HTTP緩存頭是一組HTTP響應(yīng)頭，它們控制瀏覽器和中間代理服務(wù)器如何緩存網(wǎng)頁內(nèi)容。合理使用HTTP緩存頭可以顯著提高網(wǎng)站的加載速度和性能，減少服務(wù)器的負(fù)載。 1. HTTP緩存頭概述
緩存之美——如何選擇合適的本地緩存？
Guava cache是Google開發(fā)的Guava工具包中一套完善的JVM本地緩存框架，底層實現(xiàn)的數(shù)據(jù)結(jié)構(gòu)類似于ConcurrentHashMap，但是進(jìn)行了更多的能力拓展，包括緩存過期時間設(shè)置、
工商網(wǎng)監(jiān)
評論