ConcurrentHashMap的初步使用及場景

CHM的使用

ConcurrentHashMap是J.U.C包里面提供的一個線程安全并且高效的HashMap左刽，所以ConcurrentHashMap在并發(fā)編程的場景中使用的頻率比較高商佑，那么這一節(jié)課我們就從ConcurrentHashMap的使用上以及源碼層面來分析ConcurrentHashMap到底是如何實現(xiàn)安全性的

api使用

ConcurrentHashMap是Map的派生類，所以api基本和Hashmap是類似细卧，主要就是put、get這些方法筒占，接下來基于ConcurrentHashMap的put和get這兩個方法作為切入點來分析ConcurrentHashMap的源碼實現(xiàn)

ConcurrentHashMap的源碼分析

先要做一個說明贪庙，這節(jié)課分析的ConcurrentHashMap是基于Jdk1.8的版本。 JDK1.7和Jdk1.8版本的變化 ConcurrentHashMap和HashMap的實現(xiàn)原理是差不多的翰苫，但是因為ConcurrentHashMap需要支持并發(fā)操作止邮，所以在實現(xiàn)上要比hashmap稍微復(fù)雜一些这橙。 在JDK1.7的實現(xiàn)上，ConrruentHashMap由一個個Segment組成导披，簡單來說屈扎，ConcurrentHashMap是一個Segment數(shù)組，它通過繼承ReentrantLock來進行加鎖撩匕，通過每次鎖住一個segment來保證每個segment內(nèi)的操作的線程安全性從而實現(xiàn)全局線程安全鹰晨。整個結(jié)構(gòu)圖如下

深度分析ConcurrentHashMap原理分析

當每個操作分布在不同的segment上的時候，默認情況下止毕，理論上可以同時支持16個線程的并發(fā)寫入模蜡。 相比于1.7版本，它做了兩個改進

取消了segment分段設(shè)計扁凛，直接使用Node數(shù)組來保存數(shù)據(jù)忍疾，并且采用Node數(shù)組元素作為鎖來實現(xiàn)每一行數(shù)據(jù)進行加鎖來進一步減少并發(fā)沖突的概率

將原本數(shù)組+單向鏈表的數(shù)據(jù)結(jié)構(gòu)變更為了數(shù)組+單向鏈表+紅黑樹的結(jié)構(gòu)。為什么要引入紅黑樹呢谨朝？在正常情況下卤妒，key hash之后如果能夠很均勻的分散在數(shù)組中，那么table數(shù)組中的每個隊列的長度主要為0或者1.但是實際情況下叠必，還是會存在一些隊列長度過長的情況荚孵。如果還采用單向列表方式，那么查詢某個節(jié)點的時間復(fù)雜度就變?yōu)镺(n); 因此對于隊列長度超過8的列表纬朝，JDK1.8采用了紅黑樹的結(jié)構(gòu)收叶，那么查詢的時間復(fù)雜度就會降低到O(logN),可以提升查找的性能；

深度分析ConcurrentHashMap原理分析

這個結(jié)構(gòu)和JDK1.8版本中的Hashmap的實現(xiàn)結(jié)構(gòu)基本一致共苛，但是為了保證線程安全性判没，ConcurrentHashMap的實現(xiàn)會稍微復(fù)雜一下。接下來我們從源碼層面來了解一下它的原理. 我們基于put和get方法來分析它的實現(xiàn)即可

put方法第一階段

1.  public V put(K key, V value) {`

2.  return putVal(key, value, false);`

3.  }

/** Implementation for put and putIfAbsent */

final V putVal(K key, V value, boolean onlyIfAbsent) {

if (key ==  || value == ) throw new PointerException;

int hash = spread(key.hashCode); //計算hash值

int binCount = 0; //用來記錄鏈表的長度

for (Node<K,V>[] tab = table;;) { //這里其實就是自旋操作隅茎，當出現(xiàn)線程競爭時不斷自旋

Node<K,V> f; int n, i, fh;

if (tab ==  || (n = tab.length) == 0)//如果數(shù)組為空澄峰，則進行數(shù)組初始化

tab = initTable; //初始化數(shù)組

//通過hash值對應(yīng)的數(shù)組下標得到第一個節(jié)點; 以volatile讀的方式來讀取table數(shù)組中的元素，保證每次拿到的數(shù)據(jù)都是最新的

else if ((f = tabAt(tab, i = (n - 1) & hash)) == ) { 

//如果該下標返回的節(jié)點為空辟犀，則直接通過cas將新的值封裝成node插入即可俏竞；如果cas失敗，說明存在競爭堂竟，則進入下一次循環(huán)

if (casTabAt(tab, i, ,

new Node<K,V>(hash, key, value, )))

break; // no lock when adding to empty bin

}

//...

｝

假如在上面這段代碼中存在兩個線程魂毁，在不加鎖的情況下：線程A成功執(zhí)行casTabAt操作后，隨后的線程B可以通過tabAt方法立刻看到table[i]的改變出嘹。原因如下：線程A的casTabAt操作席楚，具有volatile讀寫相同的內(nèi)存語義，根據(jù)volatile的happens-before規(guī)則：線程A的casTabAt操作税稼，一定對線程B的tabAt操作可見

initTable

數(shù)組初始化方法烦秩，這個方法比較簡單垮斯，就是初始化一個合適大小的數(shù)組 sizeCtl這個要單獨說一下，如果沒搞懂這個屬性的意義只祠，可能會被搞暈 這個標志是在Node數(shù)組初始化或者擴容的時候的一個控制位標識兜蠕，負數(shù)代表正在進行初始化或者擴容操作 -1 代表正在初始化 -N 代表有N-1有二個線程正在進行擴容操作，這里不是簡單的理解成n個線程铆农，sizeCtl就是-N,這塊后續(xù)在講擴容的時候會說明 0標識Node數(shù)組還沒有被初始化牺氨，正數(shù)代表初始化或者下一次擴容的大小

private final Node<K,V> initTable {

Node<K,V> tab; int sc;

while ((tab = table) ==  || tab.length == 0) {

if ((sc = sizeCtl) < 0)//被其他線程搶占了初始化的操作,則直接讓出自己的CPU時間片

Thread.yield; // lost initialization race; just spin

//通過cas操作狡耻，將sizeCtl替換為-1墩剖，標識當前線程搶占到了初始化資格

else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {

try {

if ((tab = table) ==  || tab.length == 0) {

int n = (sc > 0) ? sc : DEFAULT_CAPACITY;//默認初始容量為16

@SuppressWarnings("unchecked")

//初始化數(shù)組，長度為16夷狰，或者初始化在構(gòu)造ConcurrentHashMap的時候傳入的長度

Node<K,V> nt = (Node<K,V>[])new Node<?,?>[n];

table = tab = nt;//將這個數(shù)組賦值給table

sc = n - (n >>> 2); //計算下次擴容的大小岭皂，實際就是當前容量的0.75倍，這里使用了右移來計算

}

} finally {

sizeCtl = sc; //設(shè)置sizeCtl為sc, 如果默認是16的話沼头，那么這個時候sc=16*0.75=12

}

break;

}

}

return tab;

}

tabAt

該方法獲取對象中offset偏移地址對應(yīng)的對象field的值爷绘。實際上這段代碼的含義等價于tab[i], 但是為什么不直接使用tab[i]來計算呢？ getObjectVolatile进倍，一旦看到volatile關(guān)鍵字土至，就表示可見性。因為對volatile寫操作happen-before于volatile讀操作猾昆，因此其他線程對table的修改均對get讀取可見陶因； 雖然table數(shù)組本身是增加了volatile屬性，但是“volatile的數(shù)組只針對數(shù)組的引用具有volatile的語義垂蜗，而不是它的元素”楷扬。 所以如果有其他線程對這個數(shù)組的元素進行寫操作，那么當前線程來讀的時候不一定能讀到最新的值贴见。 出于性能考慮烘苹，Doug Lea直接通過Unsafe類來對table進行操作。

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);

}

