1.ConcurrentHashmap簡(jiǎn)介
在使用HashMap時(shí)在多線程情況下擴(kuò)容會(huì)出現(xiàn)CPU接近100%的情況，因?yàn)閔ashmap并不是線程安全的，通常我們可以使用在java體系中古老的hashtable類(lèi)，該類(lèi)基本上所有的方法都采用synchronized進(jìn)行線程安全的控制，可想而知，在高并發(fā)的情況下，每次只有一個(gè)線程能夠獲取對(duì)象監(jiān)視器鎖，這樣的并發(fā)性能的確不令人滿意。另外一種方式通過(guò)Collections的Map<K,V> synchronizedMap(Map<K,V> m)將hashmap包裝成一個(gè)線程安全的map。比如SynchronzedMap的put方法源碼為：
public V put(K key, V value) {
synchronized (mutex) {return m.put(key, value);}
}
實(shí)際上SynchronizedMap實(shí)現(xiàn)依然是采用synchronized獨(dú)占式鎖進(jìn)行線程安全的并發(fā)控制的。同樣，這種方案的性能也是令人不太滿意的。針對(duì)這種境況，Doug Lea大師不遺余力的為我們創(chuàng)造了一些線程安全的并發(fā)容器，讓每一個(gè)java開(kāi)發(fā)人員倍感幸福。相對(duì)于hashmap來(lái)說(shuō)，ConcurrentHashMap就是線程安全的map，其中利用了鎖分段的思想提高了并發(fā)度。
ConcurrentHashMap在JDK1.6的版本網(wǎng)上資料很多，有興趣的可以去看看。
JDK 1.6版本關(guān)鍵要素：
- segment繼承了ReentrantLock充當(dāng)鎖的角色，為每一個(gè)segment提供了線程安全的保障；
- segment維護(hù)了哈希散列表的若干個(gè)桶，每個(gè)桶由HashEntry構(gòu)成的鏈表。
而到了JDK 1.8的ConcurrentHashMap就有了很大的變化，光是代碼量就足足增加了很多。1.8版本舍棄了segment，并且大量使用了synchronized，以及CAS無(wú)鎖操作以保證ConcurrentHashMap操作的線程安全性。至于為什么不用ReentrantLock而是Synchronzied呢？實(shí)際上，synchronzied做了很多的優(yōu)化，包括偏向鎖，輕量級(jí)鎖，重量級(jí)鎖，可以依次向上升級(jí)鎖狀態(tài)，但不能降級(jí)（關(guān)于synchronized可以看這篇文章），因此，使用synchronized相較于ReentrantLock的性能會(huì)持平甚至在某些情況更優(yōu)，具體的性能測(cè)試可以去網(wǎng)上查閱一些資料。另外，底層數(shù)據(jù)結(jié)構(gòu)改變?yōu)椴捎脭?shù)組+鏈表+紅黑樹(shù)的數(shù)據(jù)形式。
2.關(guān)鍵屬性及類(lèi)
在了解ConcurrentHashMap的具體方法實(shí)現(xiàn)前，我們需要系統(tǒng)的來(lái)看一下幾個(gè)關(guān)鍵的地方。
ConcurrentHashMap的關(guān)鍵屬性
table
volatile Node<K,V>[] table://裝載Node的數(shù)組，作為ConcurrentHashMap的數(shù)據(jù)容器，采用懶加載的方式，直到第一次插入數(shù)據(jù)的時(shí)候才會(huì)進(jìn)行初始化操作，數(shù)組的大小總是為2的冪次方。nextTable
volatile Node<K,V>[] nextTable; //擴(kuò)容時(shí)使用，平時(shí)為null，只有在擴(kuò)容的時(shí)候才為非nullsizeCtl
volatile int sizeCtl;
該屬性用來(lái)控制table數(shù)組的大小，根據(jù)是否初始化和是否正在擴(kuò)容有幾種情況：
當(dāng)值為負(fù)數(shù)時(shí)：如果為-1表示正在初始化，如果為-N則表示當(dāng)前正有N-1個(gè)線程進(jìn)行擴(kuò)容操作；
當(dāng)值為正數(shù)時(shí)：如果當(dāng)前數(shù)組為null的話表示table在初始化過(guò)程中，sizeCtl表示為需要新建數(shù)組的長(zhǎng)度；
若已經(jīng)初始化了，表示當(dāng)前數(shù)據(jù)容器（table數(shù)組）可用容量也可以理解成臨界值（插入節(jié)點(diǎn)數(shù)超過(guò)了該臨界值就需要擴(kuò)容），具體指為數(shù)組的長(zhǎng)度n 乘以 加載因子loadFactor；
當(dāng)值為0時(shí)，即數(shù)組長(zhǎng)度為默認(rèn)初始值。-
sun.misc.Unsafe U
在ConcurrentHashMapde的實(shí)現(xiàn)中可以看到大量的U.compareAndSwapXXXX的方法去修改ConcurrentHashMap的一些屬性。這些方法實(shí)際上是利用了CAS算法保證了線程安全性，這是一種樂(lè)觀策略，假設(shè)每一次操作都不會(huì)產(chǎn)生沖突，當(dāng)且僅當(dāng)沖突發(fā)生的時(shí)候再去嘗試。而CAS操作依賴于現(xiàn)代處理器指令集，通過(guò)底層CMPXCHG指令實(shí)現(xiàn)。CAS(V,O,N)核心思想為：若當(dāng)前變量實(shí)際值V與期望的舊值O相同，則表明該變量沒(méi)被其他線程進(jìn)行修改，因此可以安全的將新值N賦值給變量；若當(dāng)前變量實(shí)際值V與期望的舊值O不相同，則表明該變量已經(jīng)被其他線程做了處理，此時(shí)將新值N賦給變量操作就是不安全的，在進(jìn)行重試。而在大量的同步組件和并發(fā)容器的實(shí)現(xiàn)中使用CAS是通過(guò)sun.misc.Unsafe類(lèi)實(shí)現(xiàn)的，該類(lèi)提供了一些可以直接操控內(nèi)存和線程的底層操作，可以理解為java中的“指針”。該成員變量的獲取是在靜態(tài)代碼塊中：static { try { U = sun.misc.Unsafe.getUnsafe(); ....... } catch (Exception e) { throw new Error(e); } }
ConcurrentHashMap中關(guān)鍵內(nèi)部類(lèi)
-
Node
Node類(lèi)實(shí)現(xiàn)了Map.Entry接口，主要存放key-value對(duì)，并且具有next域static class Node<K,V> implements Map.Entry<K,V> { final int hash; final K key; volatile V val; volatile Node<K,V> next; ...... }
另外可以看出很多屬性都是用volatile進(jìn)行修飾的，也就是為了保證內(nèi)存可見(jiàn)性。
-
TreeNode
樹(shù)節(jié)點(diǎn)，繼承于承載數(shù)據(jù)的Node類(lèi)。而紅黑樹(shù)的操作是針對(duì)TreeBin類(lèi)的，從該類(lèi)的注釋也可以看出，也就是TreeBin會(huì)將TreeNode進(jìn)行再一次封裝** * Nodes for use in TreeBins */ static final class TreeNode<K,V> extends Node<K,V> { TreeNode<K,V> parent; // red-black tree links TreeNode<K,V> left; TreeNode<K,V> right; TreeNode<K,V> prev; // needed to unlink next upon deletion boolean red; ...... } -
TreeBin
這個(gè)類(lèi)并不負(fù)責(zé)包裝用戶的key、value信息，而是包裝的很多TreeNode節(jié)點(diǎn)。實(shí)際的ConcurrentHashMap“數(shù)組”中，存放的是TreeBin對(duì)象，而不是TreeNode對(duì)象。static final class TreeBin<K,V> extends Node<K,V> { TreeNode<K,V> root; volatile TreeNode<K,V> first; volatile Thread waiter; volatile int lockState; // values for lockState static final int WRITER = 1; // set while holding write lock static final int WAITER = 2; // set when waiting for write lock static final int READER = 4; // increment value for setting read lock ...... } -
ForwardingNode
在擴(kuò)容時(shí)才會(huì)出現(xiàn)的特殊節(jié)點(diǎn)，其key,value,hash全部為null。并擁有nextTable指針引用新的table數(shù)組。static final class ForwardingNode<K,V> extends Node<K,V> { final Node<K,V>[] nextTable; ForwardingNode(Node<K,V>[] tab) { super(MOVED, null, null, null); this.nextTable = tab; } ..... }
CAS關(guān)鍵操作
在上面我們提及到在ConcurrentHashMap中會(huì)大量使用CAS修改它的屬性和一些操作。因此，在理解ConcurrentHashMap的方法前我們需要了解下面幾個(gè)常用的利用CAS算法來(lái)保障線程安全的操作。
-
tabAt
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); }
該方法用來(lái)獲取table數(shù)組中索引為i的Node元素。
-
casTabAt
static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i, Node<K,V> c, Node<K,V> v) { return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v); }利用CAS操作設(shè)置table數(shù)組中索引為i的元素
-
setTabAt
static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) { U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v); }該方法用來(lái)設(shè)置table數(shù)組中索引為i的元素
3.重點(diǎn)方法講解
在熟悉上面的這核心信息之后，我們接下來(lái)就來(lái)依次看看幾個(gè)常用的方法是怎樣實(shí)現(xiàn)的。
3.1 實(shí)例構(gòu)造器方法
在使用ConcurrentHashMap第一件事自然而然就是new 出來(lái)一個(gè)ConcurrentHashMap對(duì)象，一共提供了如下幾個(gè)構(gòu)造器方法：
// 1. 構(gòu)造一個(gè)空的map，即table數(shù)組還未初始化，初始化放在第一次插入數(shù)據(jù)時(shí)，默認(rèn)大小為16
ConcurrentHashMap()
// 2. 給定map的大小
ConcurrentHashMap(int initialCapacity)
// 3. 給定一個(gè)map
ConcurrentHashMap(Map<? extends K, ? extends V> m)
// 4. 給定map的大小以及加載因子
ConcurrentHashMap(int initialCapacity, float loadFactor)
// 5. 給定map大小，加載因子以及并發(fā)度（預(yù)計(jì)同時(shí)操作數(shù)據(jù)的線程）
ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel)
ConcurrentHashMap一共給我們提供了5中構(gòu)造器方法，具體使用請(qǐng)看注釋，我們來(lái)看看第2種構(gòu)造器，傳入指定大小時(shí)的情況，該構(gòu)造器源碼為：
public ConcurrentHashMap(int initialCapacity) {
//1. 小于0直接拋異常
if (initialCapacity < 0)
throw new IllegalArgumentException();
//2. 判斷是否超過(guò)了允許的最大值，超過(guò)了話則取最大值，否則再對(duì)該值進(jìn)一步處理
int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
MAXIMUM_CAPACITY :
tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
//3. 賦值給sizeCtl
this.sizeCtl = cap;
}
這段代碼的邏輯請(qǐng)看注釋，很容易理解，如果小于0就直接拋出異常，如果指定值大于了所允許的最大值的話就取最大值，否則，在對(duì)指定值做進(jìn)一步處理。最后將cap賦值給sizeCtl,關(guān)于sizeCtl的說(shuō)明請(qǐng)看上面的說(shuō)明，當(dāng)調(diào)用構(gòu)造器方法之后，sizeCtl的大小應(yīng)該就代表了ConcurrentHashMap的大小，即table數(shù)組長(zhǎng)度。tableSizeFor做了哪些事情了？源碼為：
/**
* Returns a power of two table size for the given desired capacity.
* See Hackers Delight, sec 3.2
*/
private static final int tableSizeFor(int c) {
int n = c - 1;
n |= n >>> 1;
n |= n >>> 2;
n |= n >>> 4;
n |= n >>> 8;
n |= n >>> 16;
return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;
}
通過(guò)注釋就很清楚了，該方法會(huì)將調(diào)用構(gòu)造器方法時(shí)指定的大小轉(zhuǎn)換成一個(gè)2的冪次方數(shù)，也就是說(shuō)ConcurrentHashMap的大小一定是2的冪次方，比如，當(dāng)指定大小為18時(shí)，為了滿足2的冪次方特性，實(shí)際上concurrentHashMapd的大小為2的5次方（32）。另外，需要注意的是，調(diào)用構(gòu)造器方法的時(shí)候并未構(gòu)造出table數(shù)組（可以理解為ConcurrentHashMap的數(shù)據(jù)容器），只是算出table數(shù)組的長(zhǎng)度，當(dāng)?shù)谝淮蜗駽oncurrentHashMap插入數(shù)據(jù)的時(shí)候才真正的完成初始化創(chuàng)建table數(shù)組的工作。
3.2 initTable方法
直接上源碼：
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
if ((sc = sizeCtl) < 0)
// 1. 保證只有一個(gè)線程正在進(jìn)行初始化操作
Thread.yield(); // lost initialization race; just spin
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
if ((tab = table) == null || tab.length == 0) {
// 2. 得出數(shù)組的大小
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
// 3. 這里才真正的初始化數(shù)組
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;
// 4. 計(jì)算數(shù)組中可用的大?。簩?shí)際大小n*0.75（加載因子）
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}
代碼的邏輯請(qǐng)見(jiàn)注釋，有可能存在一個(gè)情況是多個(gè)線程同時(shí)走到這個(gè)方法中，為了保證能夠正確初始化，在第1步中會(huì)先通過(guò)if進(jìn)行判斷，若當(dāng)前已經(jīng)有一個(gè)線程正在初始化即sizeCtl值變?yōu)?1，這個(gè)時(shí)候其他線程在If判斷為true從而調(diào)用Thread.yield()讓出CPU時(shí)間片。正在進(jìn)行初始化的線程會(huì)調(diào)用U.compareAndSwapInt方法將sizeCtl改為-1即正在初始化的狀態(tài)。另外還需要注意的事情是，在第四步中會(huì)進(jìn)一步計(jì)算數(shù)組中可用的大小即為數(shù)組實(shí)際大小n乘以加載因子0.75.可以看看這里乘以0.75是怎么算的，0.75為四分之三，這里n - (n >>> 2)是不是剛好是n-(1/4)n=(3/4)n，挺有意思的吧:)。如果選擇是無(wú)參的構(gòu)造器的話，這里在new Node數(shù)組的時(shí)候會(huì)使用默認(rèn)大小為DEFAULT_CAPACITY（16），然后乘以加載因子0.75為12，也就是說(shuō)數(shù)組的可用大小為12。
3.3 put方法
使用ConcurrentHashMap最長(zhǎng)用的也應(yīng)該是put和get方法了吧，我們先來(lái)看看put方法是怎樣實(shí)現(xiàn)的。調(diào)用put方法時(shí)實(shí)際具體實(shí)現(xiàn)是putVal方法，源碼如下：
/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
//1. 計(jì)算key的hash值
int hash = spread(key.hashCode());
int binCount = 0;
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
//2. 如果當(dāng)前table還沒(méi)有初始化先調(diào)用initTable方法將tab進(jìn)行初始化
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//3. tab中索引為i的位置的元素為null，則直接使用CAS將值插入即可
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. 當(dāng)前正在擴(kuò)容
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
else {
V oldVal = null;
synchronized (f) {
if (tabAt(tab, i) == f) {
//5. 當(dāng)前為鏈表，在鏈表中插入新的鍵值對(duì)
if (fh >= 0) {
binCount = 1;
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;
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;
}
}
}
// 6.當(dāng)前為紅黑樹(shù)，將新的鍵值對(duì)插入到紅黑樹(shù)中
else if (f instanceof TreeBin) {
Node<K,V> p;
binCount = 2;
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
// 7.插入完鍵值對(duì)后再根據(jù)實(shí)際大小看是否需要轉(zhuǎn)換成紅黑樹(shù)
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
//8.對(duì)當(dāng)前容量大小進(jìn)行檢查，如果超過(guò)了臨界值（實(shí)際大小*加載因子）就需要擴(kuò)容
addCount(1L, binCount);
return null;
}
put方法的代碼量有點(diǎn)長(zhǎng)，我們按照上面的分解的步驟一步步來(lái)看。從整體而言，為了解決線程安全的問(wèn)題，ConcurrentHashMap使用了synchronzied和CAS的方式。在之前了解過(guò)HashMap以及1.8版本之前的ConcurrenHashMap都應(yīng)該知道ConcurrentHashMap結(jié)構(gòu)圖，為了方面下面的講解這里先直接給出，如果對(duì)這有疑問(wèn)的話，可以在網(wǎng)上隨便搜搜即可。
如圖（圖片摘自網(wǎng)絡(luò)），ConcurrentHashMap是一個(gè)哈希桶數(shù)組，如果不出現(xiàn)哈希沖突的時(shí)候，每個(gè)元素均勻的分布在哈希桶數(shù)組中。當(dāng)出現(xiàn)哈希沖突的時(shí)候，是標(biāo)準(zhǔn)的鏈地址的解決方式，將hash值相同的節(jié)點(diǎn)構(gòu)成鏈表的形式，稱為“拉鏈法”，另外，在1.8版本中為了防止拉鏈過(guò)長(zhǎng)，當(dāng)鏈表的長(zhǎng)度大于8的時(shí)候會(huì)將鏈表轉(zhuǎn)換成紅黑樹(shù)。table數(shù)組中的每個(gè)元素實(shí)際上是單鏈表的頭結(jié)點(diǎn)或者紅黑樹(shù)的根節(jié)點(diǎn)。當(dāng)插入鍵值對(duì)時(shí)首先應(yīng)該定位到要插入的桶，即插入table數(shù)組的索引i處。那么，怎樣計(jì)算得出索引i呢？當(dāng)然是根據(jù)key的hashCode值。
- spread()重哈希，以減小Hash沖突
我們知道對(duì)于一個(gè)hash表來(lái)說(shuō)，hash值分散的不夠均勻的話會(huì)大大增加哈希沖突的概率，從而影響到hash表的性能。因此通過(guò)spread方法進(jìn)行了一次重hash從而大大減小哈希沖突的可能性。spread方法為：
static final int spread(int h) {
return (h ^ (h >>> 16)) & HASH_BITS;
}
該方法主要是將key的hashCode的低16位于高16位進(jìn)行異或運(yùn)算，這樣不僅能夠使得hash值能夠分散能夠均勻減小hash沖突的概率，另外只用到了異或運(yùn)算，在性能開(kāi)銷(xiāo)上也能兼顧，做到平衡的trade-off。
2.初始化table
緊接著到第2步，會(huì)判斷當(dāng)前table數(shù)組是否初始化了，沒(méi)有的話就調(diào)用initTable進(jìn)行初始化，該方法在上面已經(jīng)講過(guò)了。
3.能否直接將新值插入到table數(shù)組中
從上面的結(jié)構(gòu)示意圖就可以看出存在這樣一種情況，如果插入值待插入的位置剛好所在的table數(shù)組為null的話就可以直接將值插入即可。那么怎樣根據(jù)hash確定在table中待插入的索引i呢？很顯然可以通過(guò)hash值與數(shù)組的長(zhǎng)度取模操作，從而確定新值插入到數(shù)組的哪個(gè)位置。而之前我們提過(guò)ConcurrentHashMap的大小總是2的冪次方，(n - 1) & hash運(yùn)算等價(jià)于對(duì)長(zhǎng)度n取模，也就是hash%n，但是位運(yùn)算比取模運(yùn)算的效率要高很多，Doug lea大師在設(shè)計(jì)并發(fā)容器的時(shí)候也是將性能優(yōu)化到了極致，令人欽佩。
確定好數(shù)組的索引i后，就可以可以tabAt()方法（該方法在上面已經(jīng)說(shuō)明了，有疑問(wèn)可以回過(guò)頭去看看）獲取該位置上的元素，如果當(dāng)前Node f為null的話，就可以直接用casTabAt方法將新值插入即可。
4.當(dāng)前是否正在擴(kuò)容
如果當(dāng)前節(jié)點(diǎn)不為null，且該節(jié)點(diǎn)為特殊節(jié)點(diǎn)（forwardingNode）的話，就說(shuō)明當(dāng)前concurrentHashMap正在進(jìn)行擴(kuò)容操作，關(guān)于擴(kuò)容操作，下面會(huì)作為一個(gè)具體的方法進(jìn)行講解。那么怎樣確定當(dāng)前的這個(gè)Node是不是特殊的節(jié)點(diǎn)了？是通過(guò)判斷該節(jié)點(diǎn)的hash值是不是等于-1（MOVED）,代碼為(fh = f.hash) == MOVED，對(duì)MOVED的解釋在源碼上也寫(xiě)的很清楚了：
static final int MOVED = -1; // hash for forwarding nodes
5.當(dāng)table[i]為鏈表的頭結(jié)點(diǎn)，在鏈表中插入新值
在table[i]不為null并且不為forwardingNode時(shí)，并且當(dāng)前Node f的hash值大于0（fh >= 0）的話說(shuō)明當(dāng)前節(jié)點(diǎn)f為當(dāng)前桶的所有的節(jié)點(diǎn)組成的鏈表的頭結(jié)點(diǎn)。那么接下來(lái)，要想向ConcurrentHashMap插入新值的話就是向這個(gè)鏈表插入新值。通過(guò)synchronized (f)的方式進(jìn)行加鎖以實(shí)現(xiàn)線程安全性。往鏈表中插入節(jié)點(diǎn)的部分代碼為：
if (fh >= 0) {
binCount = 1;
for (Node<K,V> e = f;; ++binCount) {
K ek;
// 找到hash值相同的key,覆蓋舊值即可
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
oldVal = e.val;
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;
}
}
}
這部分代碼很好理解，就是兩種情況：1. 在鏈表中如果找到了與待插入的鍵值對(duì)的key相同的節(jié)點(diǎn)，就直接覆蓋即可；2. 如果直到找到了鏈表的末尾都沒(méi)有找到的話，就直接將待插入的鍵值對(duì)追加到鏈表的末尾即可
6.當(dāng)table[i]為紅黑樹(shù)的根節(jié)點(diǎn)，在紅黑樹(shù)中插入新值
按照之前的數(shù)組+鏈表的設(shè)計(jì)方案，這里存在一個(gè)問(wèn)題，即使負(fù)載因子和Hash算法設(shè)計(jì)的再合理，也免不了會(huì)出現(xiàn)拉鏈過(guò)長(zhǎng)的情況，一旦出現(xiàn)拉鏈過(guò)長(zhǎng)，甚至在極端情況下，查找一個(gè)節(jié)點(diǎn)會(huì)出現(xiàn)時(shí)間復(fù)雜度為O(n)的情況，則會(huì)嚴(yán)重影響ConcurrentHashMap的性能，于是，在JDK1.8版本中，對(duì)數(shù)據(jù)結(jié)構(gòu)做了進(jìn)一步的優(yōu)化，引入了紅黑樹(shù)。而當(dāng)鏈表長(zhǎng)度太長(zhǎng)（默認(rèn)超過(guò)8）時(shí)，鏈表就轉(zhuǎn)換為紅黑樹(shù)，利用紅黑樹(shù)快速增刪改查的特點(diǎn)提高ConcurrentHashMap的性能，其中會(huì)用到紅黑樹(shù)的插入、刪除、查找等算法。當(dāng)table[i]為紅黑樹(shù)的樹(shù)節(jié)點(diǎn)時(shí)的操作為：
if (f instanceof TreeBin) {
Node<K,V> p;
binCount = 2;
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
首先在if中通過(guò)f instanceof TreeBin判斷當(dāng)前table[i]是否是樹(shù)節(jié)點(diǎn)，這下也正好驗(yàn)證了我們?cè)谧钌厦娼榻B時(shí)說(shuō)的TreeBin會(huì)對(duì)TreeNode做進(jìn)一步封裝，對(duì)紅黑樹(shù)進(jìn)行操作的時(shí)候針對(duì)的是TreeBin而不是TreeNode。這段代碼很簡(jiǎn)單，調(diào)用putTreeVal方法完成向紅黑樹(shù)插入新節(jié)點(diǎn)，同樣的邏輯，如果在紅黑樹(shù)中存在于待插入鍵值對(duì)的Key相同（hash值相等并且equals方法判斷為true）的節(jié)點(diǎn)的話，就覆蓋舊值，否則就向紅黑樹(shù)追加新節(jié)點(diǎn)。
7.根據(jù)當(dāng)前節(jié)點(diǎn)個(gè)數(shù)進(jìn)行調(diào)整
當(dāng)完成數(shù)據(jù)新節(jié)點(diǎn)插入之后，會(huì)進(jìn)一步對(duì)當(dāng)前鏈表大小進(jìn)行調(diào)整，這部分代碼為：
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
很容易理解，如果當(dāng)前鏈表節(jié)點(diǎn)個(gè)數(shù)大于等于8（TREEIFY_THRESHOLD）的時(shí)候，就會(huì)調(diào)用treeifyBin方法將tabel[i]（第i個(gè)散列桶）拉鏈轉(zhuǎn)換成紅黑樹(shù)。
至此，關(guān)于Put方法的邏輯就基本說(shuō)的差不多了，現(xiàn)在來(lái)做一些總結(jié)：
整體流程：
- 首先對(duì)于每一個(gè)放入的值，首先利用spread方法對(duì)key的hashcode進(jìn)行一次hash計(jì)算，由此來(lái)確定這個(gè)值在 table中的位置；
- 如果當(dāng)前table數(shù)組還未初始化，先將table數(shù)組進(jìn)行初始化操作；
- 如果這個(gè)位置是null的，那么使用CAS操作直接放入；
- 如果這個(gè)位置存在結(jié)點(diǎn)，說(shuō)明發(fā)生了hash碰撞，首先判斷這個(gè)節(jié)點(diǎn)的類(lèi)型。如果該節(jié)點(diǎn)fh==MOVED(代表forwardingNode,數(shù)組正在進(jìn)行擴(kuò)容)的話，說(shuō)明正在進(jìn)行擴(kuò)容；
- 如果是鏈表節(jié)點(diǎn)（fh>0）,則得到的結(jié)點(diǎn)就是hash值相同的節(jié)點(diǎn)組成的鏈表的頭節(jié)點(diǎn)。需要依次向后遍歷確定這個(gè)新加入的值所在位置。如果遇到key相同的節(jié)點(diǎn)，則只需要覆蓋該結(jié)點(diǎn)的value值即可。否則依次向后遍歷，直到鏈表尾插入這個(gè)結(jié)點(diǎn)；
- 如果這個(gè)節(jié)點(diǎn)的類(lèi)型是TreeBin的話，直接調(diào)用紅黑樹(shù)的插入方法進(jìn)行插入新的節(jié)點(diǎn)；
- 插入完節(jié)點(diǎn)之后再次檢查鏈表長(zhǎng)度，如果長(zhǎng)度大于8，就把這個(gè)鏈表轉(zhuǎn)換成紅黑樹(shù)；
- 對(duì)當(dāng)前容量大小進(jìn)行檢查，如果超過(guò)了臨界值（實(shí)際大小*加載因子）就需要擴(kuò)容。
3.4 get方法
看完了put方法再來(lái)看get方法就很容易了，用逆向思維去看就好，這樣存的話我反過(guò)來(lái)這么取就好了。get方法源碼為：
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
// 1. 重hash
int h = spread(key.hashCode());
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
// 2. table[i]桶節(jié)點(diǎn)的key與查找的key相同，則直接返回
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
// 3. 當(dāng)前節(jié)點(diǎn)hash小于0說(shuō)明為樹(shù)節(jié)點(diǎn)，在紅黑樹(shù)中查找即可
else if (eh < 0)
return (p = e.find(h, key)) != null ? p.val : null;
while ((e = e.next) != null) {
//4. 從鏈表中查找，查找到則返回該節(jié)點(diǎn)的value，否則就返回null即可
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
}
代碼的邏輯請(qǐng)看注釋，首先先看當(dāng)前的hash桶數(shù)組節(jié)點(diǎn)即table[i]是否為查找的節(jié)點(diǎn)，若是則直接返回；若不是，則繼續(xù)再看當(dāng)前是不是樹(shù)節(jié)點(diǎn)？通過(guò)看節(jié)點(diǎn)的hash值是否為小于0，如果小于0則為樹(shù)節(jié)點(diǎn)。如果是樹(shù)節(jié)點(diǎn)在紅黑樹(shù)中查找節(jié)點(diǎn)；如果不是樹(shù)節(jié)點(diǎn)，那就只剩下為鏈表的形式的一種可能性了，就向后遍歷查找節(jié)點(diǎn)，若查找到則返回節(jié)點(diǎn)的value即可，若沒(méi)有找到就返回null。
3.5 transfer方法
當(dāng)ConcurrentHashMap容量不足的時(shí)候，需要對(duì)table進(jìn)行擴(kuò)容。這個(gè)方法的基本思想跟HashMap是很像的，但是由于它是支持并發(fā)擴(kuò)容的，所以要復(fù)雜的多。原因是它支持多線程進(jìn)行擴(kuò)容操作，而并沒(méi)有加鎖。我想這樣做的目的不僅僅是為了滿足concurrent的要求，而是希望利用并發(fā)處理去減少擴(kuò)容帶來(lái)的時(shí)間影響。transfer方法源碼為：
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
//1. 新建Node數(shù)組，容量為之前的兩倍
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 = n;
}
int nextn = nextTab.length;
//2. 新建forwardingNode引用，在之后會(huì)用到
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
boolean advance = true;
boolean finishing = false; // to ensure sweep before committing nextTab
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
// 3. 確定遍歷中的索引i
while (advance) {
int nextIndex, nextBound;
if (--i >= bound || finishing)
advance = false;
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
bound = nextBound;
i = nextIndex - 1;
advance = false;
}
}
//4.將原數(shù)組中的元素復(fù)制到新數(shù)組中去
//4.5 for循環(huán)退出，擴(kuò)容結(jié)束修改sizeCtl屬性
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
if (finishing) {
nextTable = null;
table = nextTab;
sizeCtl = (n << 1) - (n >>> 1);
return;
}
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
return;
finishing = advance = true;
i = n; // recheck before commit
}
}
//4.1 當(dāng)前數(shù)組中第i個(gè)元素為null，用CAS設(shè)置成特殊節(jié)點(diǎn)forwardingNode(可以理解成占位符)
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
//4.2 如果遍歷到ForwardingNode節(jié)點(diǎn) 說(shuō)明這個(gè)點(diǎn)已經(jīng)被處理過(guò)了 直接跳過(guò) 這里是控制并發(fā)擴(kuò)容的核心
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
else {
synchronized (f) {
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;
if (fh >= 0) {
//4.3 處理當(dāng)前節(jié)點(diǎn)為鏈表的頭結(jié)點(diǎn)的情況，構(gòu)造兩個(gè)鏈表，一個(gè)是原鏈表 另一個(gè)是原鏈表的反序排列
int runBit = fh & n;
Node<K,V> lastRun = f;
for (Node<K,V> p = f.next; p != null; p = p.next) {
int b = p.hash & n;
if (b != runBit) {
runBit = b;
lastRun = p;
}
}
if (runBit == 0) {
ln = lastRun;
hn = null;
}
else {
hn = lastRun;
ln = null;
}
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);
}
//在nextTable的i位置上插入一個(gè)鏈表
setTabAt(nextTab, i, ln);
//在nextTable的i+n的位置上插入另一個(gè)鏈表
setTabAt(nextTab, i + n, hn);
//在table的i位置上插入forwardNode節(jié)點(diǎn) 表示已經(jīng)處理過(guò)該節(jié)點(diǎn)
setTabAt(tab, i, fwd);
//設(shè)置advance為true 返回到上面的while循環(huán)中 就可以執(zhí)行i--操作
advance = true;
}
//4.4 處理當(dāng)前節(jié)點(diǎn)是TreeBin時(shí)的情況，操作和上面的類(lèi)似
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;
}
}
}
}
}
}
代碼邏輯請(qǐng)看注釋,整個(gè)擴(kuò)容操作分為兩個(gè)部分：
第一部分是構(gòu)建一個(gè)nextTable,它的容量是原來(lái)的兩倍，這個(gè)操作是單線程完成的。新建table數(shù)組的代碼為:Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1],在原容量大小的基礎(chǔ)上右移一位。
第二個(gè)部分就是將原來(lái)table中的元素復(fù)制到nextTable中，主要是遍歷復(fù)制的過(guò)程。
根據(jù)運(yùn)算得到當(dāng)前遍歷的數(shù)組的位置i，然后利用tabAt方法獲得i位置的元素再進(jìn)行判斷：
- 如果這個(gè)位置為空，就在原table中的i位置放入forwardNode節(jié)點(diǎn)，這個(gè)也是觸發(fā)并發(fā)擴(kuò)容的關(guān)鍵點(diǎn)；
- 如果這個(gè)位置是Node節(jié)點(diǎn)（fh>=0），如果它是一個(gè)鏈表的頭節(jié)點(diǎn)，就構(gòu)造一個(gè)反序鏈表，把他們分別放在nextTable的i和i+n的位置上
- 如果這個(gè)位置是TreeBin節(jié)點(diǎn)（fh<0），也做一個(gè)反序處理，并且判斷是否需要untreefi，把處理的結(jié)果分別放在nextTable的i和i+n的位置上
- 遍歷過(guò)所有的節(jié)點(diǎn)以后就完成了復(fù)制工作，這時(shí)讓nextTable作為新的table，并且更新sizeCtl為新容量的0.75倍 ，完成擴(kuò)容。設(shè)置為新容量的0.75倍代碼為
sizeCtl = (n << 1) - (n >>> 1)，仔細(xì)體會(huì)下是不是很巧妙，n<<1相當(dāng)于n右移一位表示n的兩倍即2n,n>>>1左右一位相當(dāng)于n除以2即0.5n,然后兩者相減為2n-0.5n=1.5n,是不是剛好等于新容量的0.75倍即2n*0.75=1.5n。最后用一個(gè)示意圖來(lái)進(jìn)行總結(jié)（圖片摘自網(wǎng)絡(luò)）：
3.6 與size相關(guān)的一些方法
對(duì)于ConcurrentHashMap來(lái)說(shuō)，這個(gè)table里到底裝了多少東西其實(shí)是個(gè)不確定的數(shù)量，因?yàn)?strong>不可能在調(diào)用size()方法的時(shí)候像GC的“stop the world”一樣讓其他線程都停下來(lái)讓你去統(tǒng)計(jì)，因此只能說(shuō)這個(gè)數(shù)量是個(gè)估計(jì)值。對(duì)于這個(gè)估計(jì)值，ConcurrentHashMap也是大費(fèi)周章才計(jì)算出來(lái)的。
為了統(tǒng)計(jì)元素個(gè)數(shù)，ConcurrentHashMap定義了一些變量和一個(gè)內(nèi)部類(lèi)
/**
* A padded cell for distributing counts. Adapted from LongAdder
* and Striped64. See their internal docs for explanation.
*/
@sun.misc.Contended static final class CounterCell {
volatile long value;
CounterCell(long x) { value = x; }
}
/******************************************/
/**
* 實(shí)際上保存的是hashmap中的元素個(gè)數(shù) 利用CAS鎖進(jìn)行更新
但它并不用返回當(dāng)前hashmap的元素個(gè)數(shù)
*/
private transient volatile long baseCount;
/**
* Spinlock (locked via CAS) used when resizing and/or creating CounterCells.
*/
private transient volatile int cellsBusy;
/**
* Table of counter cells. When non-null, size is a power of 2.
*/
private transient volatile CounterCell[] counterCells;
mappingCount與size方法
mappingCount與size方法的類(lèi)似 從給出的注釋來(lái)看，應(yīng)該使用mappingCount代替size方法 兩個(gè)方法都沒(méi)有直接返回basecount 而是統(tǒng)計(jì)一次這個(gè)值，而這個(gè)值其實(shí)也是一個(gè)大概的數(shù)值，因此可能在統(tǒng)計(jì)的時(shí)候有其他線程正在執(zhí)行插入或刪除操作。
public int size() {
long n = sumCount();
return ((n < 0L) ? 0 :
(n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
(int)n);
}
/**
* Returns the number of mappings. This method should be used
* instead of {@link #size} because a ConcurrentHashMap may
* contain more mappings than can be represented as an int. The
* value returned is an estimate; the actual count may differ if
* there are concurrent insertions or removals.
*
* @return the number of mappings
* @since 1.8
*/
public long mappingCount() {
long n = sumCount();
return (n < 0L) ? 0L : n; // ignore transient negative values
}
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;//所有counter的值求和
}
}
return sum;
}
addCount方法
在put方法結(jié)尾處調(diào)用了addCount方法，把當(dāng)前ConcurrentHashMap的元素個(gè)數(shù)+1這個(gè)方法一共做了兩件事,更新baseCount的值，檢測(cè)是否進(jìn)行擴(kuò)容。
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)) {
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))) {
fullAddCount(x, uncontended);
return;
}
if (check <= 1)
return;
s = sumCount();
}
//如果check值大于等于0 則需要檢驗(yàn)是否需要進(jìn)行擴(kuò)容操作
if (check >= 0) {
Node<K,V>[] tab, nt; int n, sc;
while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
(n = tab.length) < MAXIMUM_CAPACITY) {
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;
//如果已經(jīng)有其他線程在執(zhí)行擴(kuò)容操作
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
transfer(tab, nt);
}
//當(dāng)前線程是唯一的或是第一個(gè)發(fā)起擴(kuò)容的線程 此時(shí)nextTable=null
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
transfer(tab, null);
s = sumCount();
}
}
}
4. 總結(jié)
JDK6,7中的ConcurrentHashmap主要使用Segment來(lái)實(shí)現(xiàn)減小鎖粒度，分割成若干個(gè)Segment，在put的時(shí)候需要鎖住Segment，get時(shí)候不加鎖，使用volatile來(lái)保證可見(jiàn)性，當(dāng)要統(tǒng)計(jì)全局時(shí)（比如size），首先會(huì)嘗試多次計(jì)算modcount來(lái)確定，這幾次嘗試中，是否有其他線程進(jìn)行了修改操作，如果沒(méi)有，則直接返回size。如果有，則需要依次鎖住所有的Segment來(lái)計(jì)算。
1.8之前put定位節(jié)點(diǎn)時(shí)要先定位到具體的segment，然后再在segment中定位到具體的桶。而在1.8的時(shí)候摒棄了segment臃腫的設(shè)計(jì)，直接針對(duì)的是Node[] tale數(shù)組中的每一個(gè)桶，進(jìn)一步減小了鎖粒度。并且防止拉鏈過(guò)長(zhǎng)導(dǎo)致性能下降，當(dāng)鏈表長(zhǎng)度大于8的時(shí)候采用紅黑樹(shù)的設(shè)計(jì)。
主要設(shè)計(jì)上的變化有以下幾點(diǎn):
- 不采用segment而采用node，鎖住node來(lái)實(shí)現(xiàn)減小鎖粒度。
- 設(shè)計(jì)了MOVED狀態(tài) 當(dāng)resize的中過(guò)程中 線程2還在put數(shù)據(jù)，線程2會(huì)幫助resize。
- 使用3個(gè)CAS操作來(lái)確保node的一些操作的原子性，這種方式代替了鎖。
- sizeCtl的不同值來(lái)代表不同含義，起到了控制的作用。
- 采用synchronized而不是ReentrantLock
更多關(guān)于1.7版本與1.8版本的ConcurrentHashMap的實(shí)現(xiàn)對(duì)比，可以參考這篇文章。
參考文章
1.8版本ConcurrentHashMap
1.8版本的HashMap
