六、并發(fā)集合

1 ConcurrentHashMap

1.1 存儲結(jié)構(gòu)

ConcurrentHashMap 是線程安全的 HashMap，在 JDK1.8 中是以 CAS + synchronized 實現(xiàn)的線程安全。

• CAS：在沒有 hash 沖突時（Node 要放在數(shù)組上時）
• synchronized：在出現(xiàn) hash 沖突時（Node 存放的位置已經(jīng)有數(shù)據(jù)了）
• 存儲結(jié)構(gòu)：數(shù)組+鏈表+紅黑樹

1.2 存儲操作

1.2.1 put方法

public V put(K key, V value) {// 在調(diào)用put方法時，會調(diào)用putVal方法，第三個參數(shù)默認傳遞false// 在調(diào)用putIfAbsent時，會調(diào)用putVal方法，第三個參數(shù)傳遞true// false: 代表key一致時，直接覆蓋數(shù)據(jù)// true: 代表key一致時，什么都不做，key不存在正常添加(類似Redis的setnx)return putVal(key, value, false);
}

1.2.2 putVal方法-散列算法

final V putVal(K key, V value, boolean onlyIfAbsent) {// ConcurrentHashMap不允許key或者value出現(xiàn)為null的值，跟HashMap的區(qū)別if (key == null || value == null) throw new NullPointerException();// 根據(jù)key的hashCode計算出一個hash值，后期得出當前key-value要存儲在哪個數(shù)組索引位置int hash = spread(key.hashCode());int binCount = 0; // 一個標識，在后面有用// ...省略大量代碼
}
// 計算當前Node的hash值的方法
static final int spread(int h) {// 將key的hashCode值的高低16位進行^運算，最終又與HASH_BITS進行了&運算// 將高位的hash也參與到計算索引位置的運算當中，盡可能將數(shù)據(jù)打散// 為什么HashMap、ConcurrentHashMap，都要求數(shù)組長度為2^n// HASH_BITS讓hash值的最高位符號位肯定為0，代表當前hash值默認情況下一定是正數(shù)，因為hash值為負數(shù)時，有特殊的含義// static final int MOVED     = -1; // 代表當前hash位置的數(shù)據(jù)正在擴容// static final int TREEBIN   = -2; // 代表當前hash位置下掛載的是一個紅黑樹// static final int RESERVED  = -3; // 預(yù)留當前索引位置return (h ^ (h >>> 16)) & HASH_BITS;// 計算數(shù)組放到哪個索引位置的方法   (f = tabAt(tab, i = (n - 1) & hash)// n：是數(shù)組的長度
}
運算方式
00000000 00000000 00000000 00001111  - 15 (n - 1)
&
((00001101 00001101 00101111 10001111  - h^00000000 00000000 00001101 00001101  - h >>> 16)&01111111 11111111 11111111 11111111  - HASH_BITS
)

1.2.3 putVal方法-添加數(shù)據(jù)到數(shù)組&初始化數(shù)組

• 添加數(shù)據(jù)到數(shù)組：CAS
• 初始化數(shù)組：DCL + CAS