圖解分析

深度分析ConcurrentHashMap原理分析

put方法第二階段

在putVal方法執(zhí)行完成以后片部，會通過addCount來增加ConcurrentHashMap中的元素個數(shù)镣衡，并且還會可能觸發(fā)擴容操作。這里會有兩個非常經(jīng)典的設(shè)計

1. 高并發(fā)下的擴容

2. 如何保證addCount的數(shù)據(jù)安全性以及性能

//將當前ConcurrentHashMap的元素數(shù)量加1档悠，有可能觸發(fā)transfer操作(擴容)

addCount(1L, binCount);

return ;

}

addCount

在putVal最后調(diào)用addCount的時候廊鸥，傳遞了兩個參數(shù)，分別是1和binCount(鏈表長度)站粟，看看addCount方法里面做了什么操作 x表示這次需要在表中增加的元素個數(shù)黍图，check參數(shù)表示是否需要進行擴容檢查，大于等于0都需要進行檢查

private final void addCount(long x, int check) {

CounterCell as; long b, s;

判斷counterCells是否為空奴烙，

1. 如果為空助被，就通過cas操作嘗試修改baseCount變量剖张，對這個變量進行原子累加操作(做這個操作的意義是：如果在沒有競爭的情況下，仍然采用baseCount來記錄元素個數(shù))

2. 如果cas失敗說明存在競爭揩环，這個時候不能再采用baseCount來累加搔弄，而是通過CounterCell來記錄 

if ((as = counterCells) !=  ||

!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {

CounterCell a; long v; int m;

boolean uncontended = true;//是否沖突標識，默認為沒有沖突

這里有幾個判斷

1. 計數(shù)表為空則直接調(diào)用fullAddCount

2. 從計數(shù)表中隨機取出一個數(shù)組的位置為空丰滑，直接調(diào)用fullAddCount

3. 通過CAS修改CounterCell隨機位置的值顾犹，如果修改失敗說明出現(xiàn)并發(fā)情況（這里又用到了一種巧妙的方法），調(diào)用fullAndCount

Random在線程并發(fā)的時候會有性能問題以及可能會產(chǎn)生相同的隨機數(shù),ThreadLocalRandom.getProbe可以解決這個問題褒墨，并且性能要比Random高

if (as ==  || (m = as.length - 1) < 0 ||

(a = as[ThreadLocalRandom.getProbe() & m]) ==  ||

!(uncontended =

U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {

fullAddCount(x, uncontended);//執(zhí)行fullAddCount方法

return;

}

if (check <= 1)//鏈表長度小于等于1炫刷，不需要考慮擴容

return;

s = sumCount;//統(tǒng)計ConcurrentHashMap元素個數(shù)

}

//….

}

CounterCells解釋

ConcurrentHashMap是采用CounterCell數(shù)組來記錄元素個數(shù)的，像一般的集合記錄集合大小郁妈，直接定義一個size的成員變量即可浑玛，當出現(xiàn)改變的時候只要更新這個變量就行。為什么ConcurrentHashMap要用這種形式來處理呢噩咪？ 問題還是處在并發(fā)上顾彰，ConcurrentHashMap是并發(fā)集合，如果用一個成員變量來統(tǒng)計元素個數(shù)的話胃碾，為了保證并發(fā)情況下共享變量的的難全興涨享，勢必會需要通過加鎖或者自旋來實現(xiàn)，如果競爭比較激烈的情況下仆百，size的設(shè)置上會出現(xiàn)比較大的沖突反而影響了性能厕隧，所以在ConcurrentHashMap采用了分片的方法來記錄大小，具體什么意思儒旬，我們來分析下

private transient volatile int cellsBusy;// 標識當前cell數(shù)組是否在初始化或擴容中的CAS標志位

/**

* Table of counter cells. When non-, size is a power of 2.

*/

private transient volatile CounterCell counterCells;// counterCells數(shù)組栏账，總數(shù)值的分值分別存在每個cell中


@sun.misc.Contended static final class CounterCell {

volatile long value;

CounterCell(long x) { value = x; }

}

//看到這段代碼就能夠明白了，CounterCell數(shù)組的每個元素栈源，都存儲一個元素個數(shù)挡爵，而實際我們調(diào)用size方法就是通過這個循環(huán)累加來得到的

//又是一個設(shè)計精華，大家可以借鑒甚垦； 有了這個前提茶鹃，再會過去看addCount這個方法，就容易理解一些了

final long sumCount {

CounterCell as = counterCells; CounterCell a;

long sum = baseCount;

if (as != ) {

for (int i = 0; i < as.length; ++i) {

if ((a = as[i]) != )

sum += a.value;

}

}

return sum;

}

fullAddCount源碼分析

fullAddCount主要是用來初始化CounterCell艰亮，來記錄元素個數(shù)闭翩，里面包含擴容，初始化等操作

private final void fullAddCount(long x, boolean wasUncontended) {

int h;

//獲取當前線程的probe的值迄埃，如果值為0疗韵，則初始化當前線程的probe的值,probe就是隨機數(shù)

if ((h = ThreadLocalRandom.getProbe) == 0) {

ThreadLocalRandom.localInit; // force initialization

h = ThreadLocalRandom.getProbe;

wasUncontended = true; // 由于重新生成了probe，未沖突標志位設(shè)置為true

}

boolean collide = false; // True if last slot nonempty

for (;;) {//自旋

CounterCell as; CounterCell a; int n; long v;

//說明counterCells已經(jīng)被初始化過了侄非，我們先跳過這個代碼蕉汪，先看初始化部分

if ((as = counterCells) !=  && (n = as.length) > 0) { 

if ((a = as[(n - 1) & h]) == ) {// 通過該值與當前線程probe求與流译，獲得cells的下標元素，和hash 表獲取索引是一樣的

if (cellsBusy == 0) { //cellsBusy=0表示counterCells不在初始化或者擴容狀態(tài)下 

CounterCell r = new CounterCell(x); //構(gòu)造一個CounterCell的值者疤，傳入元素個數(shù)

if (cellsBusy == 0 &&//通過cas設(shè)置cellsBusy標識福澡，防止其他線程來對counterCells并發(fā)處理

U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {

boolean created = false;

try { // Recheck under lock

CounterCell rs; int m, j;

//將初始化的r對象的元素個數(shù)放在對應(yīng)下標的位置

if ((rs = counterCells) !=  &&

(m = rs.length) > 0 &&

rs[j = (m - 1) & h] == ) {

rs[j] = r;

created = true;

}

} finally {//恢復(fù)標志位

cellsBusy = 0;

}

if (created)//創(chuàng)建成功，退出循環(huán)

break;

continue;//說明指定cells下標位置的數(shù)據(jù)不為空驹马，則進行下一次循環(huán)

}

}

collide = false;

}

//說明在addCount方法中cas失敗了革砸，并且獲取probe的值不為空

else if (!wasUncontended) // CAS already known to fail

wasUncontended = true; //設(shè)置為未沖突標識，進入下一次自旋

//由于指定下標位置的cell值不為空糯累，則直接通過cas進行原子累加算利，如果成功，則直接退出

else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))//

break;

//如果已經(jīng)有其他線程建立了新的counterCells或者CounterCells大于CPU核心數(shù)（很巧妙寇蚊，線程的并發(fā)數(shù)不會超過cpu核心數(shù)）

else if (counterCells != as || n >= NCPU)

collide = false; //設(shè)置當前線程的循環(huán)失敗不進行擴容

else if (!collide)//恢復(fù)collide狀態(tài)笔时，標識下次循環(huán)會進行擴容

collide = true;

//進入這個步驟，說明CounterCell數(shù)組容量不夠仗岸，線程競爭較大，所以先設(shè)置一個標識表示為正在擴容

else if (cellsBusy == 0 &&

U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {

try {

if (counterCells == as) {// Expand table unless stale

//擴容一倍 2變成4借笙，這個擴容比較簡單

CounterCell rs = new CounterCell[n << 1];

for (int i = 0; i < n; ++i)

rs[i] = as[i];

counterCells = rs;

}

} finally {

cellsBusy = 0;//恢復(fù)標識

}

collide = false;

continue;//繼續(xù)下一次自旋

}

h = ThreadLocalRandom.advanceProbe(h);//更新隨機數(shù)的值

}

初始化CounterCells數(shù)組

//cellsBusy=0表示沒有在做初始化扒怖，通過cas更新cellsbusy的值標注當前線程正在做初始化操作

else if (cellsBusy == 0 && counterCells == as &&

U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {

boolean init = false;

try { // Initialize table

if (counterCells == as) {

CounterCell rs = new CounterCell[2]; //初始化容量為2

rs[h & 1] = new CounterCell(x);//將x也就是元素的個數(shù)放在指定的數(shù)組下標位置

counterCells = rs;//賦值給counterCells

init = true;//設(shè)置初始化完成標識

}

} finally {

cellsBusy = 0;//恢復(fù)標識

}

if (init)

break;

}

//競爭激烈，其它線程占據(jù)cell 數(shù)組业稼，直接累加在base變量中

else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x))

break; // Fall back on using base

}

}

CounterCells初始化圖解

初始化長度為2的數(shù)組盗痒，然后隨機得到指定的一個數(shù)組下標，將需要新增的值加入到對應(yīng)下標位置處

深度分析ConcurrentHashMap原理分析

transfer擴容階段

判斷是否需要擴容低散，也就是當更新后的鍵值對總數(shù)baseCount >= 閾值sizeCtl時俯邓，進行rehash，這里面會有兩個邏輯熔号。

1. 如果當前正在處于擴容階段稽鞭，則當前線程會加入并且協(xié)助擴容

2. 如果當前沒有在擴容，則直接觸發(fā)擴容操作

if (check >= 0) {//如果binCount>=0引镊，標識需要檢查擴容

Node<K,V> tab, nt; int n, sc;

//s標識集合大小朦蕴，如果集合大小大于或等于擴容閾值（默認值的0.75）

//并且table不為空并且table的長度小于最大容量

while (s >= (long)(sc = sizeCtl) && (tab = table) !=  &&

(n = tab.length) < MAXIMUM_CAPACITY) {

int rs = resizeStamp(n);//這里是生成一個唯一的擴容戳，這個是干嘛用的呢弟头？且聽我慢慢分析

if (sc < 0) {//sc<0吩抓，也就是sizeCtl<0，說明已經(jīng)有別的線程正在擴容了

//這5個條件只要有一個條件為true赴恨，說明當前線程不能幫助進行此次的擴容疹娶，直接跳出循環(huán)

//sc >>> RESIZE_STAMP_SHIFT!=rs 表示比較高RESIZE_STAMP_BITS位生成戳和rs是否相等，相同

//sc=rs+1 表示擴容結(jié)束

//sc==rs+MAX_RESIZERS 表示幫助線程線程已經(jīng)達到最大值了

//nt=nextTable -> 表示擴容已經(jīng)結(jié)束

//transferIndex<=0 表示所有的transfer任務(wù)都被領(lǐng)取完了伦连，沒有剩余的hash桶來給自己自己好這個線程來做transfer

if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||

sc == rs + MAX_RESIZERS || (nt = nextTable) ==  ||

transferIndex <= 0)

break;

if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))//當前線程嘗試幫助此次擴容雨饺，如果成功挣饥，則調(diào)用transfer

transfer(tab, nt);

}

// 如果當前沒有在擴容，那么rs肯定是一個正數(shù)沛膳，通過rs<<RESIZE_STAMP_SHIFT 將sc設(shè)置為一個負數(shù)扔枫，+2 表示有一個線程在執(zhí)行擴容

else if (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2))

