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