final V putVal(K key, V value, boolean onlyIfAbsent) {// 省略部分代碼...// 將Map的數(shù)組賦值給tab，死循環(huán)for (Node<K,V>[] tab = table;;) {// n: 數(shù)組長度；i: 當前Node需要存放的索引位置// f: 當前數(shù)組i索引位置的Node對象；fn: 當前數(shù)組i索引位置上數(shù)據(jù)的hash值Node<K,V> f; int n, i, fh;// 判斷當前數(shù)組是否還沒有初始化if (tab == null || (n = tab.length) == 0)tab = initTable();	// 將數(shù)組進行初始化// 基于 (n - 1) & hash 計算出當前Node需要存放在哪個索引位置// 基于tabAt獲取到i位置的數(shù)據(jù)else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {// 現(xiàn)在數(shù)組的i位置上沒有數(shù)據(jù)，基于CAS的方式將數(shù)據(jù)存在i位置上if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null)))break; // 如果成功，執(zhí)行break跳出循環(huán)，插入數(shù)據(jù)成功}// 判斷當前位置數(shù)據(jù)是否正在擴容else if ((fh = f.hash) == MOVED)tab = helpTransfer(tab, f);	// 讓當前插入數(shù)據(jù)的線程協(xié)助擴容// 省略部分代碼...}addCount(1L, binCount);return null;
}
// 初始化數(shù)組方法
private final Node<K,V>[] initTable() {Node<K,V>[] tab; int sc;// 再次判斷數(shù)組沒有初始化，并且完成tab的賦值while ((tab = table) == null || tab.length == 0) {// sizeCtl：是數(shù)組在初始化和擴容操作時的一個控制變量。// -1: 代表當前數(shù)組正在初始化；// 小于-1: 低16位代表當前數(shù)組正在擴容的線程個數(shù)(如果1個線程擴容，值為-2，如果2個線程擴容，值為-3)；// 0: 代表數(shù)組還沒初始化；// 大于0: 代表當前數(shù)組的擴容閾值，或者是當前數(shù)組的初始化大小// 將sizeCtl賦值給sc變量，并判斷是否小于0if ((sc = sizeCtl) < 0)Thread.yield(); // lost initialization race; just spin// 可以嘗試初始化數(shù)組，線程會以CAS的方式，將sizeCtl修改為-1，代表當前線程可以初始化數(shù)組else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {try {	// 嘗試初始化// 再次判斷當前數(shù)組是否已經(jīng)初始化完畢if ((tab = table) == null || tab.length == 0) {// 開始初始化: 如果sizeCtl > 0，就初始化sizeCtl長度的數(shù)組；如果sizeCtl == 0，就初始化默認的長度16int n = (sc > 0) ? sc : DEFAULT_CAPACITY;// 初始化數(shù)組Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];// 將初始化的數(shù)組nt，賦值給tab和tabletable = tab = nt;// sc賦值為了數(shù)組長度 - 數(shù)組長度 右移 2位    16 - 4 = 12，將sc賦值為下次擴容的閾值sc = n - (n >>> 2);}} finally {// 將賦值好的sc，設(shè)置給sizeCtlsizeCtl = sc;}break;}}return tab;
}

1.2.4 putVal方法-添加數(shù)據(jù)到鏈表

• 添加數(shù)據(jù)到鏈表：利用 synchronized 基于當前索引位置的Node，作為鎖對象

final V putVal(K key, V value, boolean onlyIfAbsent) {// 省略部分代碼...int binCount = 0;for (Node<K,V>[] tab = table;;) {// n: 數(shù)組長度；i: 當前Node需要存放的索引位置// f: 當前數(shù)組i索引位置的Node對象；fn: 當前數(shù)組i索引位置上數(shù)據(jù)的hash值Node<K,V> f; int n, i, fh;// 省略部分代碼...else {V oldVal = null;	// 聲明變量為oldValsynchronized (f) {	// 基于當前索引位置的Node，作為鎖對象// 判斷當前位置的數(shù)據(jù)還是之前的f么……（避免并發(fā)操作的安全問題）if (tabAt(tab, i) == f) {if (fh >= 0) {	// 再次判斷hash值是否大于0（不是樹）// binCount設(shè)置為1（在鏈表情況下，記錄鏈表長度的一個標識）binCount = 1;// 死循環(huán)，每循環(huán)一次，對binCountfor (Node<K,V> e = f;; ++binCount) { K ek;// 當前i索引位置的數(shù)據(jù)，是否和當前put的key的hash值一致if (e.hash == hash &&// 如果當前i索引位置數(shù)據(jù)的key和put的key == 返回為true// 或者equals相等((ek = e.key) == key || (ek != null && key.equals(ek)))) {// key一致，可能需要覆蓋數(shù)據(jù)，當前i索引位置數(shù)據(jù)的value賦值給oldValoldVal = e.val;// 如果傳入的是false，代表key一致，覆蓋value；如果傳入的是true，代表key一致，什么都不做if (!onlyIfAbsent)e.val = value; // 覆蓋valuebreak;}Node<K,V> pred = e;	// 拿到當前指定的Node對象// 將e指向下一個Node對象,如果next指向的是一個null，可以掛在當前Node下面if ((e = e.next) == null) {// 將hash，key，value封裝為Node對象，掛在pred的next上pred.next = new Node<K,V>(hash, key, value, null);break;}}}// 省略部分代碼...}}if (binCount != 0) {if (binCount >= TREEIFY_THRESHOLD)	// binCount是否大于8（鏈表長度是否 >= 8）// 嘗試轉(zhuǎn)為紅黑樹或者擴容// 基于treeifyBin方法和上面的if判斷，可以得知鏈表想要轉(zhuǎn)為紅黑樹，必須保證數(shù)組長度大于等于64，并且鏈表長度大于等于8// 如果數(shù)組長度沒有達到64的話，會首先將數(shù)組擴容treeifyBin(tab, i);if (oldVal != null)	// 如果出現(xiàn)了數(shù)據(jù)覆蓋的情況，返回之前的值return oldVal;break;}}}// 省略部分代碼...
}