transfer(tab, );

s = sumCount;// 重新計數(shù)，判斷是否需要開啟下一輪擴容

}

}

resizeStamp

這塊邏輯要理解起來锹安，也有一點復(fù)雜短荐。 resizeStamp用來生成一個和擴容有關(guān)的擴容戳，具體有什么作用呢叹哭？我們基于它的實現(xiàn)來做一個分析

static final int resizeStamp(int n) {

return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1));

}

Integer.numberOfLeadingZeros 這個方法是返回?zé)o符號整數(shù)n最高位非0位前面的0的個數(shù)

比如10的二進制是 0000 0000 0000 0000 0000 0000 0000 1010

那么這個方法返回的值就是28

根據(jù)resizeStamp的運算邏輯忍宋，我們來推演一下，假如n=16风罩，那么resizeStamp(16)=32796轉(zhuǎn)化為二進制是

[0000 0000 0000 0000 1000 0000 0001 1100]

接著再來看,當?shù)谝粋€線程嘗試進行擴容的時候糠排，會執(zhí)行下面這段代碼

U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2)

rs左移16位，相當于原本的二進制低位變成了高位1000 0000 0001 1100 0000 0000 0000 0000

然后再+2 =1000 0000 0001 1100 0000 0000 0000 0000+10=1000 0000 0001 1100 0000 0000 0000 0010

