資訊專欄INFORMATION COLUMN
摘要：上一篇文章已經(jīng)就進行了部分說明，介紹了其中涉及的常量和變量的含義，有些部分需要結(jié)合方法源碼來理解，今天這篇文章就繼續(xù)講解并發(fā)前言本文主要介紹中的一些重要方法，結(jié)合上篇文章中的講解部分進行更進一步的介紹回顧下上篇文章，我們應(yīng)該已經(jīng)知道的整體結(jié)
上一篇文章已經(jīng)就ConcurrentHashMap進行了部分說明，介紹了其中涉及的常量和變量的含義，有些部分需要結(jié)合方法源碼來理解，今天這篇文章就繼續(xù)講解并發(fā)ConcurrentHashMap
前言本文主要介紹ConcurrentHashMap中的一些重要方法，結(jié)合上篇文章中的講解部分進行更進一步的介紹
回顧下上篇文章，我們應(yīng)該已經(jīng)知道ConcurrentHashMap的整體結(jié)構(gòu)和HashMap基本一致，不同的是處理多線程并發(fā)下保證操作的正確性，ConcurrentHashMap通過CAS和synchronized進行并發(fā)控制，當然，這種情況下各種處理都會變的更為復(fù)雜，下面我們就通過方法來深入理解ConcurrentHashMap的操作
重要方法在一些方法中展示了各個變量以及常量的使用，能讓我們更好的理解其中的操作
tabAt/casTabAt/setTabAt下列方法用于讀寫table數(shù)組，使用Unsafe提供的更新獲取volatile變量，CAS更新數(shù)組元素等操作
// 讀取table[i]
@SuppressWarnings("unchecked")
static final Node tabAt(Node[] tab, int i) {
return (Node)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
}
// CAS更新table[i]
static final boolean casTabAt(Node[] tab, int i,
Node c, Node v) {
return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
}
// 插入table[i]
static final void setTabAt(Node[] tab, int i, Node v) {
U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
}
size
size方法返回了一個不精確的值，在多線程環(huán)境下，返回一個不精確的值，通過sumCount迭代counterCells統(tǒng)計sum值。
public int size() {
long n = sumCount();
return ((n < 0L) ? 0 :
(n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
(int)n);
}
final long sumCount() {
CounterCell[] as = counterCells; CounterCell a;
long sum = baseCount;
if (as != null) {
for (int i = 0; i < as.length; ++i) {
if ((a = as[i]) != null)
sum += a.value;
}
}
return sum;
}
這里很多人可能會問，為什么需要疊加counterCells數(shù)組的值呢？
其實這和ConcurrentHashMap特點有關(guān)，多線程環(huán)境下，同時插入值，執(zhí)行CAS操作，執(zhí)行成功的更新了baseCount，而執(zhí)行失敗的則將值放入到了counterCells數(shù)組中，可以查閱CounterCell內(nèi)部類源碼，只有一個long類型變量，每次進行插入或者刪除時調(diào)用addCount通過CAS操作更新baseCount，失敗時執(zhí)行fullAddCount方法，初始化counterCells數(shù)組，并將1（相當于插入或刪除一個元素）插入到CounterCell類中，這樣盡可能保證了Map長度的正確性，這里理解流程即可，不深入，addCount部分有具體操作可查看
@sun.misc.Contended static final class CounterCell {
volatile long value;
CounterCell(long x) { value = x; }
}
get
參考HashMap，類似操作流程，需要注意的也就是在eh < 0處，如果是特殊節(jié)點，比如TreeBin或者ForwardingNode節(jié)點，則調(diào)用其具體類實現(xiàn)的find方法完成遍歷查詢，內(nèi)部類解釋可以參考我的上一篇文章
計算hash值
判斷table是否為空，不為空，找到對應(yīng)hash桶根節(jié)點判斷
非根節(jié)點繼續(xù)遍歷樹或者鏈表，存在對應(yīng)值則返回對應(yīng)值，否則返回null
public V get(Object key) {
Node[] tab; Node e, p; int n, eh; K ek;
// 計算key的hash值
int h = spread(key.hashCode());
// table非空并且對應(yīng)的hash桶根節(jié)點不為空
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
// 查找節(jié)點為根節(jié)點
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
// 樹節(jié)點或者擴容中（FN節(jié)點）
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;
}
// 參考HashMap的hash方法，不同之處在于和HASH_BITS進行了一次與操作，最高位變?yōu)榱?，即為正數(shù)，因為前一篇文章也已經(jīng)說過負數(shù)hash值有特殊意義
static final int spread(int h) {
return (h ^ (h >>> 16)) & HASH_BITS;
}
containsValue
這里通過靜態(tài)內(nèi)部類實現(xiàn)Traverser來遍歷數(shù)組，具體的內(nèi)部實現(xiàn)查看上篇文章里中的內(nèi)部類說明，advance相當于查找到下一個非空節(jié)點
public boolean containsValue(Object value) {
if (value == null)
throw new NullPointerException();
Node[] t;
if ((t = table) != null) {
Traverser it = new Traverser(t, t.length, 0, t.length);
for (Node p; (p = it.advance()) != null; ) {
V v;
if ((v = p.val) == value || (v != null && value.equals(v)))
return true;
}
}
return false;
}
遍歷時遇見特殊節(jié)點的處理上一篇文章中已經(jīng)畫圖說明，如下：
putValputVal整體同HashMap的putVal操作，操作流程上基本類似，只是在多線程操作下需要正確的處理插入值操作，同時如果發(fā)現(xiàn)有線程在進行擴容操作時，需幫助擴容，然后再進行插入值的流程操作
final V putVal(K key, V value, boolean onlyIfAbsent) {
// 空值判斷
if (key == null || value == null) throw new NullPointerException();
// hash值計算，保證了hash值為正數(shù)
int hash = spread(key.hashCode());
// 當前bin中元素的個數(shù)，判斷是否樹化處理
int binCount = 0;
// 無限循環(huán)直到被正確處理
for (Node[] tab = table;;) {
Node f; int n, i, fh;
// 空表進行初始化操作
if (tab == null || (n = tab.length) == 0)
tab = initTable();
// 計算出的hash桶位置鏈表頭節(jié)點無值則通過CAS插入值
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
if (casTabAt(tab, i, null,
new Node(hash, key, value, null)))
break; // no lock when adding to empty bin
}
// 如果hash桶鏈表頭節(jié)點為MOVED狀態(tài)，即說明有線程在進行擴容操作，則通過helpTransfer幫助擴容操作
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
else {
V oldVal = null;
// hash桶鏈表頭節(jié)點加鎖，在多線程環(huán)境下其他線程不能同時操作當前相同的頭節(jié)點代碼塊
synchronized (f) {
if (tabAt(tab, i) == f) {
// 正常鏈表插入操作
if (fh >= 0) {
binCount = 1;
for (Node e = f;; ++binCount) {
K ek;
// key和hash值相同則進行替換
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
oldVal = e.val;
if (!onlyIfAbsent)
e.val = value;
break;
}
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;
}
}
}
}
if (binCount != 0) {
// 達到樹化閾值，則可能進行樹化操作
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
// size+1操作
addCount(1L, binCount);
return null;
}
resizeStamp
前一篇文章在對sizeCtl說明時在多個線程幫忙擴容時其值小于0時做過一些說明，在源碼中涉及到了下面這個方法，先理解清楚這個方法比較重要
這里的參數(shù)每次傳入的都是當前數(shù)組的長度，也就是說每次這里生成的數(shù)都與當時擴容時的數(shù)組長度有關(guān)，Integer.numberOfLeadingZeros(n)，返回二進制表示，前面有多少個連續(xù)的0，RESIZE_STAMP_BITS固定為16，沒發(fā)現(xiàn)有提供方法來修改這個變量，位或運算得到一個值，這個值表示了與擴容時數(shù)組的長度相關(guān)，這里需記得是左移了（RESIZE_STAMP_BITS - 1），因為后邊代碼中我們需要反向操作右移來重新獲取
這里通過這個方法與數(shù)組長度關(guān)聯(lián)，同時sizeCtl也會與之關(guān)聯(lián)，同時也記錄當前擴容中的線程數(shù)，故sizeCtl在擴容中同時兼顧了兩種作用，一是判斷是否是在同一個批次的擴容中（都是從16擴容到32），同時判斷當前擴容中參與的線程數(shù)來確定是否結(jié)束和初始化操作
/**
* Returns the stamp bits for resizing a table of size n.
* Must be negative when shifted left by RESIZE_STAMP_SHIFT.
*/
static final int resizeStamp(int n) {
return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1));
}
addCount
整體來看主要進行了兩部分內(nèi)容，一是更新baseCount，二是檢查是否進行擴容操作。其實這個方法里還是相當復(fù)雜的，涉及到了線程私有的偽隨機數(shù)生成器ThreadLocalRandom，并發(fā)效率更高的LongAdder，不過初學者可以不用研究那么深入，這里不詳細說明，大概了解就好
private final void addCount(long x, int check) {
CounterCell[] as; long b, s;
// 通過CAS更新baseCount
if ((as = counterCells) != null ||
!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
// 更新baseCount失敗
CounterCell a; long v; int m;
boolean uncontended = true;
// 相當于每個線程的probe就是它在CounterCell數(shù)組中的hash code，用來定位counterCells數(shù)組
if (as == null || (m = as.length - 1) < 0 ||
(a = as[ThreadLocalRandom.getProbe() & m]) == null ||
!(uncontended =
U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
// 更新cellvalue失敗則執(zhí)行fullAddCount，具體不看了，比較復(fù)雜，不停嘗試更新計數(shù)
// 源碼注釋上也寫了類似LongAdder
fullAddCount(x, uncontended);
return;
}
// 執(zhí)行到此說明更新計數(shù)器成功,判斷是否退出，為什么是1其實還是有點困惑
if (check <= 1)
return;
s = sumCount();
}
// check大于0代表著對應(yīng)hash桶下的節(jié)點數(shù)，檢查是否擴容
// 滿足條件幫助擴容，不滿足退出
if (check >= 0) {
Node[] tab, nt; int n, sc;
// 注意，這里條件中3個變量賦值同時while判斷
// sc = sizeCtl，tab = table，n = tab.length
// 在并發(fā)操作中可能會出現(xiàn)變量錯誤的情況造成擴容處理出錯，通過resizeStamp保證擴容時版本一致
while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
(n = tab.length) < MAXIMUM_CAPACITY) {
// resizeStamp根據(jù)n返回一個擴容版本戳，保證唯一性，上邊一個方法我已經(jīng)說明了
int rs = resizeStamp(n);
// 說明有別的線程在擴容
if (sc < 0) {
// 判斷是否幫助擴容，滿足條件，不幫助擴容，這里會分析下，看下面的分析部分
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
transferIndex <= 0)
break;
// 幫助擴容，線程數(shù)+1
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);
s = sumCount();
}
}
}
在上面這段代碼中，不幫助擴容的條件中有些地方讓人困惑
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
transferIndex <= 0)
break;
(sc >>> RESIZE_STAMP_SHIFT) != rs
首先需要明白上邊整個擴容中的第一個線程會通過U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2)設(shè)置sizeCtl，之后擴容線程增加則通過U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)更新
sc右移RESIZE_STAMP_SHIFT（由于RESIZE_STAMP_BITS不提供修改方法，RESIZE_STAMP_SHIFT也只能取到16），第一個條件為什么是這個？需要結(jié)合擴容代碼來看，首個線程搶到擴容任務(wù)時需先創(chuàng)建nextTable，設(shè)置transferIndex，在執(zhí)行之前需要將sizeCtl更新，即U.compareAndSwapInt(this, SIZECTL, sc,(rs << RESIZE_STAMP_SHIFT) + 2)，代碼存在于addCount和tryPresize方法中，sizeCtl在每次擴容時會更新成(rs << RESIZE_STAMP_SHIFT) + 2。判斷條件里sc無符號右移，如果是相同的一次擴容過程，則與rs相等是肯定的，rs是由resizeStamp根據(jù)長度n計算得來，其實最終這里比較的也就是table的長度，防止多次擴容下錯誤的幫助了擴容
另外在已經(jīng)有線程擴容的情況下增加擴容線程會會更新sizeCtl，U.compareAndSwapInt(this, SIZECTL, sc, sc + 1) 看出在首次更新的基礎(chǔ)上加1即可，擴容線程完成自己的任務(wù)同理減1，結(jié)合上邊對resizeStamp的說明應(yīng)該算很清楚了
以上部分也證實了上篇文章中sizeCtl注釋是不正確的
sc == rs + 1 || sc == rs + MAX_RESIZERS
這個條件是有問題的，sc小于0，rs大于0，兩個條件一直為false，沒有true的可能，從這個條件上看，應(yīng)該是判斷擴容完畢和擴容線程數(shù)達到最大時不能幫助擴容。
我們想一下，第一次線程擴容時已經(jīng)將sc更新成(rs << RESIZE_STAMP_SHIFT) + 2，這里判斷的話需要改為sc == ( rs << RESIZE_STAMP_SHIFT ) +1 才對，不能將sc右移，右移將會導(dǎo)致低16位記錄的線程數(shù)數(shù)據(jù)丟失，最大線程數(shù)判斷同理，應(yīng)改為sc == ( rs << RESIZE_STAMP_SHIFT ) + MAX_RESIZERS
我在Oracle官網(wǎng)bug庫里看到已經(jīng)提到了這個問題：https://bugs.java.com/bugdata...
(nt = nextTable) == null
此時狀態(tài)可能表明擴容已經(jīng)結(jié)束或者第一個線程在擴容中，不能幫助擴容
transferIndex <= 0
transfer任務(wù)已經(jīng)被分配完畢，不能分配任務(wù)給當前線程，不能幫助擴容，幫助擴容部分下面會說到
helpTransfer如果正在進行擴容操作，則幫助擴容
/**
* Helps transfer if a resize is in progress.
*/
final Node[] helpTransfer(Node[] tab, Node f) {
Node[] nextTab; int sc;
// 判斷是否為ForwardingNode并且nextTable是否已經(jīng)被創(chuàng)建
if (tab != null && (f instanceof ForwardingNode) &&
(nextTab = ((ForwardingNode)f).nextTable) != null) {
// 根據(jù)長度獲取擴容戳
int rs = resizeStamp(tab.length);
// 再次驗證是否正在擴容
while (nextTab == nextTable && table == tab &&
(sc = sizeCtl) < 0) {
// 幾個條件上邊已經(jīng)解釋過了，滿足不幫助擴容
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || transferIndex <= 0)
break;
// sizeCtl加1，表示當前線程加入擴容，多了一個線程幫忙
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
transfer(tab, nextTab);
break;
}
}
return nextTab;
}
return table;
}
tryPresize
預(yù)先擴容，putAll和treeifyBin中使用到，不滿足table容量時，進行一次擴容操作
/**
* Tries to presize table to accommodate the given number of elements.
*
* @param size number of elements (doesn"t need to be perfectly accurate)
*/
private final void tryPresize(int size) {
// 判斷長度是否超過最大值，超過則賦值為最大值，正常則通過tableSizeFor計算擴容后的長度
int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :
tableSizeFor(size + (size >>> 1) + 1);
int sc;
// 未初始化或擴容完成才能執(zhí)行本次擴容操作
while ((sc = sizeCtl) >= 0) {
Node[] tab = table; int n;
// table未初始化
if (tab == null || (n = tab.length) == 0) {
n = (sc > c) ? sc : c;
// 置為-1表示數(shù)組初始化，前一篇文章已經(jīng)說明
if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
if (table == tab) {
@SuppressWarnings("unchecked")
Node[] nt = (Node[])new Node[n];
table = nt;
sc = n - (n >>> 2);
}
} finally {
// 此時相當于閾值
sizeCtl = sc;
}
}
}
// 已經(jīng)初始化，擴容長度小于閾值或者大于最大值，不進行擴容操作
else if (c <= sc || n >= MAXIMUM_CAPACITY)
break;
// 再次驗證table未改變
else if (tab == table) {
int rs = resizeStamp(n);
// 同上邊代碼部分，判斷是否幫助擴容
if (sc < 0) {
Node[] 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
table遷移操作,通過transferIndex來完成任務(wù)的分配，之前文章變量中也提及了MIN_TRANSFER_STRIDE（最小步長），對每個擴容線程申請遷移的hash桶數(shù)量做了限制，每次需要擴容線程執(zhí)行完畢已經(jīng)領(lǐng)取完的hash桶遷移任務(wù)才可以繼續(xù)領(lǐng)取任務(wù)幫助遷移，最后一個遷移線程在遷移完畢后會進行檢查