為什么鏈表長度為8轉(zhuǎn)換為紅黑樹，不是能其他數(shù)值嘛？

因為泊松分布

The main disadvantage of per-bin locks is that other update* operations on other nodes in a bin list protected by the same* lock can stall, for example when user equals() or mapping* functions take a long time.  However, statistically, under* random hash codes, this is not a common problem.  Ideally, the* frequency of nodes in bins follows a Poisson distribution* (http://en.wikipedia.org/wiki/Poisson_distribution) with a* parameter of about 0.5 on average, given the resizing threshold* of 0.75, although with a large variance because of resizing* granularity. Ignoring variance, the expected occurrences of* list size k are (exp(-0.5) * pow(0.5, k) / factorial(k)). The* first values are:** 0:    0.60653066* 1:    0.30326533* 2:    0.07581633* 3:    0.01263606* 4:    0.00157952* 5:    0.00015795* 6:    0.00001316* 7:    0.00000094* 8:    0.00000006* more: less than 1 in ten million

1.3 擴容操作

1.3.1 treeifyBin方法觸發(fā)擴容

// 在鏈表長度大于等于8時，嘗試將鏈表轉(zhuǎn)為紅黑樹
private final void treeifyBin(Node<K,V>[] tab, int index) {Node<K,V> b; int n, sc;// 數(shù)組不能為空if (tab != null) {// 數(shù)組的長度n，是否小于64if ((n = tab.length) < MIN_TREEIFY_CAPACITY)// 如果數(shù)組長度小于64，不能將鏈表轉(zhuǎn)為紅黑樹，先嘗試擴容操作tryPresize(n << 1);// 省略部分代碼……}
}

1.3.2 tryPresize方法-針對putAll的初始化操作

// size是將之前的數(shù)組長度 左移 1位得到的結(jié)果
private final void tryPresize(int size) {// 如果擴容的長度達到了最大值，就使用最大值，否則需要保證數(shù)組的長度為2的n次冪// 這塊的操作，是為了初始化操作準備的，因為調(diào)用putAll方法時，也會觸發(fā)tryPresize方法// 如果剛剛new的ConcurrentHashMap直接調(diào)用了putAll方法的話，會通過tryPresize方法進行初始化int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :tableSizeFor(size + (size >>> 1) + 1);// 這些代碼和initTable一模一樣int sc;// 將sizeCtl的值賦值給sc，并判斷是否大于0，這里代表沒有初始化操作，也沒有擴容操作while ((sc = sizeCtl) >= 0) {// 將ConcurrentHashMap的table賦值給tab，并聲明數(shù)組長度nNode<K,V>[] tab = table; int n;// 數(shù)組是否需要初始化if (tab == null || (n = tab.length) == 0) {// 進來執(zhí)行初始化// sc是初始化長度，初始化長度如果比計算出來的c要大的話，直接使用sc，如果沒有sc大，說明sc無法容納下putAll中傳入的map，使用更大的數(shù)組長度n = (sc > c) ? sc : c;// 設(shè)置sizeCtl為-1，代表初始化操作if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {try {// 再次判斷數(shù)組的引用有沒有變化if (table == tab) {// 初始化數(shù)組Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];// 數(shù)組賦值table = nt;// 計算擴容閾值sc = n - (n >>> 2);}} finally {// 最終賦值給sizeCtlsizeCtl = sc;}}}// 如果計算出來的長度c小于等于c，或者數(shù)組長度大于等于最大長度，直接退出循環(huán)結(jié)束方法else if (c <= sc || n >= MAXIMUM_CAPACITY)break;// 省略部分代碼...}
}// 將c這個長度設(shè)置到最近的2的n次冪的值，   15 -> 16     17 -> 32
// c == size + (size >>> 1) + 1
// size = 17
00000000 00000000 00000000 00010001
+ 
00000000 00000000 00000000 00001000
+
00000000 00000000 00000000 00000001
// c = 26
00000000 00000000 00000000 00011010
private static final int tableSizeFor(int c) { // c = 26// 00000000 00000000 00000000 00011001int n = c - 1;// 00000000 00000000 00000000 00011001// 00000000 00000000 00000000 00001100// 00000000 00000000 00000000 00011101n |= n >>> 1;// 00000000 00000000 00000000 00011101// 00000000 00000000 00000000 00000111// 00000000 00000000 00000000 00011111n |= n >>> 2;// 00000000 00000000 00000000 00011111// 00000000 00000000 00000000 00000001// 00000000 00000000 00000000 00011111n |= n >>> 4;// 00000000 00000000 00000000 00011111// 00000000 00000000 00000000 00000000// 00000000 00000000 00000000 00011111n |= n >>> 8;// 00000000 00000000 00000000 00011111n |= n >>> 16;// 00000000 00000000 00000000 00100000return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;
}

1.3.3 tryPreSize方法-計算擴容戳&查看BUG

private final void tryPresize(int size) {// n：數(shù)組長度while ((sc = sizeCtl) >= 0) {// 省略部分代碼…// 判斷當前的tab是否和table一致else if (tab == table) {// 計算擴容標識戳，根據(jù)當前數(shù)組的長度計算一個16位的擴容戳// 第一個作用是為了保證后面的sizeCtl賦值時，保證sizeCtl為小于-1的負數(shù)// 第二個作用用來記錄當前是從什么長度開始擴容的int rs = resizeStamp(n);// BUG --- sc < 0，永遠進不去if (sc < 0) { // 如果sc小于0，代表有線程正在擴容// 省略部分代碼……協(xié)助擴容的代碼（進不來~~~~）}// 代表沒有線程正在擴容，我是第一個擴容的。else if (U.compareAndSwapInt(this, SIZECTL, sc,(rs << RESIZE_STAMP_SHIFT) + 2))// 省略部分代碼……第一個擴容的線程……}}
}
// 計算擴容標識戳
// 32 =  00000000 00000000 00000000 00100000
// Integer.numberOfLeadingZeros(32) = 26
// 1 << (RESIZE_STAMP_BITS - 1) 
// 00000000 00000000 10000000 00000000
// 00000000 00000000 00000000 00011010
// 00000000 00000000 10000000 00011010
static final int resizeStamp(int n) {return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1));
}