高16位代表擴容的標記超升、低16位代表并行擴容的線程數(shù)

? 這樣來存儲有什么好處呢入宦？

1. 首先在CHM中是支持并發(fā)擴容的，也就是說如果當前的數(shù)組需要進行擴容操作室琢，可以由多個線程來共同負責(zé)乾闰，這塊后續(xù)會單獨講

2. 可以保證每次擴容都生成唯一的生成戳，每次新的擴容盈滴，都有一個不同的n涯肩，這個生成戳就是根據(jù)n來計算出來的一個數(shù)字，n不同巢钓，這個數(shù)字也不同

? 第一個線程嘗試擴容的時候病苗，為什么是+2

因為1表示初始化，2表示一個線程在執(zhí)行擴容症汹，而且對sizeCtl的操作都是基于位運算的硫朦，所以不會關(guān)心它本身的數(shù)值是多少，只關(guān)心它在二進制上的數(shù)值烈菌，而sc + 1會在

低16位上加1阵幸。

transfer

擴容是ConcurrentHashMap的精華之一，擴容操作的核心在于數(shù)據(jù)的轉(zhuǎn)移芽世，在單線程環(huán)境下數(shù)據(jù)的轉(zhuǎn)移很簡單挚赊，無非就是把舊數(shù)組中的數(shù)據(jù)遷移到新的數(shù)組。但是這在多線程環(huán)境下济瓢，在擴容的時候其他線程也可能正在添加元素荠割，這時又觸發(fā)了擴容怎么辦？可能大家想到的第一個解決方案是加互斥鎖，把轉(zhuǎn)移過程鎖住蔑鹦，雖然是可行的解決方案夺克，但是會帶來較大的性能開銷。因為互斥鎖會導(dǎo)致所有訪問臨界區(qū)的線程陷入到阻塞狀態(tài)嚎朽，持有鎖的線程耗時越長铺纽，其他競爭線程就會一直被阻塞，導(dǎo)致吞吐量較低哟忍。而且還可能導(dǎo)致死鎖狡门。 而ConcurrentHashMap并沒有直接加鎖，而是采用CAS實現(xiàn)無鎖的并發(fā)同步策略锅很，最精華的部分是它可以利用多線程來進行協(xié)同擴容 簡單來說其馏，它把Node數(shù)組當作多個線程之間共享的任務(wù)隊列，然后通過維護一個指針來劃分每個線程鎖負責(zé)的區(qū)間爆安，每個線程通過區(qū)間逆向遍歷來實現(xiàn)擴容叛复，一個已經(jīng)遷移完的bucket會被替換為一個ForwardingNode節(jié)點，標記當前bucket已經(jīng)被其他線程遷移完了扔仓。接下來分析一下它的源碼實現(xiàn) 1褐奥、fwd:這個類是個標識類，用于指向新表用的当辐，其他線程遇到這個類會主動跳過這個類抖僵，因為這個類要么就是擴容遷移正在進行，要么就是已經(jīng)完成擴容遷移缘揪，也就是這個類要保證線程安全，再進行操作义桂。 2找筝、advance:這個變量是用于提示代碼是否進行推進處理，也就是當前桶處理完慷吊，處理下一個桶的標識 3袖裕、finishing:這個變量用于提示擴容是否結(jié)束用的

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {

int n = tab.length, stride;

//將 (n>>>3相當于 n/8) 然后除以 CPU核心數(shù)。如果得到的結(jié)果小于 16溉瓶，那么就使用 16

// 這里的目的是讓每個 CPU 處理的桶一樣多急鳄，避免出現(xiàn)轉(zhuǎn)移任務(wù)不均勻的現(xiàn)象，如果桶較少的話堰酿，默認一個 CPU（一個線程）處理 16 個桶疾宏，也就是長度為16的時候，擴容的時候只會有一個線程來擴容

if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)

