ConcurentHashMap設計與原始碼分析
阿新 • • 發佈：2021-07-08
簡介
ConcurentHashMap是java.util.concurrent包下的一個執行緒安全的類，繼承自Map類，用於儲存具有鍵（key）、值（value）對映關係的雙列集合。其資料結構與HashMap類似，都是使用陣列+連結串列+樹（紅黑樹）的結構實現。
優點
- 執行緒安全，在高併發情況下與HashTable相比效率更高（HashMap Vs. ConcurrentHashMap Vs. HashTable）
- 在使用Iterator迭代時不會丟擲ConcurrentModificationException異常（fail-fast機制）
資料結構
ConcurentHashMap底層使用陣列加連結串列的形式儲存，K-V通過內部類Node包裝，當連結串列長度大於8時，轉換為樹節點（TreeNode），超過64時改用紅黑樹（一種自平衡二叉查詢樹）
ConcurentHashMap資料結構的實現主要通過Node、TreeNode、TreeBin等內部類實現，其UML圖如下：
Node
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; } }
Node類實現了Entry介面，用於儲存節點的hash（雜湊值）、key（鍵）、value（值）以及next（下一個節點的地址）四個屬性。
TreeNode
TreeNode繼承了Node類，用於儲存ConcurentHashMap中的樹結構，其構造方法如下：
TreeNode(int hash, K key, V val, Node<K,V> next,
TreeNode<K,V> parent) {
super(hash, key, val, next);
this.parent = parent;
}
在構造方法中，TreeNode呼叫了Node的構造方法，並指定了該節點的父節點。
TreeBin
TreeBin用於包裝TreeNode類，當連結串列過長時，TreeBin會把TreeNode轉換為紅黑樹。事實上，在ConcurentHashMap的“陣列”中（也就是樹的根節點）所儲存的並不是TreeNode而是TreeBin。TreeBin不儲存key/value，TreeBin還維護了一個讀寫鎖，使得讀必須等待寫操作完成才能進行。
ForwardingNode
ForwardingNode用於標記正在遷移中的Node。在其構造方法會生成一個key、value 和 next 都為 null，且 hash 為 MOVED 的 Node。
static final class ForwardingNode<K,V> extends Node<K,V> {
final Node<K, V>[] nextTable;
ForwardingNode(Node<K, V>[] tab) {
super(MOVED, null, null, null);
this.nextTable = tab;
}
}
結構示意圖
執行緒安全
1、CAS機制
對節點的修改操作都通過CAS來完成，CAS機制實現了無鎖化的修改值的操作，可以大大降低鎖代理的效能消耗。
2、volatile關鍵字
在ConcurentHashMap中，部分變數使用了volatile關鍵字修飾，保證了變數的可見性和指令的有序性。例如對節點操作進行控制的sizeCtl變數，在Node類中的val、next變數。
3、synchronized
ConcurentHashMap需要使用synchronized對陣列中的的非空節點進行加鎖操作（空節點可通過CAS直接進行操作，不需要加鎖），如put方法及transfer方法。
4、Unsafe類和三個tabAt方法
Java是無法對作業系統底層進行操作的，所以CAS等操作的具體實現都需要Unsafe類以對底層進行操作。而對於節點的取值、設值、修改等操作，ConcurentHashMap基於Unsafe類封裝了三個tabAt方法。
/**
* ((long)i << ASHIFT) + ABASE用於計算出元素的真實地址
* ASHIFT為每個節點（Node）的偏移量（位數）
* ABASE為頭節點的地址（arrayBaseOffset）
*/
// 獲得在i位置上的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);
}
// 利用CAS演算法設定i位置上的Node節點
static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,
Node<K,V> c, Node<K,V> v) {
return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
}
// 設定節點位置的值
static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
}
方法詳解
建構函式
ConcurentHashMap有多個過載的構造方法，可傳入三個引數
- (int) initialCapacity 指ConcurrentHashMap的初始容量
- (float) loadFactor 載入因子
- (int) concurrencyLevel 併發度
在Java7中，ConcurentHashMap使用Segment分片的形式實現，Segment之間允許執行緒進行併發操作，而concurrencyLevel則是用來設定Segment[]陣列長度的，concurrencyLevel的最小2次冪便為實際併發度。
而在Java8中，ConcurentHashMap摒棄了Segment，改用CAS加上TreeBin等輔助類實現，併發度concurrencyLevel也就沒有實際意義了。
public ConcurrentHashMap(int initialCapacity) {
if (initialCapacity < 0)
throw new IllegalArgumentException();
// MAXIMUM_CAPACITY = 1 << 30（2^30 = 1073741824）
// 如果大小為MAXIMUM_CAPACITY最大總量的一半，那麼直接將容量設為MAXIMUM_CAPACITY，否則計算最小冪次方
int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
MAXIMUM_CAPACITY :
// 1.5 * initialCapacity + 1
tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
this.sizeCtl = cap;
}
建構函式進行了sizeCtl的賦值，sizeCtl作為控制識別符號，不同的數值代表不同的意義
- -1代表正在初始化
- -N表示有N-1個執行緒正在進行擴容操作
- hash表還沒有被初始化時，該數值表示初始化或下一次進行擴容的大小。
- 初始化之後，它的值始終是當前ConcurrentHashMap容量的0.75倍
initTable初始化
初始化一個容量為sizeCtl的Node陣列
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(); // lost initialization race; just spin
// 利用CAS方法把sizectl的值置為-1，表示本執行緒正在進行初始化，防止其他執行緒進入
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
if ((tab = table) == null || tab.length == 0) {
// 預設大小16
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];// 初始化Node
table = tab = nt;
// 設定一個擴容的閾值 相當於0.75*n
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}
互斥同步進入阻塞狀態需要很大的開銷，initTable方法使用了自旋鎖，通過Thread.yield()使執行緒讓步，然後忙迴圈直到sizeCtl滿足條件
tableSizeFor函式詳解
Returns a power of two table size for the given desired capacity.
返回大於輸入引數且最近的2的整數次冪的數
/**
* 使最高位的1後面的位全變為1，最後再讓結果n+1，即得到了2的整數次冪的值
*/
private static final int tableSizeFor(int c) {
int n = c - 1;
n |= n >>> 1;
n |= n >>> 2;
n |= n >>> 4;
n |= n >>> 8;
n |= n >>> 16;
return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;
}
put
public V put(K key, V value) {
return putVal(key, value, false);
}
final V putVal(K key, V value, boolean onlyIfAbsent) {
// 判空，key和value不能為空
if (key == null || value == null) throw new NullPointerException();
// spread將較高的雜湊值擴充套件為較低的雜湊值，並將最高位強制為0
int hash = spread(key.hashCode());
// binCount用於記錄相應連結串列的長度
int binCount = 0;
// 死迴圈
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
// tab為空，初始化table
if (tab == null || (n = tab.length) == 0)
tab = initTable();
// 根據hash值計算出在table裡面的位置，若該位置的值為空，直接放入元素
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
}
// 存在節點，說明發生了hash碰撞，需要對錶進行擴容
// 如果該位置的節點存在值且為MOVED(-1)，說明正在擴容
else if ((fh = f.hash) == MOVED)
// helpTransfer方法用於增加執行緒以協助擴容
tab = helpTransfer(tab, f);
else {
V oldVal = null;
// 節點上鎖
synchronized (f) {
if (tabAt(tab, i) == f) {
// fh > 0 說明這個節點是一個連結串列的節點 不是樹的節點
if (fh >= 0) {
binCount = 1;
// 遍歷節點
for (Node<K,V> e = f;; ++binCount) {
K ek;
// 存在該key，替換其值
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;
// 不存在該key，插入新Node
if ((e = e.next) == null) {
pred.next = new Node<K,V>(hash, key,
value, null);
break;
}
}
}
// 樹節點
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) {
// TREEIFY_THRESHOLD = 8
// 連結串列長度大於8，轉換為樹節點
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
// 將當前ConcurrentHashMap的元素數量+1
addCount(1L, binCount);
return null;
}
putVal()方法首先獲取到key的hash值，該key所對應位置的值為空，直接放入，若該鍵對應的位置存在節點，則判斷是否該節點為連結串列節點還是樹節點，再使用其相應的方法將Node放入
putVal()方法大概的流程圖如下：
get方法
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) {
// 雜湊值相等，返回該節點的值
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
// TreeBin或ForwardingNode，呼叫find方法
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;
}
treeifyBin
Replaces all linked nodes in bin at given index unless table is too small, in which case resizes instead.
將所有索引處的連結串列節點替換為二叉樹，如果表太小則改為調整大小
private final void treeifyBin(Node<K,V>[] tab, int index) {
Node<K,V> b; int n, sc;
if (tab != null) {
// 連結串列tab的長度小於64（MIN_TREEIFY_CAPACITY=64）時，進行擴容
if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
// 呼叫tryPresize進行擴容（所傳引數n即連結串列長度 * 2）
tryPresize(n << 1);
else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
synchronized (b) {
if (tabAt(tab, index) == b) {
TreeNode<K,V> hd = null, tl = null;
// 遍歷連結串列，建立紅黑樹
for (Node<K,V> e = b; e != null; e = e.next) {
TreeNode<K,V> p =
new TreeNode<K,V>(e.hash, e.key, e.val,
null, null);
if ((p.prev = tl) == null)
hd = p;
else
tl.next = p;
tl = p;
}
setTabAt(tab, index, new TreeBin<K,V>(hd));
}
}
}
}
}
tryPresize
Tries to presize table to accommodate the given number of elements.
嘗試調整表的大小以適應給定的元素數量。
private final void tryPresize(int size) {
int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :
tableSizeFor(size + (size >>> 1) + 1);
int sc;
// 正在擴容
while ((sc = sizeCtl) >= 0) {
Node<K,V>[] tab = table; int n;
// tab未初始化，進行初始化，與initTable類似
if (tab == null || (n = tab.length) == 0) {
n = (sc > c) ? sc : c;
if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
if (table == tab) {
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = nt;
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
}
}
// 如果擴容大小沒有達到閾值，或者超過最大容量
else if (c <= sc || n >= MAXIMUM_CAPACITY)
break;
// 呼叫transfer()擴容
else if (tab == table) {
int rs = resizeStamp(n);
if (sc < 0) {
Node<K,V>[] nt;
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);
}
}
}
transfer
Moves and/or copies the nodes in each bin to new table.
將樹中的節點移動和複製到新表中
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
// NCPU為可用的CPU執行緒數
// stride可以理解為步長，最小值為16
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
// nextTab為null，進行初始化
if (nextTab == null) { // initiating
try {
// 容量*2的節點陣列
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
nextTab = nt;
} catch (Throwable ex) { // try to cope with OOME
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;
// transferIndex用於控制遷移的位置
transferIndex = n;
}
int nextn = nextTab.length;
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
// advance為true說明這個節點已經處理過
boolean advance = true;
boolean finishing = false; // to ensure sweep before committing nextTab
// i 是位置索引，bound 是邊界，從後往前移
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
// 此處的while迴圈用作遍歷hash原表中的節點
while (advance) {
int nextIndex, nextBound;
// 處理從bound到i的節點
if (--i >= bound || finishing)
advance = false;
// transferIndex 小於等於 0，說明原hash表的所有位置都有相應的執行緒去處理了
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
// i指向transferIndex，bound指向(transferIndex-stride)
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
bound = nextBound;
i = nextIndex - 1;
advance = false;
}
}
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
// 遷移完成，nextTab賦值給table
if (finishing) {
nextTable = null;
table = nextTab;
// 重新計算sizeCtl，得出的值是新陣列長度的 0.75 倍
sizeCtl = (n << 1) - (n >>> 1);
return;
}
// 任務完成，使用CAS將sizeCtl減1
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
return;
finishing = advance = true;
i = n; // recheck before commit
}
}
// 位置 i 為空，放入空節點fwd
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
// 節點為ForwardingNode，表示已遷移
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
else {
// 加鎖
synchronized (f) {
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;
// 連結串列節點
if (fh >= 0) {
int runBit = fh & 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;
}
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);
}
// 在i位置放入一個連結串列
setTabAt(nextTab, i, ln);
// 在i+n位置放入另一個連結串列
setTabAt(nextTab, i + n, hn);
// 在table的i位置上插入forwardNode節點， 表示已經處理過該節點
setTabAt(tab, i, fwd);
advance = true;
}
// 樹結構
else if (f instanceof TreeBin) {
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;
}
}
// 小於UNTREEIFY_THRESHOLD(6)時轉為連結串列
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);
advance = true;
}
}
}
}
}
}
還有些事
1、為什麼要取最小二次冪
因為HashMap通過對hash值的i = (n - 1) & hash運算實現均勻分佈，若n不為2的次冪數，就不能保證均勻分佈。
參考文章