文章
http://www.itdecent.cn/p/289cf670733e ，主要講 JDK 8
http://www.itdecent.cn/p/487d00afe6ca ， 將擴(kuò)容操作
http://www.itdecent.cn/p/7db1ac8395ee ， 1.7 ， 1.8 都有講
在 1.8 版本以前，ConcurrentHashMap 采用分段鎖的概念，使鎖更加細(xì)化，但是 1.8 已經(jīng)改變了這種思路，而是利用 CAS + Synchronized 來保證并發(fā)更新的安全，當(dāng)然底層采用數(shù)組 + 鏈表 + 紅黑樹的存儲結(jié)構(gòu)。
數(shù)據(jù)結(jié)構(gòu)
重要字段
/**
* races. Updated via CAS.
* 記錄容器的容量大小，通過CAS更新
*/
private transient volatile long baseCount;
/**
* 這個sizeCtl是volatile的，那么他是線程可見的，一個思考:它是所有修改都在CAS中進(jìn)行，但是sizeCtl為什么不設(shè)計成LongAdder(jdk8出現(xiàn)的)類型呢？
* 或者設(shè)計成AtomicLong(在高并發(fā)的情況下比LongAdder低效)，這樣就能減少自己操作CAS了。
*
* 來看下注釋，當(dāng)sizeCtl小于0說明有多個線程正則等待擴(kuò)容結(jié)果，參考transfer函數(shù)
*
* sizeCtl等于0是默認(rèn)值，大于0是擴(kuò)容的閥值
*/
private transient volatile int sizeCtl;
/**
* 自旋鎖 （鎖定通過 CAS） 在調(diào)整大小和/或創(chuàng)建 CounterCells 時使用。 在CounterCell類更新value中會使用，功能類似顯示鎖和內(nèi)置鎖，性能更好
* 在Striped64類也有應(yīng)用
*/
private transient volatile int cellsBusy;
還有最重要的節(jié)點類Node，注意val和next是volatile類型
static class Node<K,V> implements Map.Entry<K,V> {
final int hash;
final K key;
volatile V val;
volatile Node<K,V> next;
Node(int hash, K key, V val, Node<K,V> next) {
this.hash = hash;
this.key = key;
this.val = val;
this.next = next;
}
下面對重點的常量進(jìn)行說明：
table：用來存放Node節(jié)點數(shù)據(jù)的，默認(rèn)為null，默認(rèn)大小為16的數(shù)組，每次擴(kuò)容時大小總是2的冪次方
nextTable：擴(kuò)容時新生成的數(shù)據(jù)，數(shù)組為table的兩倍
Node：節(jié)點，保存key-value的數(shù)據(jù)結(jié)構(gòu)
ForwardingNode：一個特殊的Node節(jié)點，hash值為-1，其中存儲nextTable的引用。只有table發(fā)生擴(kuò)容的時候，F(xiàn)orwardingNode才會發(fā)揮作用，作為一個占位符放在table中表示當(dāng)前節(jié)點為null或則已經(jīng)被移動
-
sizeCtl：控制標(biāo)識符，用來控制table初始化和擴(kuò)容操作的，在不同的地方有不同的用途，其值也不同，所代表的含義也不同
負(fù)數(shù)代表正在進(jìn)行初始化或擴(kuò)容操作
-1代表正在初始化
-N 表示有N-1個線程正在進(jìn)行擴(kuò)容操作
正數(shù)或0代表hash表還沒有被初始化，這個數(shù)值表示初始化或下一次進(jìn)行擴(kuò)容的大小
重點內(nèi)部類
-
Node
Node 存儲 key-value 鍵值對，所有插入ConcurrentHashMap的中數(shù)據(jù)都將會包裝在Node中。
static class Node<K,V> implements Map.Entry<K,V> { final int hash; final K key; volatile V val; //帶有volatile，保證可見性 volatile Node<K,V> next; //下一個節(jié)點的指針 /** 不允許修改value的值 */ public final V setValue(V value) { throw new UnsupportedOperationException(); } /** 賦值get()方法 */ Node<K,V> find(int h, Object k) { Node<K,V> 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; } }在Node內(nèi)部類中，其屬性value、next都是帶有volatile的。同時其對value的setter方法進(jìn)行了特殊處理，不允許直接調(diào)用其setter方法來修改value的值。最后Node還提供了find方法來賦值map.get()。
-
TreeNode
在 1.8 的 ConcurrentHashMap 中，如果鏈表的數(shù)據(jù)過長是會轉(zhuǎn)換為紅黑樹來處理。當(dāng)它并不是直接轉(zhuǎn)換，而是將這些鏈表的節(jié)點包裝成TreeNode放在TreeBin對象中，然后由TreeBin完成紅黑樹的轉(zhuǎn)換。所以TreeNode也必須是ConcurrentHashMap的一個核心類，其為樹節(jié)點類。源碼展示TreeNode繼承Node，且提供了findTreeNode用來查找查找hash為h，key為k的節(jié)點。
-
TreeBin
該類并不負(fù)責(zé)key-value的鍵值對包裝，它用于在鏈表轉(zhuǎn)換為紅黑樹時包裝TreeNode節(jié)點，也就是說ConcurrentHashMap紅黑樹存放是TreeBin，不是TreeNode。該類封裝了一系列的方法，包括putTreeVal、lookRoot、UNlookRoot、remove、balanceInsetion、balanceDeletion。
-
ForwardingNode
這是一個真正的輔助類，該類僅僅只存活在ConcurrentHashMap擴(kuò)容操作時。只是一個標(biāo)志節(jié)點，并且指向nextTable，它提供find方法而已。該類也是集成Node節(jié)點，其hash為-1，key、value、next均為null。
構(gòu)造函數(shù)
初始化： initTable()
ConcurrentHashMap的初始化主要由initTable()方法實現(xiàn)，在上面的構(gòu)造函數(shù)中我們可以看到，其實ConcurrentHashMap在構(gòu)造函數(shù)中并沒有做什么事，僅僅只是設(shè)置了一些參數(shù)而已。其真正的初始化是發(fā)生在插入的時候，例如put、merge、compute、computeIfAbsent、computeIfPresent操作時。其方法定義如下：
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
//sizeCtl < 0 表示有其他線程在初始化，該線程必須掛起
if ((sc = sizeCtl) < 0)
Thread.yield();
// 如果該線程獲取了初始化的權(quán)利，則用CAS將sizeCtl設(shè)置為-1，表示本線程正在初始化
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
// 進(jìn)行初始化
try {
if ((tab = table) == null || tab.length == 0) {
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;
// 下次擴(kuò)容的大小
sc = n - (n >>> 2); ///相當(dāng)于0.75*n 設(shè)置一個擴(kuò)容的閾值
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}
初始化方法initTable()的關(guān)鍵就在于sizeCtl，該值默認(rèn)為0，如果在構(gòu)造函數(shù)時有參數(shù)傳入該值則為2的冪次方。該值如果 < 0，表示有其他線程正在初始化，則必須暫停該線程。如果線程獲得了初始化的權(quán)限則先將sizeCtl設(shè)置為-1，防止有其他線程進(jìn)入，最后將sizeCtl設(shè)置0.75 * n，表示擴(kuò)容的閾值。
put
核心思想依然是根據(jù)hash值計算節(jié)點插入在table的位置，如果該位置為空，則直接插入，否則插入到鏈表或者樹中。但是ConcurrentHashMap會涉及到多線程情況就會復(fù)雜很多。
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值
int hash = spread(key.hashCode());
int binCount = 0;
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
// table為null，進(jìn)行初始化工作
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//如果i位置沒有節(jié)點，則直接插入，不需要加鎖
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
break; // no lock when adding to empty bin
}
// 有線程正在進(jìn)行擴(kuò)容操作，則先幫助擴(kuò)容
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
else {
V oldVal = null;
//對該節(jié)點進(jìn)行加鎖處理（hash值相同的鏈表的頭節(jié)點），對性能有點兒影響
synchronized (f) {
if (tabAt(tab, i) == f) {
//fh > 0 表示為鏈表，將該節(jié)點插入到鏈表尾部
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;
//putIfAbsent()
if (!onlyIfAbsent)
e.val = value;
break;
}
Node<K,V> pred = e;
//鏈表尾部 直接插入
if ((e = e.next) == null) {
pred.next = new Node<K,V>(hash, key,
value, null);
break;
}
}
}
//樹節(jié)點，按照樹的插入操作進(jìn)行插入
else if (f instanceof TreeBin) {
Node<K,V> p;
binCount = 2;
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
if (binCount != 0) {
// 如果鏈表長度已經(jīng)達(dá)到臨界值8 就需要把鏈表轉(zhuǎn)換為樹結(jié)構(gòu)
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
//size + 1
addCount(1L, binCount);
return null;
}
按照上面的源碼，我們可以確定put整個流程如下：
判空；ConcurrentHashMap的key、value都不允許為null
計算hash。利用方法計算hash值。
-
遍歷table，進(jìn)行節(jié)點插入操作，過程如下：
如果table為空，則表示ConcurrentHashMap還沒有初始化，則進(jìn)行初始化操作：initTable()
根據(jù)hash值獲取節(jié)點的位置i，若該位置為空，則直接插入，這個過程是不需要加鎖的。計算f位置：i=(n - 1) & hash
如果檢測到fh = f.hash == -1，則f是ForwardingNode節(jié)點，表示有其他線程正在進(jìn)行擴(kuò)容操作，則幫助線程一起進(jìn)行擴(kuò)容操作
如果f.hash >= 0 表示是鏈表結(jié)構(gòu)，則遍歷鏈表，如果存在當(dāng)前key節(jié)點則替換value，否則插入到鏈表尾部。如果f是TreeBin類型節(jié)點，則按照紅黑樹的方法更新或者增加節(jié)點
若鏈表長度 > TREEIFY_THRESHOLD(默認(rèn)是8)，則將鏈表轉(zhuǎn)換為紅黑樹結(jié)構(gòu), 此時還需要進(jìn)一步判斷，并不是直接轉(zhuǎn)換的（當(dāng)數(shù)組大小已經(jīng)超過64并且鏈表中的元素個數(shù)超過默認(rèn)設(shè)定（8個）時，將鏈表轉(zhuǎn)化為紅黑樹）
調(diào)用addCount方法，ConcurrentHashMap的size + 1，此方法還會判斷是否需要擴(kuò)容。
這里整個put操作已經(jīng)完成。
get
get操作的整個邏輯非常清楚： - 計算hash值 - 判斷table是否為空，如果為空，直接返回null - 根據(jù)hash值獲取table中的Node節(jié)點（tabAt(tab, (n - 1) & h)），然后根據(jù)鏈表或者樹形方式找到相對應(yīng)的節(jié)點，返回其value值。
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
// 計算hash
int h = spread(key.hashCode());
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
// 搜索到的節(jié)點key與傳入的key相同且不為null,直接返回這個節(jié)點
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
// 樹
else if (eh < 0)
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;
}
remove
size
ConcurrentHashMap的size()方法我們雖然用得不是很多，但是我們還是很有必要去了解的。ConcurrentHashMap的size()方法返回的是一個不精確的值，因為在進(jìn)行統(tǒng)計的時候有其他線程正在進(jìn)行插入和刪除操作。當(dāng)然為了這個不精確的值，ConcurrentHashMap也是操碎了心。
為了更好地統(tǒng)計size，ConcurrentHashMap提供了baseCount、counterCells兩個輔助變量和一個CounterCell輔助內(nèi)部類。
@sun.misc.Contended static final class CounterCell {
volatile long value;
CounterCell(long x) { value = x; }
}
//ConcurrentHashMap中元素個數(shù),但返回的不一定是當(dāng)前Map的真實元素個數(shù)?；贑AS無鎖更新
private transient volatile long baseCount;
private transient volatile CounterCell[] counterCells;
size()方法定義如下：
public int size() {
long n = sumCount();
return ((n < 0L) ? 0 :
(n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
(int)n);
}
內(nèi)部調(diào)用sunmCount()：
final long sumCount() {
CounterCell[] as = counterCells; CounterCell a;
long sum = baseCount;
if (as != null) {
for (int i = 0; i < as.length; ++i) {
//遍歷，所有counter求和
if ((a = as[i]) != null)
sum += a.value;
}
}
return sum;
}
sumCount()就是迭代counterCells來統(tǒng)計sum的過程。我們知道put操作時，肯定會影響size()，我們就來看看CouncurrentHashMap是如何為了這個不和諧的size()操碎了心。
在put()方法最后會調(diào)用addCount()方法，該方法主要做兩件事，一件更新baseCount的值，第二件檢測是否進(jìn)行擴(kuò)容，我們只看更新baseCount部分：
private final void addCount(long x, int check) {
CounterCell[] as; long b, s;
// s = b + x，完成baseCount++操作；
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))) {
// 多線程CAS發(fā)生失敗時執(zhí)行
fullAddCount(x, uncontended);
return;
}
if (check <= 1)
return;
s = sumCount();
}
// 檢查是否進(jìn)行擴(kuò)容
}
x == 1，如果counterCells == null，則U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)，如果并發(fā)競爭比較大可能會導(dǎo)致改過程失敗，如果失敗則最終會調(diào)用 fullAddCount() 方法。
其實為了提高高并發(fā)的時候 baseCount 可見性的失敗問題，又避免一直重試，JDK 8 引入了類 Striped64,其中 LongAdder 和 DoubleAdder 都是基于該類實現(xiàn)的，而 CounterCell 也是基于 Striped64 實現(xiàn)的。如果counterCells ！= null，且uncontended = U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x)也失敗了，同樣會調(diào)用 fullAddCount() 方法，最后調(diào)用 sumCount() 計算 s。
其實在1.8中，它不推薦 size() 方法，而是推崇 mappingCount() 方法，該方法的定義和 size() 方法基本一致：
public long mappingCount() {
long n = sumCount();
return (n < 0L) ? 0L : n; // ignore transient negative values
}
擴(kuò)容
什么時候擴(kuò)容
當(dāng)前容量超過閾值
當(dāng)鏈表中元素個數(shù)超過默認(rèn)設(shè)定（8個），當(dāng)數(shù)組的大小還未超過64的時候，此時進(jìn)行數(shù)組的擴(kuò)容，如果超過則將鏈表轉(zhuǎn)化成紅黑樹
其他線程在擴(kuò)容時幫助擴(kuò)容
擴(kuò)容相關(guān)的屬性
采用多線程擴(kuò)容。整個擴(kuò)容過程，通過CAS設(shè)置sizeCtl，transferIndex等變量協(xié)調(diào)多個線程進(jìn)行并發(fā)擴(kuò)容。
nextTable
擴(kuò)容期間，將table數(shù)組中的元素 遷移到 nextTable。
sizeCtl屬性
多線程之間，以volatile的方式讀取sizeCtl屬性，來判斷ConcurrentHashMap當(dāng)前所處的狀態(tài)。通過cas設(shè)置sizeCtl屬性，告知其他線程ConcurrentHashMap的狀態(tài)變更。
不同狀態(tài)，sizeCtl所代表的含義也有所不同。
未初始化：
sizeCtl=0：表示沒有指定初始容量。
sizeCtl>0：表示初始容量。
初始化中：
sizeCtl=-1,標(biāo)記作用，告知其他線程，正在初始化
正常狀態(tài)：
sizeCtl=0.75n ,擴(kuò)容閾值
擴(kuò)容中:
sizeCtl < 0 : 表示有其他線程正在執(zhí)行擴(kuò)容
sizeCtl = (resizeStamp(n) << RESIZE_STAMP_SHIFT) + 2 :表示此時只有一個線程在執(zhí)行擴(kuò)容
transferIndex屬性
private transient volatile int transferIndex;
/**
擴(kuò)容線程每次最少要遷移16個hash桶
*/
private static final int MIN_TRANSFER_STRIDE = 16;
擴(kuò)容索引，表示已經(jīng)分配給擴(kuò)容線程的table數(shù)組索引位置。主要用來協(xié)調(diào)多個線程，并發(fā)安全地獲取遷移任務(wù)（hash桶）。
ForwardingNode節(jié)點
標(biāo)記作用，表示其他線程正在擴(kuò)容，并且此節(jié)點已經(jīng)擴(kuò)容完畢
關(guān)聯(lián)了nextTable,擴(kuò)容期間可以通過find方法，訪問已經(jīng)遷移到了nextTable中的數(shù)據(jù)
具體實現(xiàn)
- 在擴(kuò)容之前，transferIndex 在數(shù)組的最右邊 。此時有一個線程發(fā)現(xiàn)已經(jīng)到達(dá)擴(kuò)容閾值，準(zhǔn)備開始擴(kuò)容。
2 擴(kuò)容線程，在遷移數(shù)據(jù)之前，首先要將transferIndex左移（以cas的方式修改 transferIndex=transferIndex-stride(要遷移hash桶的個數(shù))），獲取遷移任務(wù)。每個擴(kuò)容線程都會通過for循環(huán)+CAS的方式設(shè)置transferIndex，因此可以確保多線程擴(kuò)容的并發(fā)安全。
換個角度，我們可以將待遷移的table數(shù)組，看成一個任務(wù)隊列，transferIndex看成任務(wù)隊列的頭指針。而擴(kuò)容線程，就是這個隊列的消費者。擴(kuò)容線程通過CAS設(shè)置transferIndex索引的過程，就是消費者從任務(wù)隊列中獲取任務(wù)的過程。為了性能考慮，我們當(dāng)然不會每次只獲取一個任務(wù)（hash桶），因此ConcurrentHashMap規(guī)定，每次至少要獲取16個遷移任務(wù)（遷移16個hash桶，MIN_TRANSFER_STRIDE = 16）
cas設(shè)置transferIndex的源碼如下：
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
//計算每次遷移的node個數(shù)
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // 確保每次遷移的node個數(shù)不少于16個
...
for (int i = 0, bound = 0;;) {
...
//cas無鎖算法設(shè)置 transferIndex = transferIndex - stride
if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
...
...
}
...//省略遷移邏輯
}
}
過程分析
- 線程執(zhí)行put操作，發(fā)現(xiàn)容量已經(jīng)達(dá)到擴(kuò)容閾值，需要進(jìn)行擴(kuò)容操作，此時transferindex=tab.length=32
- 擴(kuò)容線程A 以cas的方式修改transferindex=31-16=16 ,然后按照降序遷移table[31]--table[16]這個區(qū)間的hash桶
- 遷移hash桶時，會將桶內(nèi)的鏈表或者紅黑樹，按照一定算法，拆分成2份，將其插入nextTable[i]和nextTable[i+n]（n是table數(shù)組的長度）。 遷移完畢的hash桶,會被設(shè)置成ForwardingNode節(jié)點，以此告知訪問此桶的其他線程，此節(jié)點已經(jīng)遷移完畢。
- 此時線程2訪問到了ForwardingNode節(jié)點，如果線程2執(zhí)行的put或remove等寫操作，那么就會先幫其擴(kuò)容。如果線程2執(zhí)行的是get等讀方法，則會調(diào)用ForwardingNode的find方法，去nextTable里面查找相關(guān)元素。
- 線程2加入擴(kuò)容操作
-
如果準(zhǔn)備加入擴(kuò)容的線程，發(fā)現(xiàn)以下情況，放棄擴(kuò)容，直接返回。
發(fā)現(xiàn)transferIndex=0,即所有node均已分配
發(fā)現(xiàn)擴(kuò)容線程已經(jīng)達(dá)到最大擴(kuò)容線程數(shù)
image