stride = MIN_TRANSFER_STRIDE; // subdivide range

//nextTab未初始化触创，nextTab是用來擴容的node數(shù)組

if (nextTab == ) { // initiating

try {

@SuppressWarnings("unchecked")

//新建一個n<<1原始table大小的nextTab,也就是32

Node<K,V> nt = (Node<K,V>[])new Node<?,?>[n << 1];

nextTab = nt;//賦值給nextTab

} catch (Throwable ex) { // try to cope with OOME

sizeCtl = Integer.MAX_VALUE; //擴容失敗坎藐，sizeCtl使用int的最大值

return;

}

nextTable = nextTab; //更新成員變量

transferIndex = n;//更新轉(zhuǎn)移下標，表示轉(zhuǎn)移時的下標

}

int nextn = nextTab.length;//新的tab的長度

// 創(chuàng)建一個 fwd 節(jié)點，表示一個正在被遷移的Node岩馍，并且它的hash值為-1(MOVED)碉咆，也就是前面我們在講putval方法的時候，會有一個判斷MOVED的邏輯蛀恩。它的作用是用來占位疫铜，表示原數(shù)組中位置i處的節(jié)點完成遷移以后，就會在i位置設(shè)置一個fwd來告訴其他線程這個位置已經(jīng)處理過了双谆，具體后續(xù)還會在講

ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);

// 首次推進為 true壳咕，如果等于 true，說明需要再次推進一個下標（i--）佃乘，反之囱井，如果是 false，那么就不能推進下標趣避，需要將當前的下標處理完畢才能繼續(xù)推進

boolean advance = true;

//判斷是否已經(jīng)擴容完成庞呕，完成就return，退出循環(huán)

boolean finishing = false; // to ensure sweep before committing nextTab

通過for自循環(huán)處理每個槽位中的鏈表元素程帕，默認advace為真住练，通過CAS設(shè)置transferIndex屬性值，并初始化i和bound值愁拭，i指當前處理的槽位序號讲逛，bound指需要處理的槽位邊界，先處理槽位15的節(jié)點岭埠；

for (int i = 0, bound = 0;;) {

// 這個循環(huán)使用CAS不斷嘗試為當前線程分配任務(wù)

// 直到分配成功或任務(wù)隊列已經(jīng)被全部分配完畢

// 如果當前線程已經(jīng)被分配過bucket區(qū)域

// 那么會通過--i指向下一個待處理bucket然后退出該循環(huán)

Node<K,V> f; int fh;

while (advance) {

int nextIndex, nextBound;

//--i表示下一個待處理的bucket盏混，如果它>=bound,表示當前線程已經(jīng)分配過bucket區(qū)域

if (--i >= bound || finishing)

advance = false;

else if ((nextIndex = transferIndex) <= 0) {//表示所有bucket已經(jīng)被分配完畢

i = -1;

advance = false;

}

//通過cas來修改TRANSFERINDEX,為當前線程分配任務(wù)，處理的節(jié)點區(qū)間為(nextBound,nextIndex)->(0,15)

else if (U.compareAndSwapInt

(this, TRANSFERINDEX, nextIndex,

nextBound = (nextIndex > stride ?

nextIndex - stride : 0))) {

bound = nextBound;//0

i = nextIndex - 1;//15

advance = false;

}

}

//i<0說明已經(jīng)遍歷完舊的數(shù)組惜论，也就是當前線程已經(jīng)處理完所有負責(zé)的bucket

if (i < 0 || i >= n || i + n >= nextn) {

int sc;

if (finishing) {//如果完成了擴容

nextTable = ;//刪除成員變量

table = nextTab;//更新table數(shù)組

sizeCtl = (n << 1) - (n >>> 1);//更新閾值(32*0.75=24)

return;

}

// sizeCtl 在遷移前會設(shè)置為 (rs << RESIZE_STAMP_SHIFT) + 2 

// 然后许赃，每增加一個線程參與遷移就會將 sizeCtl 加 1，

// 這里使用 CAS 操作對 sizeCtl 的低16位進行減 1馆类，代表做完了屬于自己的任務(wù)

if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {

第一個擴容的線程混聊，執(zhí)行transfer方法之前，會設(shè)置 sizeCtl = (resizeStamp(n) << RESIZE_STAMP_SHIFT) + 2)

后續(xù)幫其擴容的線程乾巧，執(zhí)行transfer方法之前句喜，會設(shè)置 sizeCtl = sizeCtl+1

每一個退出transfer的方法的線程，退出之前沟于，會設(shè)置 sizeCtl = sizeCtl-1

那么最后一個線程退出時：必然有

sc == (resizeStamp(n) << RESIZE_STAMP_SHIFT) + 2)咳胃，即 (sc - 2) == resizeStamp(n) << RESIZE_STAMP_SHIFT 

// 如果 sc - 2 不等于標識符左移 16 位。如果他們相等了社裆，說明沒有線程在幫助他們擴容了拙绊。也就是說，擴容結(jié)束了。

if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)

return;

// 如果相等标沪，擴容結(jié)束了榄攀，更新 finising 變量

finishing = advance = true;

// 再次循環(huán)檢查一下整張表

i = n; // recheck before commit

}

}

// 如果位置 i 處是空的，沒有任何節(jié)點金句，那么放入剛剛初始化的 ForwardingNode ”空節(jié)點“

else if ((f = tabAt(tab, i)) == )