1.3.4 tryPreSize方法-對sizeCtl的修改&條件判斷的BUG

private final void tryPresize(int size) {// sc默認為sizeCtlwhile ((sc = sizeCtl) >= 0) {else if (tab == table) {// rs: 擴容戳  00000000 00000000 10000000 00011010int rs = resizeStamp(n);if (sc < 0) {// 說明有線程正在擴容，過來幫助擴容Node<K,V>[] nt;// 依然有BUG// 當前線程擴容時，老數(shù)組長度是否和我當前線程擴容時的老數(shù)組長度一致// 00000000 00000000 10000000 00011010if ((sc >>> RESIZE_STAMP_SHIFT) != rs  // 10000000 00011010 00000000 00000010 // 00000000 00000000 10000000 00011010// 這兩個判斷都是有問題的，核心問題就應(yīng)該先將rs左移16位，再追加當前值// 判斷當前擴容是否已經(jīng)即將結(jié)束|| sc == rs + 1   // sc == rs << 16 + 1 BUG// 判斷當前擴容的線程是否達到了最大限度|| sc == rs + MAX_RESIZERS   // sc == rs << 16 + MAX_RESIZERS BUG// 擴容已經(jīng)結(jié)束了|| (nt = nextTable) == null // 記錄遷移的索引位置，從高位往低位遷移，也代表擴容即將結(jié)束|| transferIndex <= 0)break;// 如果線程需要協(xié)助擴容，首先就是對sizeCtl進行+1操作，代表當前要進來一個線程協(xié)助擴容if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))// 上面的判斷沒進去的話，nt就代表新數(shù)組transfer(tab, nt);}// 是第一個來擴容的線程// 基于CAS將sizeCtl修改為  10000000 00011010 00000000 00000010 // 將擴容戳左移16位之后，符號位是1，就代碼這個值為負數(shù)，低16位在表示當前正在擴容的線程有多少個// 為什么低位值為2時，代表有一個線程正在擴容// 每一個線程擴容完畢后，會對低16位進行-1操作，當最后一個線程擴容完畢后，減1的結(jié)果還是-1，當值為-1時，要對老數(shù)組進行一波掃描，查看是否有遺漏的數(shù)據(jù)沒有遷移到新數(shù)組else if (U.compareAndSwapInt(this, SIZECTL, sc,(rs << RESIZE_STAMP_SHIFT) + 2))// 調(diào)用transfer方法，并且將第二個參數(shù)設(shè)置為null，就代表是第一次來擴容！transfer(tab, null);}}
}

