ConcurrentHashMap源碼分析_JDK1.8版本
聲明
文章均為本人技術(shù)筆記，轉(zhuǎn)載請注明出處
[1] https://segmentfault.com/u/yzwall
[2] blog.csdn.net/j_dark/
JDK1.6版本
ConcurrentHashMap結(jié)構(gòu)
在JDK1.6中，ConcurrentHashMap將數(shù)據(jù)分成一段一段存儲，給每一段數(shù)據(jù)配一把鎖，當(dāng)一個線程獲得鎖互斥訪問一個段數(shù)據(jù)時，其他段的數(shù)據(jù)也可被其他線程訪問；每個Segment擁有一把可重入鎖，因此ConcurrentHashMap的分段鎖數(shù)目即為Segment數(shù)組長度。ConcurrentHashMap結(jié)構(gòu)：每一個segment都是一個HashEntry[] table， table中的每一個元素本質(zhì)上都是一個HashEntry的單向隊列（單向鏈表實現(xiàn)）。每一個segment都是一個HashEntry[] table， table中的每一個元素本質(zhì)上都是一個HashEntry的單向隊列。
鎖分離實現(xiàn)
當(dāng)一個線程訪問Node/鍵值對數(shù)據(jù)時，必須獲得與它對應(yīng)的segment鎖，其他線程可以訪問其他Segment中的數(shù)據(jù)（鎖分離）；
ConcurrentHashMap聲明
public class ConcurrentHashMap extends AbstractMap implements ConcurrentMap, Serializable
無鎖算法：CAS
樂觀鎖與悲觀鎖
悲觀鎖比如synchronized鎖，為確保其他線程不會干擾當(dāng)前線程工作，因此掛起其他需要鎖的線程，等待持有鎖的線程釋放；
樂觀鎖總是假設(shè)沒有沖突發(fā)生去做操作，如果檢測到?jīng)_突就失敗重試，知道成功為止；
CAS算法
CAS(Compare And Swap)：CAS算法包含三個參數(shù)CAS(V, E, N)，判斷預(yù)期值E和內(nèi)存舊值是否相同(Compare)，如果相等用新值N覆蓋舊值V(Swap)，否則失??；
當(dāng)多個線程嘗試使用CAS同時更新同一個變量時，只有其中一個線程能更新變量的值，其他線程失?。ㄊ【€程不會被阻塞，而是被告知“失敗”，可以繼續(xù)嘗試）；
CAS在硬件層面可以被編譯為機(jī)器指令執(zhí)行，因此性能高于基于鎖占有方式實現(xiàn)線程安全；
ConcurrentHashMap結(jié)構(gòu)圖示
與JDK1.6對比
JDK 1.8取消類segments字段，直接用table數(shù)組存儲鍵值對，JDK1.6中每個bucket中鍵值對組織方式是單向鏈表，查找復(fù)雜度是O(n)，JDK1.8中當(dāng)鏈表長度超過TREEIFY_THRESHOLD時，鏈表轉(zhuǎn)換為紅黑樹，查詢復(fù)雜度可以降低到O(log n)，改進(jìn)性能；
鎖分離
JDK1.8中，一個線程每次對一個桶（鏈表 or 紅黑樹）進(jìn)行加鎖，其他線程仍然可以訪問其他桶；
線程安全
ConcurrentHashMap底層數(shù)據(jù)結(jié)構(gòu)與HashMap相同，仍然采用table數(shù)組+鏈表+紅黑樹結(jié)構(gòu)；
一個線程進(jìn)行put/remove操作時，對桶（鏈表 or 紅黑樹）加上synchronized獨占鎖；
ConcurrentHashMap采用CAS算法保證線程安全；
ConcurrentHashMap基本數(shù)據(jù)結(jié)構(gòu)
transient volatile Node[] table：鍵值對桶數(shù)組
private transient volatile Node[] nextTable： rehash擴(kuò)容時用到的新鍵值對數(shù)組
private transient volatile long baseCount：記錄當(dāng)前鍵值對總數(shù)，通過CAS更新，對所有線程可見
private transient volatile int sizeCtl
sizeCtl表示鍵值對總數(shù)閾值，通過CAS更新, 對所有線程可見
當(dāng)sizeCtl < 0時，表示多個線程在等待擴(kuò)容；
當(dāng)sizeCtl = 0時，默認(rèn)值；
當(dāng)sizeCtl > 0時，表示擴(kuò)容的閾值；
private transient volatile int cellBusy：自旋鎖；
private transient volatile CounterCell[] counterCells: counter cell表，長度總為2的冪次；
static class Segment：在JDK1.8中，Segment類僅僅在序列化和反序列化時發(fā)揮作用；
// 視圖
private transient KeySetView keySet
private transient ValuesView values
private transient EntrySetView entrySet
描述鍵值對：Node
static class Node implements Map.Entry {
final int hash;
final K key;
// 鍵值對的value和next均為volatile類型
volatile V val;
volatile Node next;
...
}
ConcurrentHashMap重要方法分析
構(gòu)造函數(shù)
ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel)
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {
if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
if (initialCapacity < concurrencyLevel) // Use at least as many bins
initialCapacity = concurrencyLevel; // as estimated threads
long size = (long)(1.0 + (long)initialCapacity / loadFactor);
int cap = (size >= (long)MAXIMUM_CAPACITY) ?
MAXIMUM_CAPACITY : tableSizeFor((int)size);
this.sizeCtl = cap;
}
該構(gòu)造器會根據(jù)輸入的initialCapacity確定一個 >= initialCapacity的最小2的次冪；
concurrentLevel：在JDK1.8之前本質(zhì)是ConcurrentHashMap分段鎖總數(shù)，表示同時更新ConcurrentHashMap且不產(chǎn)生鎖競爭的最大線程數(shù)；在JDK1.8中，僅在構(gòu)造器中確保初始容量>=concurrentLevel，為兼容舊版本而保留；
添加/更新鍵值對：putVal
putVal方法分析
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode());
int binCount = 0;
// 不斷CAS探測，如果其他線程正在修改tab，CAS嘗試失敗，直到成功為止
for (Node[] tab = table;;) {
Node f; int n, i, fh;
// 空表，對tab進(jìn)行初始化
if (tab == null || (n = tab.length) == 0)
tab = initTable();
/**
* CAS探測空桶
* 計算key所在bucket表中數(shù)組索引: i = (n - 1) & hash)
*/
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
// CAS添加新鍵值對
if (casTabAt(tab, i, null,
new Node(hash, key, value, null)))
break; // no lock when adding to empty bin
}
// 檢測到tab[i]桶正在進(jìn)行rehash,
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
else {
V oldVal = null;
// 對桶的首元素上鎖獨占
synchronized (f) {
if (tabAt(tab, i) == f) {
// 桶中鍵值對組織形式是鏈表
if (fh >= 0) {
binCount = 1;
for (Node e = f;; ++binCount) {
K ek;
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
oldVal = e.val;
// 查找到對應(yīng)鍵值對，更新值
if (!onlyIfAbsent)
e.val = value;
break;
}
// 桶中沒有對應(yīng)鍵值對，插入到鏈表尾部
Node pred = e;
if ((e = e.next) == null) {
pred.next = new Node(hash, key,
value, null);
break;
}
}
}
// 桶中鍵值對組織形式是紅黑樹
else if (f instanceof TreeBin) {
Node p;
binCount = 2;
if ((p = ((TreeBin)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
// 檢查桶中鍵值對總數(shù)
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD)
// 鏈表轉(zhuǎn)換為紅黑樹
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
// 更新baseCount
addCount(1L, binCount);
return null;
}
synchronized (f) {}操作通過對桶的首元素 = 鏈表表頭 Or 紅黑樹根節(jié)點加鎖，從而實現(xiàn)對整個桶進(jìn)行加鎖，有鎖分離思想的體現(xiàn)；
獲取鍵值對：get
public V get(Object key) {
Node[] tab; Node e, p; int n, eh; K ek;
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;
}
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;
}
get方法通過CAS保證鍵值對的原子性，當(dāng)tab[i]被鎖住，CAS失敗并不斷重試，保證get不會出錯；
刪除鍵值對：remove
擴(kuò)容機(jī)制
transfer
當(dāng)baseCount超過sizeCtl，將table中所有bin內(nèi)的鍵值對拷貝到nextTable；
待補充；
helpTransfer
待補充；
table原子操作方法
獲取tab[i]：tabAt
static final Node tabAt(Node[] tab, int i) {
return (Node)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
}
tabAt方法原子讀取table[i]；調(diào)用Unsafe對象的getObjectVolatile方法獲取tab[i]，由于對volatile寫操作happen-before于volatile讀操作，因此其他線程對table的修改均對get讀取可見；
((long)i << ASHIFT) + ABASE)計算i元素的地址
CAS算法更新鍵值對：casTabAt
static final boolean casTabAt(Node[] tab, int i,
Node c, Node v) {
return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
}
casTabAt通過compareAndSwapObject方法比較tabp[i]和v是否相等，相等就用c更新tab[i];
更新鍵值對：setTabAt
static final void setTabAt(Node[] tab, int i, Node v) {
U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
}
僅在synchronized同步塊中被調(diào)用，更新鍵值對；
CAS更新baseCount
addCountaddCount
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();
}
if (check >= 0) {
Node[] tab, nt; int n, sc;
// 當(dāng)更新后的鍵值對總數(shù)baseCount >= 閾值sizeCtl時，進(jìn)行rehash
while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
(n = tab.length) < MAXIMUM_CAPACITY) {
int rs = resizeStamp(n);
// sc < 0 表示其他線程已經(jīng)在rehash
if (sc < 0) {
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
transferIndex <= 0)
break;
// 其他線程的rehash操作已經(jīng)完成，當(dāng)前線程可以進(jìn)行rehash
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
transfer(tab, nt);
}
// sc >= 0 表示只有當(dāng)前線程在進(jìn)行rehash操作，調(diào)用輔助擴(kuò)容方法transfer
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
transfer(tab, null);
s = sumCount();
}
}
}
addCount負(fù)責(zé)對baseCount + 1操作，CounterCell是Striped64類型，否則應(yīng)對高并發(fā)問題；
fullAddCount
待補充；
參考
[1] 《Java并發(fā)編程的藝術(shù)》
[2] ??http://www.cnblogs.com/leesf4...
[3] ??http://blog.csdn.net/u0108877...
[4] ??http://www.cnblogs.com/Mainz/...
[5] ??http://www.cnblogs.com/huaizu...
[6] ??http://www.cnblogs.com/everSe...