advance = casTabAt(tab, i, , fwd);

//表示該位置已經(jīng)完成了遷移檩赢，也就是如果線程A已經(jīng)處理過這個節(jié)點，那么線程B處理這個節(jié)點時违寞，hash值一定為MOVED

else if ((fh = f.hash) == MOVED) 

advance = true; // already processed 

}

}

擴容過程圖解

ConcurrentHashMap支持并發(fā)擴容贞瞒，實現(xiàn)方式是，把Node數(shù)組進行拆分趁曼，讓每個線程處理自己的區(qū)域军浆，假設(shè)table數(shù)組總長度是64，默認情況下挡闰，那么每個線程可以分到16個bucket乒融。 然后每個線程處理的范圍，按照倒序來做遷移 通過for自循環(huán)處理每個槽位中的鏈表元素摄悯，默認advace為真赞季，通過CAS設(shè)置transferIndex屬性值，并初始化i和bound值奢驯，i指當前處理的槽位序號申钩，bound指需要處理的槽位邊界，先處理槽位31的節(jié)點瘪阁； （bound,i） =(16,31) 從31的位置往前推動撒遣。

深度分析ConcurrentHashMap原理分析

假設(shè)這個時候ThreadA在進行transfer，那么邏輯圖表示如下

深度分析ConcurrentHashMap原理分析

在當前假設(shè)條件下管跺，槽位15中沒有節(jié)點愉舔，則通過CAS插入在第二步中初始化的ForwardingNode節(jié)點，用于告訴其它線程該槽位已經(jīng)處理過了伙菜；

深度分析ConcurrentHashMap原理分析

sizeCtl擴容退出機制

在擴容操作transfer的第2414行，代碼如下

if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {

if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)

每存在一個線程執(zhí)行完擴容操作命迈，就通過cas執(zhí)行sc-1贩绕。

接著判斷(sc-2) !=resizeStamp(n) << RESIZE_STAMP_SHIFT ; 如果相等，表示當前為整個擴容操作的 最后一個線程壶愤，那么意味著整個擴容操作就結(jié)束了淑倾；如果不想等，說明還得繼續(xù)

這么做的目的征椒，一方面是防止不同擴容之間出現(xiàn)相同的sizeCtl娇哆，另外一方面，還可以避免sizeCtl的ABA問題導(dǎo)致的擴容重疊的情況

數(shù)據(jù)遷移階段的實現(xiàn)分析

通過分配好遷移的區(qū)間之后，開始對數(shù)據(jù)進行遷移碍讨。在看這段代碼之前治力，先來了解一下原理

synchronized (f) {//對數(shù)組該節(jié)點位置加鎖，開始處理數(shù)組該位置的遷移工作

if (tabAt(tab, i) == f) {//再做一次校驗

Node<K,V> ln, hn;//ln表示低位勃黍， hn表示高位;接下來這段代碼的作用是把鏈表拆分成兩部分宵统，0在低位，1在高位

if (fh >= 0) {//下面部分代碼原理點擊這里

int runBit = fh & n; 

Node<K,V> lastRun = f;

//遍歷當前bucket的鏈表覆获，目的是盡量重用Node鏈表尾部的一部分

for (Node<K,V> p = f.next; p != ; p = p.next) { 

int b = p.hash & n;

if (b != runBit) {

runBit = b;

lastRun = p;

}

}

if (runBit == 0) {如果最后更新的runBit是0马澈，設(shè)置低位節(jié)點

ln = lastRun;

hn = ;

}

else {//否則，設(shè)置高位節(jié)點

hn = lastRun;

ln = ;

}

//構(gòu)造高位以及低位的鏈表

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);

}

setTabAt(nextTab, i, ln);//將低位的鏈表放在i位置也就是不動

setTabAt(nextTab, i + n, hn);//將高位鏈表放在i+n位置

setTabAt(tab, i, fwd); // 把舊table的hash桶中放置轉(zhuǎn)發(fā)節(jié)點弄息，表明此hash桶已經(jīng)被處理

advance = true;

}

//紅黑樹的擴容部分暫時忽略

}

高低位原理分析

ConcurrentHashMap在做鏈表遷移時痊班，會用高低位來實現(xiàn)，這里有兩個問題要分析一下

如何實現(xiàn)高低位鏈表的區(qū)分 假如我們有這樣一個隊列

深度分析ConcurrentHashMap原理分析

第14個槽位插入新節(jié)點之后摹量，鏈表元素個數(shù)已經(jīng)達到了8涤伐，且數(shù)組長度為16，優(yōu)先通過擴容來緩解鏈表過長的問題荆永，擴容這塊的圖解稍后再分析废亭，先分析高低位擴容的原理 假如當前線程正在處理槽位為14的節(jié)點，它是一個鏈表結(jié)構(gòu)具钥，在代碼中豆村，首先定義兩個變量節(jié)點ln和hn，實際就是lowNode和HighNode骂删，分別保存hash值的第x位為0和不等于0的節(jié)點 通過fn&n可以把這個鏈表中的元素分為兩類掌动，A類是hash值的第X位為0，B類是hash值的第x位為不等于0（至于為什么要這么區(qū)分，稍后分析）好啰，并且通過lastRun記錄最后要處理的節(jié)點洛心。最終要達到的目的是，A類的鏈表保持位置不動眷射，B類的鏈表為14+16(擴容增加的長度)=30 我們把14槽位的鏈表單獨伶出來，我們用藍色表示 fn&n=0的節(jié)點佛掖，假如鏈表的分類是這樣

深度分析ConcurrentHashMap原理分析

for (Node<K,V> p = f.next; p != ; p = p.next) {

int b = p.hash & n;

if (b != runBit) {

runBit = b;

lastRun = p;

}

}

通過上面這段代碼遍歷妖碉，會記錄runBit以及l(fā)astRun，按照上面這個結(jié)構(gòu)芥被，那么runBit應(yīng)該是藍色節(jié)點欧宜，lastRun應(yīng)該是第6個節(jié)點

接著，再通過這段代碼進行遍歷拴魄，生成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;