1.3.5 transfer方法-計算每個線程遷移的長度

// 開始擴容   tab: oldTable
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {// n: 數(shù)組長度// stride: 每個線程一次性遷移多少數(shù)據(jù)到新數(shù)組int n = tab.length, stride;// 基于CPU的內(nèi)核數(shù)量來計算，每個線程一次性遷移多少長度的數(shù)據(jù)最合理// NCPU = 4// 舉個栗子：數(shù)組長度為1024 - 512 - 256 - 128 / 4 = 32// MIN_TRANSFER_STRIDE = 16，為每個線程遷移數(shù)據(jù)的最小長度// 根據(jù)CPU計算每個線程一次遷移多長的數(shù)據(jù)到新數(shù)組，如果結(jié)果大于16，使用計算結(jié)果。 如果結(jié)果小于16，就使用最小長度16if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)stride = MIN_TRANSFER_STRIDE; // 省略部分代碼...
}

1.3.6 transfer方法-構(gòu)建新數(shù)組&查看標識屬性

// 以32長度數(shù)組擴容到64位例子
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {// 省略部分代碼...// n: 老數(shù)組長度   32// stride: 步長   16// 第一個進來擴容的線程需要把新數(shù)組構(gòu)建出來if (nextTab == null) {try {// 將原數(shù)組長度左移一位，構(gòu)建新數(shù)組長度Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];// 賦值操作nextTab = nt;} catch (Throwable ex) {   // 到這說明已經(jīng)達到數(shù)組長度的最大取值范圍sizeCtl = Integer.MAX_VALUE;// 設(shè)置sizeCtl后直接結(jié)束return;}// 對成員變量的新數(shù)組賦值nextTable = nextTab;// 遷移數(shù)據(jù)時，用到的標識，默認值為老數(shù)組長度transferIndex = n;   // 32}// 新數(shù)組長度int nextn = nextTab.length;  // 64// 在老數(shù)組遷移完數(shù)據(jù)后，做的標識ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);// 遷移數(shù)據(jù)時，需要用到的標識boolean advance = true;boolean finishing = false; // 省略部分代碼...
}

1.3.7 transfer方法-線程領(lǐng)取遷移任務(wù)

// 以32長度擴容到64位為例子
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {// 省略部分代碼…// n: 32// stride: 16int n = tab.length, stride;if (nextTab == null) { // 省略部分代碼…nextTable = nextTab;	// 新數(shù)組// transferIndex：0transferIndex = n;}// nextn：64int nextn = nextTab.length;ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);// advance：true，代表當前線程需要接收任務(wù)，然后再執(zhí)行遷移；如果為false，代表已經(jīng)接收完任務(wù)boolean advance = true;boolean finishing = false; // 是否遷移結(jié)束// i = 15     代表當前線程遷移數(shù)據(jù)的索引值for (int i = 0, bound = 0;;) {Node<K,V> f; int fh;	// f = null，fh = 0while (advance) {		// 當前線程要接收任務(wù)// nextIndex = 16，nextBound = 16int nextIndex, nextBound;// 對i進行--，并且判斷當前任務(wù)是否處理完畢！if (--i >= bound || finishing) // 第一次進來，這兩個判斷肯定進不去advance = false;// 判斷transferIndex是否小于等于0，代表沒有任務(wù)可領(lǐng)取，結(jié)束了// 在線程領(lǐng)取任務(wù)會，會對transferIndex進行修改，修改為transferIndex - stride// 在任務(wù)都領(lǐng)取完之后，transferIndex肯定是小于等于0的，代表沒有遷移數(shù)據(jù)的任務(wù)可以領(lǐng)取else if ((nextIndex = transferIndex) <= 0) {i = -1