九張圖深入剖析ConcurrentHashMap
前言
在日常的開發中，我們經常使用key-value鍵值對的HashMap，其使用哈希表實現，用空間換取時間，提升查詢性能，但在多線程的并發場景中，HashMap并不是線程安全的。如果想使用線程安全的，可以使用ConcurrentHashMap、HashTable、Collections.synchronizedMap等。
但由于后面二者使用synchronized的粒度太大，因此一般不使用，而使用并發包中的ConcurrentHashMap在ConcurrentHashMap中，使用volatile保證內存可見性，使得讀場景下不需要“加鎖”保證原子性。
在寫場景下使用CAS+synchronized，synchronized只鎖哈希表某個索引位置上的首節點，相當于細粒度加鎖，增大并發性能。
本篇文章將從ConcurrentHashMap的使用，讀、寫、擴容的實現原理，設計思想等方面進行剖析查看本文前需要了解哈希表、volatile、CAS、synchronized等知識。
使用ConcurrentHashMap
ConcurrentHashMap是并發場景下線程安全的Map，可以在并發場景下查詢存儲K、V鍵值對不可變對象是絕對線程安全的，無論外界如何使用，都線程安全。
ConcurrentHashMap并不是絕對線程安全的，只提供方法的線程安全，如果在外層使用錯誤依舊會導致線程不安全
來看下面的案例，使用value存儲自增調用次數，開啟10個線程每個執行100次，最終結果應該是1000次，但錯誤的使用導致不足1000
public void test() {
// Map<String, Integer> map = new HashMap(16);
Map<String, Integer> map = new ConcurrentHashMap(16);
String key = "key";
CountDownLatch countDownLatch = new CountDownLatch(10);
for (int i = 0; i < 10; i++) {
new Thread(() -> {
for (int j = 0; j < 100; j++) {
incr(map, key);
// incrCompute(map, key);
}
countDownLatch.countDown();
}).start();
}
try {
//阻塞到線程跑完
countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
//1000不到
System.out.println(map.get(key));
}
private void incr(Map<String, Integer> map, String key) {
map.put(key, map.getOrDefault(key, 0) + 1);
}在自增方法incr中，先進行讀操作，再計算，最后進行寫操作，這種復合操作沒有保證原子性，導致最終所有結果累加一定不為1000，正確的使用方式是使用JDK8提供的默認方法compute
ConcurrentHashMap實現compute的原理是在put中使用同步手段后再進行計算。
private void incrCompute(Map<String, Integer> map, String key) {
map.compute(key, (k, v) -> Objects.isNull(v) ? 1 : v + 1);
}數據結構
與HashMap類似，使用哈希表+鏈表/紅黑樹實現。
哈希表
哈希表的實現由數組構成，當發生哈希沖突（哈希算法得到同一索引）時使用鏈地址法構建成鏈表。
當鏈表上的節點太長，遍歷尋找開銷大，超過閾值時（鏈表節點超過8個、哈希表長度大于64），樹化成紅黑樹減少遍歷尋找開銷，時間復雜度從O（n）優化為（log n）。
ConcurrentHashMap由Node數組構成，在擴容時會存在新舊兩個哈希表：table、nextTable。
public class ConcurrentHashMap<K,V> extends AbstractMap<K,V>
implements ConcurrentMap<K,V>, Serializable {
//哈希表 node數組
transient volatile Node<K,V>[] table;
//擴容時為了兼容讀寫，會存在兩個哈希表，這個是新哈希表
private transient volatile Node<K,V>[] nextTable;
// 默認為 0
// 當初始化時, 為 -1
// 當擴容時, 為 -(1 + 擴容線程數)
// 當初始化或擴容完成后，為 下一次的擴容的閾值大小
private transient volatile int sizeCtl;
//擴容時 用于指定遷移區間的下標
private transient volatile int transferIndex;
//統計每個哈希槽中的元素數量
private transient volatile CounterCell[] counterCells;
}節點
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;
}
// 擴容時如果某個 bin 遷移完畢, 用 ForwardingNode 作為舊 table bin 的頭結點
static final class ForwardingNode<K,V> extends Node<K,V> {}
// 用在 compute 以及 computeIfAbsent 時, 用來占位, 計算完成后替換為普通 Node
static final class ReservationNode<K,V> extends Node<K,V> {}
// 作為 treebin 的頭節點, 存儲 root 和 first
static final class TreeBin<K,V> extends Node<K,V> {}
// 作為 treebin 的節點, 存儲 parent, left, right
static final class TreeNode<K,V> extends Node<K,V> {}節點哈希值
//轉發節點
static final int MOVED = -1;
//紅黑樹在數組中的節點
static final int TREEBIN = -2;
//占位節點
static final int RESERVED = -3;轉發節點：繼承Node，用于擴容時設置在舊哈希表某索引的首節點，遇到轉發節點要去新哈希表中尋找。
static final class ForwardingNode<K,V> extends Node<K,V> {
//新哈希表
final Node<K,V>[] nextTable;
ForwardingNode(Node<K,V>[] tab) {
//哈希值設置為-1
super(MOVED, null, null, null);
this.nextTable = tab;
}
}紅黑樹在數組中的節點 TreeBin：繼承Node，first指向紅黑樹的首節點。
static final class TreeBin<K,V> extends Node<K,V> {
TreeNode<K,V> root;
//紅黑樹首節點
volatile TreeNode<K,V> first;
}
紅黑樹節點TreeNode
static final class TreeNode<K,V> extends Node<K,V> {
TreeNode<K,V> parent;
TreeNode<K,V> left;
TreeNode<K,V> right;
TreeNode<K,V> prev;
boolean red;
}占位節點：繼承Node，需要計算時（使用computer方法），先使用占位節點占位，計算完再構建節點取代占位節點
static final class ReservationNode<K,V> extends Node<K,V> {
ReservationNode() {
super(RESERVED, null, null, null);
}
Node<K,V> find(int h, Object k) {
return null;
}
}實現原理
構造
在構造時會檢查入參，然后根據需要存儲的數據容量、負載因子計算哈希表容量，最后將哈希表容量調整成2次冪。
構造時并不會初始化，而是等到使用再進行創建（懶加載）
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {
//檢查負載因子、初始容量
if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
//concurrencyLevel：1
if (initialCapacity < concurrencyLevel) // Use at least as many bins
initialCapacity = concurrencyLevel; // as estimated threads
//計算大小 = 容量/負載因子 向上取整
long size = (long)(1.0 + (long)initialCapacity / loadFactor);
//如果超過最大值就使用最大值
//tableSizeFor 將大小調整為2次冪
int cap = (size >= (long)MAXIMUM_CAPACITY) ?
MAXIMUM_CAPACITY : tableSizeFor((int)size);
//設置容量
this.sizeCtl = cap;
}讀-get
1.讀場景使用volatile保證可見性，即使其他線程修改也是可見的，不用使用其他手段保證同步
讀操作需要在哈希表中尋找元素，經過擾動算法打亂哈希值，再使用哈希值通過哈希算法得到索引，根據索引上的首節點分為多種情況處理
擾動算法將哈希值充分打亂（避免造成太多的哈希沖突），符號位&0保證結果正數。
int h = spread(key.hashCode())
擾動算法：哈希值高低16位異或運算
經過擾動算法后，&HASH_BITS = 0x7fffffff （011111...），符號位為0保證結果為正數。
負數的哈希值表示特殊的作用，比如：轉發節點、樹的首節點、占位節點等。
static final int spread(int h) {
return (h ^ (h >>> 16)) & HASH_BITS;
}2.使用打亂的哈希值經過哈希算法得到數組中的索引（下標）
n 為哈希表長度：(n = tab.length)
(e = tabAt(tab, (n - 1) & h)
h為計算后的哈希值，哈希值%（哈希表長度-1） 就能求出索引位置，為了性能提升，規定哈希表長度為2的n次冪，哈希表長度二進制一定是1000....，而(n-1)的二進制一定是0111...因此(n - 1) & h計算索引，進行與運算的結果一定在0~n-1之間 使用位運算提升性能
3.得到數組上的節點后，需要進行比較
找到哈希表上的首個節點后，進行比較key 查看是否是當前節點
比較規則：先對哈希值進行比較，如果對象哈希值相同，那么可能是同一個對象，還需要比較key（==與equals），如果哈希值都不相同，那么肯定不是同一個對象
先比較哈希值的好處就是提升查找性能，如果直接使用equals 可能時間復雜度會上升（比如String的equals）
4.使用鏈地址法解決哈希沖突，因此找到節點后可能遍歷鏈表或樹；由于哈希表存在擴容，也可能要去新節點上尋找
4.1 首節點比較成功，直接返回
4.2 首節點哈希值為負，說明該節點是特殊情況的:轉發節點、樹的首節點 、計算的預訂占位節點
- 如果是轉發節點，正在擴容則去新數組上找
- 如果是TreeBin則去紅黑樹中尋找
- 如果是占位節點 直接返回空
4.3 遍歷該鏈表依次比較
get代碼
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
//1.spread：擾動算法 + 讓key的哈希值不能為負數,因為負數哈希值代表紅黑樹或ForwardingNode
int h = spread(key.hashCode());
//2.(n - 1) & h：下標、索引 實際上就是數組長度模哈希值 位運算效率更高
//e：哈希表中對應索引位置上的節點
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
//3.如果哈希值相等,說明可能找到,再比較key
if ((eh = e.hash) == h) {
//4.1 key相等說明找到 返回
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
//4.2 首節點哈希值為負，說明該節點是轉發節點，當前正在擴容則去新數組上找
else if (eh < 0)
return (p = e.find(h, key)) != null ? p.val : null;
//4.3 遍歷該鏈表,能找到就返回值,不能返回null
while ((e = e.next) != null) {
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
}寫-put
添加元素時，使用CAS+synchronized（只鎖住哈希表中某個首節點）的同步方式保證原子性
- 獲取哈希值：擾動算法+確保哈希值為正數
- 哈希表為空，CAS保證一個線程初始化
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
//小于0 說明其他線程在初始化 讓出CPU時間片 后續初始化完退出
if ((sc = sizeCtl) < 0)
Thread.yield();
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
//CAS將SIZECTL設置成-1 （表示有線程在初始化）成功后 初始化
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;
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}
- 將哈希值通過哈希算法獲取索引上的節點 f = tabAt(tab, i = (n - 1) & hash)
- 根據不同情況進行處理
4.1 首節點為空時，直接CAS往索引位置添加節點 casTabAt(tab, i, null,new Node<K,V>(hash, key, value, null))
4.2 首節點hash為MOVED -1時，表示該節點是轉發節點，說明正在擴容，幫助擴容
4.3 首節點加鎖
- 4.3.1 遍歷鏈表尋找并添加/覆蓋
- 4.3.2 遍歷樹尋找并添加/覆蓋
addCount統計每個節點上的數據，并檢查擴容
- put代碼
//onlyIfAbsent為true時,如果原來有k,v則這次不會覆蓋
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
//1.獲取哈希值：擾動算法+確保哈希值為正數
int hash = spread(key.hashCode());
int binCount = 0;
//樂觀鎖思想 CSA+失敗重試
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
//2.哈希表為空 CAS保證只有一個線程初始化
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//3. 哈希算法求得索引找到索引上的首節點
//4.1 節點為空時,直接CAS構建節點
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
}
//4.2 索引首節點hash 為MOVED 說明該節點是轉發節點，當前正在擴容,去幫助擴容
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
else {
V oldVal = null;
//4.3 首節點 加鎖
synchronized (f) {
//首節點沒變
if (tabAt(tab, i) == f) {
//首節點哈希值大于等于0 說明節點是鏈表上的節點
//4.3.1 遍歷鏈表尋找然后添加/覆蓋
if (fh >= 0) {
//記錄鏈表上有幾個節點
binCount = 1;
//遍歷鏈表找到則替換,如果遍歷完了還沒找到就添加（尾插）
for (Node<K,V> e = f;; ++binCount) {
K ek;
//替換
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
oldVal = e.val;
//onlyIfAbsent為false允許覆蓋（使用xxIfAbsent方法時，有值就不覆蓋）
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;
}
}
}
//如果是紅黑樹首節點,則找到對應節點再覆蓋
//4.3.2 遍歷樹尋找然后添加/覆蓋
else if (f instanceof TreeBin) {
Node<K,V> p;
binCount = 2;
//如果是添加返回null,返回不是null則出來添加
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
//覆蓋
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD)
//鏈表上的節點超過TREEIFY_THRESHOLD 8個（不算首節點） 并且 數組長度超過64才樹化，否則擴容
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
//5.添加計數,用于統計元素（添加節點的情況）
addCount(1L, binCount);
return null;
}擴容
為了避免頻繁發生哈希沖突，當哈希表中的元素數量 / 哈希表長度 超過負載因子時，進行擴容（增大哈希表的長度）。
一般來說擴容都是增大哈希表長度的2倍，比如從32到64保證長度是2次冪；如果擴容長度達到整型上限則使用整型最大值。
當發生擴容時，需要將數組中每個槽里的鏈表或樹遷移到新數組中。
如果處理器是多核，那么這個遷移的操作并不是一個線程單獨完成的，而是會讓其他線程也來幫助遷移。
在遷移時讓每個線程從右往左的每次遷移多個槽，遷移完再判斷是否全部遷移完，如果沒遷移完則繼續循環遷移。
擴容操作主要在transfer方法中，擴容主要在三個場景下：
- addCount：添加完節點增加計數檢查擴容
- helpTransfer：線程put時發現正在遷移，來幫忙擴容
- tryPresize：嘗試調整容量（批量添加putAll，樹化數組長度沒超過64時treeifyBin調用）
分為以下3個步驟
- 根據CPU核數、哈希表總長度計算每次遷移多少個槽，最小16個
- 新哈希表為空，說明是初始化
- 循環遷移
3.1 分配負責遷移的區間 [bround,i]（可能存在多線程同時遷移）
3.2 遷移：分為鏈表遷移、樹遷移
鏈表遷移
- 將鏈表上的節點充分散列到新哈希表的index、index+舊哈希表長度的兩個下標中（與HashMap類似）
- 將index位置鏈表中的節點 （hash & 哈希表長度），結果為0的放到新數組的index位置，結果為1放到新數組index+舊哈希表長度的位置
比如舊哈希表長度為16，在索引3的位置上，16的二進制是10000，hash&16 => hash& 10000 ，也就是說節點哈希值第5位是0就放到新哈希表的3位置上，是1就放到新哈希表的3+16下標。
3.使用頭插法將計算結果為0構建成ln鏈表，為1構建成hn鏈表，為方便構建鏈表，會先尋找lastRun節點：lastRun節點及后續節點都為同一鏈表上的節點，方便遷移。
構建鏈表前先構建lastRun，比如圖中lastRun e->f ，先將lastRun放到ln鏈表上，在遍歷原始鏈表，遍歷到a ：a->e->f，遍歷到b：b->a->e->f。
每遷移完一個索引位置就將轉發節點設置到原哈希表對應位置，當其他線程進行讀get操作時，根據轉發節點來新哈希表中尋找，進行寫put操作時，來幫助擴容（其他區間進行遷移）
擴容代碼
//tab 舊哈希表
//nextTab 新哈希表
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
//1.計算每次遷移多少個槽
//n:哈希表長度（多少個槽）
int n = tab.length, stride;
//stride:每次負責遷移多少個槽
//NCPU: CPU核數
//如果是多核，每次遷移槽數 = 總槽數無符號右移3位（n/8）再除CPU核數
//每次最小遷移槽數 = MIN_TRANSFER_STRIDE = 16
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
//2.如果新哈希表為空，說明是初始化
if (nextTab == null) { // initiating
try {
@SuppressWarnings("unchecked")
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:是否繼續循環遷移
boolean advance = true;
//
boolean finishing = false; // to ensure sweep before committing nextTab
//3.循環遷移
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
//3.1 分配負責遷移的區間
//bound為左區間 i為右區間
while (advance) {
int nextIndex, nextBound;
//處理完一個槽 右區間 自減
if (--i >= bound || finishing)
advance = false;
//transferIndex<=0說明 要遷移的區間全分配完
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
//CAS設置本次遷移的區間，防止多線程分到相同區間
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
bound = nextBound;
i = nextIndex - 1;
advance = false;
}
}
//3.2 遷移
//3.2.1 如果右區間i不再范圍，說明遷移完
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
//如果完成遷移，設置哈希表、數量
if (finishing) {
nextTable = null;
table = nextTab;
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為true 后面再循環 去設置哈希表、數量等操作
finishing = advance = true;
i = n; // recheck before commit
}
}
//3.2.2 如果舊哈希表i位置節點為空就CAS設置成轉發節點
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
//3.2.3 如果舊哈希表該位置首節點是轉發節點，說明其他線程已處理，重新循環
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
else {
//3.2.4 對首節點加鎖 遷移
synchronized (f) {
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;
//3.2.4.1 鏈表遷移
//首節點哈希值大于等于0 說明 是鏈表節點
if (fh >= 0) {
int runBit = fh & n;
Node<K,V> lastRun = f;
//尋找lastRun節點
for (Node<K,V> p = f.next; p != null; p = p.next) {
int b = p.hash & n;
if (b != runBit) {
runBit = b;
lastRun = p;
}
}
//如果最后一次計算值是0
//lastRun節點以及后續節點計算值都是0構建成ln鏈表 否則 都是1構建成hn鏈表
if (runBit == 0) {
ln = lastRun;
hn = null;
}
else {
hn = lastRun;
ln = null;
}
//遍歷構建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;
//頭插：Node構造第四個參數是后繼節點
if ((ph & n) == 0)
ln = new Node<K,V>(ph, pk, pv, ln);
else
hn = new Node<K,V>(ph, pk, pv, hn);
}
//設置ln鏈表到i位置
setTabAt(nextTab, i, ln);
//設置hn鏈表到i+n位置
setTabAt(nextTab, i + n, hn);
//設置轉發節點
setTabAt(tab, i, fwd);
advance = true;
}
//3.2.4.2 樹遷移
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;
}
}
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;
}
}
}
}
}
}實現原理并沒有對紅黑樹進行太多描述，一方面是紅黑樹的概念太多，另一方面是我忘的差不多了（已經老了，不像大學那樣可以手寫紅黑樹了）
還有一方面是：我認為只需要知道使用紅黑樹的好處就足夠，而且工作中也不常用，就算死扣紅黑樹要怎么變色、左旋、右旋去滿足紅黑樹的條件也沒什么意義，感興趣的同學去學習就好了
迭代器
ConcurrentHashMap中的迭代器是弱一致性，在獲取時使用記錄的哈希表重新構建新對象
Entry迭代器：
public Iterator<Map.Entry<K,V>> iterator() {
ConcurrentHashMap<K,V> m = map;
Node<K,V>[] t;
int f = (t = m.table) == null ? 0 : t.length;
return new EntryIterator<K,V>(t, f, 0, f, m);
}key迭代器
public Enumeration<K> keys() {
Node<K,V>[] t;
int f = (t = table) == null ? 0 : t.length;
return new KeyIterator<K,V>(t, f, 0, f, this);
}public Enumeration<V> elements() {
Node<K,V>[] t;
int f = (t = table) == null ? 0 : t.length;
return new ValueIterator<K,V>(t, f, 0, f, this);
}總結
- ConcurrentHashMap使用哈希表的數據結構，當發生哈希沖突時，使用鏈地址法解決，將哈希到同一索引的節點構建成鏈表，當數據量達到一定閾值，會將鏈表轉化為紅黑樹。
- ConcurrentHashMap使用volatile修飾存儲數據，使得在讀場景下對其他線程的修改可見，不需要使用同步機制，使用CAS與synchronzied保證寫場景下的原子性。
- 在get查詢數據時，先將key的哈希值通過擾動算法（高低16位異或）并保證結果為正數（與上符號位0），再與上哈希表長度-1求出索引值，找到索引后再根據不同情況查找（比較先判斷哈希值，相等再判斷key）
- 在put添加/覆蓋數據時，也是先通過擾動算法和哈希求出索引位置，在根據不同情況查找，找到則覆蓋，找不到則替換。
- 在需要擴容時，會為線程安排需要遷移的槽區間，當其他線程進行put時也會來幫忙遷移，每次線程遷移完一個槽，會設置轉發節點到原哈希表中，這樣有線程查詢就可以通過轉發節點來新哈希表中查找，當遷移完所有槽時留一個線程來設置哈希表、數量等。
- 迭代器使用的是弱一致性，在獲取迭代器時通過哈希表去構建新的對象。
- ConcurrentHashMap 只保證相對線程安全，不能保證絕對線程安全，如果需要進行一系列操作時，要正確的去使用。