if ((ph & n) == 0)

ln = new Node<K,V>(ph, pk, pv, ln);

else

hn = new Node<K,V>(ph, pk, pv, hn);

}

深度分析ConcurrentHashMap原理分析

接著冗茸，通過CAS操作席镀，把hn鏈放在i+n也就是14+16的位置，ln鏈保持原來的位置不動夏漱。并且設(shè)置當前節(jié)點為fwd豪诲，表示已經(jīng)被當前線程遷移完了

setTabAt(nextTab, i, ln);

setTabAt(nextTab, i + n, hn);

setTabAt(tab, i, fwd);

遷移完成以后的數(shù)據(jù)分布如下

深度分析ConcurrentHashMap原理分析

為什么要做高低位的劃分

要想了解這么設(shè)計的目的，我們需要從ConcurrentHashMap的根據(jù)下標獲取對象的算法來看麻蹋，在putVal方法中1018行

(f = tabAt(tab, i = (n - 1) & hash)) ==  

通過(n-1) & hash來獲得在table中的數(shù)組下標來獲取節(jié)點數(shù)據(jù)跛溉，【&運算是二進制運算符，1 & 1=1扮授，其他都為0】


假設(shè)我們的table長度是16芳室， 二進制是【0001 0000】，減一以后的二進制是 【0000 1111】

假如某個key的hash值=9刹勃，對應(yīng)的二進制是【0000 1001】堪侯，那么按照（n-1） & hash的算法

0000 1111 & 0000 1001 =0000 1001 ， 運算結(jié)果是9

當我們擴容以后荔仁，16變成了32伍宦，那么(n-1)的二進制是 【0001 1111】

仍然以hash值=9的二進制計算為例

0001 1111 & 0000 1001 =0000 1001 ，運算結(jié)果仍然是9


我們換一個數(shù)字乏梁，假如某個key的hash值是20次洼，對應(yīng)的二進制是【0001 0100】，仍然按照(n-1) & hash算法遇骑，分別在16為長度和32位長度下的計算結(jié)果

16位： 0000 1111 & 0001 0100=0000 0100

32位： 0001 1111 & 0001 0100 =0001 0100

從結(jié)果來看卖毁，同樣一個hash值，在擴容前和擴容之后落萎，得到的下標位置是不一樣的亥啦，這種情況當然是不允許出現(xiàn)的，所以在擴容的時候就需要考慮练链，

而使用高低位的遷移方式翔脱，就是解決這個問題.

大家可以看到，16位的結(jié)果到32位的結(jié)果媒鼓，正好增加了16.

比如 20 & 15=4 届吁、20 & 31=20 ； 4-20 =16

比如 60 & 15=12 绿鸣、60 & 31=28瓷产； 12-28=16

所以對于高位，直接增加擴容的長度枚驻，當下次hash獲取數(shù)組位置的時候，可以直接定位到對應(yīng)的位置株旷。

這個地方又是一個很巧妙的設(shè)計再登，直接通過高低位分類以后尔邓，就使得不需要在每次擴容的時候來重新計算hash，極大提升了效率锉矢。

擴容結(jié)束以后的退出機制

如果線程擴容結(jié)束梯嗽，那么需要退出，就會執(zhí)行transfer方法的如下代碼

//i<0說明已經(jīng)遍歷完舊的數(shù)組沽损，也就是當前線程已經(jīng)處理完所有負責(zé)的bucket

if (i < 0 || i >= n || i + n >= nextn) {

int sc;

if (finishing) {//如果完成了擴容

nextTable = ;//刪除成員變量

table = nextTab;//更新table數(shù)組

sizeCtl = (n << 1) - (n >>> 1);//更新閾值(32*0.75=24)

return;

}

// sizeCtl 在遷移前會設(shè)置為 (rs << RESIZE_STAMP_SHIFT) + 2 

// 然后灯节，每增加一個線程參與遷移就會將 sizeCtl 加 1，

// 這里使用 CAS 操作對 sizeCtl 的低16位進行減 1绵估，代表做完了屬于自己的任務(wù)

if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {

第一個擴容的線程炎疆，執(zhí)行transfer方法之前，會設(shè)置 sizeCtl = (resizeStamp(n) << RESIZE_STAMP_SHIFT) + 2)

后續(xù)幫其擴容的線程国裳，執(zhí)行transfer方法之前形入，會設(shè)置 sizeCtl = sizeCtl+1

每一個退出transfer的方法的線程，退出之前缝左，會設(shè)置 sizeCtl = sizeCtl-1

那么最后一個線程退出時：必然有

sc == (resizeStamp(n) << RESIZE_STAMP_SHIFT) + 2)亿遂，即 (sc - 2) == resizeStamp(n) << RESIZE_STAMP_SHIFT 

// 如果 sc - 2 不等于標識符左移 16 位。如果他們相等了渺杉，說明沒有線程在幫助他們擴容了蛇数。也就是說，擴容結(jié)束了是越。

if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)

return;

// 如果相等耳舅，擴容結(jié)束了，更新 finising 變量

finishing = advance = true;

// 再次循環(huán)檢查一下整張表

i = n; // recheck before commit

}

}

put方法第三階段

如果對應(yīng)的節(jié)點存在英妓，判斷這個節(jié)點的hash是不是等于MOVED(-1)挽放，說明當前節(jié)點是ForwardingNode節(jié)點， 意味著有其他線程正在進行擴容蔓纠，那么當前現(xiàn)在直接幫助它進行擴容辑畦，因此調(diào)用helpTransfer方法

else if ((fh = f.hash) == MOVED)

tab = helpTransfer(tab, f);

helpTransfer

從名字上來看，代表當前是去協(xié)助擴容

final Node<K,V> helpTransfer(Node<K,V>[] tab, Node<K,V> f) {

Node<K,V> nextTab; int sc;

// 判斷此時是否仍然在執(zhí)行擴容,nextTab=的時候說明擴容已經(jīng)結(jié)束了

if (tab !=  && (f instanceof ForwardingNode) &&

(nextTab = ((ForwardingNode<K,V>)f).nextTable) != ) {

int rs = resizeStamp(tab.length);//生成擴容戳

while (nextTab == nextTable && table == tab &&

(sc = sizeCtl) < 0) {//說明擴容還未完成的情況下不斷循環(huán)來嘗試將當前線程加入到擴容操作中

//下面部分的整個代碼表示擴容結(jié)束腿倚，直接退出循環(huán)

//transferIndex<=0表示所有的Node都已經(jīng)分配了線程

//sc=rs+MAX_RESIZERS 表示擴容線程數(shù)達到最大擴容線程數(shù)

//sc >>> RESIZE_STAMP_SHIFT !=rs纯出， 如果在同一輪擴容中，那么sc無符號右移比較高位和rs的值敷燎，那么應(yīng)該是相等的暂筝。如果不相等，說明擴容結(jié)束了

//sc==rs+1 表示擴容結(jié)束

if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||

sc == rs + MAX_RESIZERS || transferIndex <= 0)

break;//跳出循環(huán)

if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {//在低16位上增加擴容線程數(shù)

transfer(tab, nextTab);//幫助擴容

break;

}

}

return nextTab;

}

return table;//返回新的數(shù)組

}

put方法第四階段

這個方法的主要作用是硬贯，如果被添加的節(jié)點的位置已經(jīng)存在節(jié)點的時候焕襟，需要以鏈表的方式加入到節(jié)點中 如果當前節(jié)點已經(jīng)是一顆紅黑樹，那么就會按照紅黑樹的規(guī)則將當前節(jié)點加入到紅黑樹中

else { //進入到這個分支饭豹，說明f是當前nodes數(shù)組對應(yīng)位置節(jié)點的頭節(jié)點鸵赖，并且不為空

V oldVal = ;

synchronized (f) { //給對應(yīng)的頭結(jié)點加鎖

if (tabAt(tab, i) == f) {//再次判斷對應(yīng)下標位置是否為f節(jié)點

if (fh >= 0) { //頭結(jié)點的hash值大于0务漩，說明是鏈表

binCount = 1; //用來記錄鏈表的長度

for (Node<K,V> e = f;; ++binCount) {//遍歷鏈表

K ek;

//如果發(fā)現(xiàn)相同的key，則判斷是否需要進行值的覆蓋

if (e.hash == hash &&

((ek = e.key) == key ||

(ek !=  && key.equals(ek)))) {

oldVal = e.val;

if (!onlyIfAbsent) //默認情況下它褪，直接覆蓋舊的值

e.val = value;

break;

}

//一直遍歷到鏈表的最末端饵骨，直接把新的值加入到鏈表的最后面

Node<K,V> pred = e;

if ((e = e.next) == ) {

pred.next = new Node<K,V>(hash, key,

value, );

break;

}

}

}

//如果當前的f節(jié)點是一顆紅黑樹

else if (f instanceof TreeBin) {

Node<K,V> p;

binCount = 2;

//則調(diào)用紅黑樹的插入方法插入新的值

if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,

value)) != ) {

oldVal = p.val; //同樣，如果值已經(jīng)存在茫打，則直接替換

if (!onlyIfAbsent)

p.val = value;

}

}

}

}

put方法第四個階段

判斷鏈表的長度是否已經(jīng)達到臨界值8. 如果達到了臨界值居触，這個時候會根據(jù)當前數(shù)組的長度來決定是擴容還是將鏈表轉(zhuǎn)化為紅黑樹。也就是說如果當前數(shù)組的長度小于64老赤，就會先擴容轮洋。否則，會把當前鏈表轉(zhuǎn)化為紅黑樹

if (binCount != 0) {//說明上面在做鏈表操作

//如果鏈表長度已經(jīng)達到臨界值8 就需要把鏈表轉(zhuǎn)換為樹結(jié)構(gòu)

if (binCount >= TREEIFY_THRESHOLD)

treeifyBin(tab, i); 

if (oldVal != )//如果val是被替換的诗越，則返回替換之前的值

return oldVal;

break;

}

treeifyBin

在putVal的最后部分砖瞧，有一個判斷，如果鏈表長度大于8嚷狞，那么就會觸發(fā)擴容或者紅黑樹的轉(zhuǎn)化操作块促。

private final void treeifyBin(Node<K,V>[] tab, int index) {

Node<K,V> b; int n, sc;

if (tab != ) {

if ((n = tab.length) < MIN_TREEIFY_CAPACITY) //tab的長度是不是小于64，如果是床未，則執(zhí)行擴容

tryPresize(n << 1); 

else if ((b = tabAt(tab, index)) !=  && b.hash >= 0) {//否則竭翠，將當前鏈表轉(zhuǎn)化為紅黑樹結(jié)構(gòu)存儲

synchronized (b) {// 將鏈表轉(zhuǎn)換成紅黑樹

if (tabAt(tab, index) == b) {

TreeNode<K,V> hd = , tl = ;

for (Node<K,V> e = b; e != ; e = e.next) {

TreeNode<K,V> p =

new TreeNode<K,V>(e.hash, e.key, e.val,

, );

if ((p.prev = tl) == )

hd = p;

else

tl.next = p;

tl = p;

}

setTabAt(tab, index, new TreeBin<K,V>(hd));

}

}

}

}

}

tryPresize

tryPresize里面部分代碼和addCount的部分代碼類似，看起來會稍微簡單一些

private final void tryPresize(int size) {

//對size進行修復(fù),主要目的是防止傳入的值不是一個2次冪的整數(shù)薇搁，然后通過tableSizeFor來講入?yún)⑥D(zhuǎn)化為離該整數(shù)最近的2次冪

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;

//下面這段代碼和initTable是一樣的斋扰，如果table沒有初始化，則開始初始化

if (tab ==  || (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);//0.75

}

} finally {

sizeCtl = sc;

}

}

}

//

else if (c <= sc || n >= MAXIMUM_CAPACITY)

break;

else if (tab == table) {//這段代碼和addCount后部分代碼是一樣的啃洋，做輔助擴容操作

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) ==  ||

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, );

}

}

}