14.并发容器之ConcurrentHashMap(JDK 1.8版本)
?
1.ConcurrentHashmap簡介
public V put(K key, V value) {
? ?synchronized (mutex) {return m.put(key, value);}
}
實際上SynchronizedMap實現依然是采用synchronized獨占式鎖進行線程安全的并發控制的。同樣，這種方案的性能也是令人不太滿意的。針對這種境況，Doug Lea大師不遺余力的為我們創造了一些線程安全的并發容器，讓每一個java開發人員倍感幸福。相對于hashmap來說，ConcurrentHashMap就是線程安全的map，其中利用了鎖分段的思想提高了并發度。
ConcurrentHashMap在JDK1.6的版本網上資料很多，有興趣的可以去看看。 JDK 1.6版本關鍵要素：
segment繼承了ReentrantLock充當鎖的角色，為每一個segment提供了線程安全的保障；
segment維護了哈希散列表的若干個桶，每個桶由HashEntry構成的鏈表。
而到了JDK 1.8的ConcurrentHashMap就有了很大的變化，光是代碼量就足足增加了很多。1.8版本舍棄了segment，并且大量使用了synchronized，以及CAS無鎖操作以保證ConcurrentHashMap操作的線程安全性。至于為什么不用ReentrantLock而是Synchronzied呢？實際上，synchronzied做了很多的優化，包括偏向鎖，輕量級鎖，重量級鎖，可以依次向上升級鎖狀態，但不能降級（關于synchronized可以看這篇文章），因此，使用synchronized相較于ReentrantLock的性能會持平甚至在某些情況更優，具體的性能測試可以去網上查閱一些資料。另外，底層數據結構改變為采用數組+鏈表+紅黑樹的數據形式。
2.關鍵屬性及類
在了解ConcurrentHashMap的具體方法實現前，我們需要系統的來看一下幾個關鍵的地方。
ConcurrentHashMap的關鍵屬性
table volatile Node<K,V>[] table://裝載Node的數組，作為ConcurrentHashMap的數據容器，采用懶加載的方式，直到第一次插入數據的時候才會進行初始化操作，數組的大小總是為2的冪次方。
nextTable volatile Node<K,V>[] nextTable; //擴容時使用，平時為null，只有在擴容的時候才為非null
sizeCtl volatile int sizeCtl; 該屬性用來控制table數組的大小，根據是否初始化和是否正在擴容有幾種情況： 當值為負數時：如果為-1表示正在初始化，如果為-N則表示當前正有N-1個線程進行擴容操作； 當值為正數時：如果當前數組為null的話表示table在初始化過程中，sizeCtl表示為需要新建數組的長度； 若已經初始化了，表示當前數據容器（table數組）可用容量也可以理解成臨界值（插入節點數超過了該臨界值就需要擴容），具體 為數組的長度n 乘以 加載因子loadFactor； 當值為0時，即數組長度為默認初始值。
sun.misc.Unsafe U 在ConcurrentHashMap的實現中可以看到大量的U.compareAndSwapXXXX的方法去修改ConcurrentHashMap的一些屬性。這些方法實際上是利用了CAS算法保證了線程安全性，這是一種樂觀策略，假設每一次操作都不會產生沖突，當且僅當沖突發生的時候再去嘗試。而CAS操作依賴于現代處理器指令集，通過底層CMPXCHG指令實現。CAS(V,O,N)核心思想為：若當前變量實際值V與期望的舊值O相同，則表明該變量沒被其他線程進行修改，因此可以安全的將新值N賦值給變量；若當前變量實際值V與期望的舊值O不相同，則表明該變量已經被其他線程做了處理，此時將新值N賦給變量操作就是不安全的，在進行重試。而在大量的同步組件和并發容器的實現中使用CAS是通過sun.misc.Unsafe類實現的，該類提供了一些可以直接操控內存和線程的底層操作，可以理解為java中的“指針”。該成員變量的獲取是在靜態代碼塊中：
? ? ?try {
? ? ? ? ?U = sun.misc.Unsafe.getUnsafe();
.......
? ? } catch (Exception e) {
? ? ? ? ?throw new Error(e);
? ? }
}
ConcurrentHashMap中關鍵內部類
Node Node類實現了Map.Entry接口，主要存放key-value對，并且具有next域
? ? ? ? ?final int hash;
? ? ? ? ?final K key;
? ? ? ? ?volatile V val;
? ? ? ? ?volatile Node<K,V> next;
......
}
另外可以看出很多屬性都是用volatile進行修飾的，也就是為了保證內存可見性。
TreeNode 樹節點，繼承于承載數據的Node類。而紅黑樹的操作是針對TreeBin類的，從該類的注釋也可以看出，也就是TreeBin會將TreeNode進行再一次封裝
? * 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 這個類并不負責包裝用戶的key、value信息，而是包裝的很多TreeNode節點。實際的ConcurrentHashMap“數組”中，存放的是TreeBin對象，而不是TreeNode對象。
? ? ? ? ?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 在擴容時才會出現的特殊節點，其key,value,hash全部為null。并擁有nextTable指針引用新的table數組。
? ? ?final Node<K,V>[] nextTable;
? ? ?ForwardingNode(Node<K,V>[] tab) {
? ? ? ? ?super(MOVED, null, null, null);
? ? ? ? ?this.nextTable = tab;
? ? }
? ? .....
}
CAS關鍵操作
在上面我們提及到在ConcurrentHashMap中會大量使用CAS修改它的屬性和一些操作。因此，在理解ConcurrentHashMap的方法前我們需要了解下面幾個常用的利用CAS算法來保障線程安全的操作。
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); } 該方法用來獲取table數組中索引為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操作設置table數組中索引為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); }
該方法用來設置table數組中索引為i的元素
3.重點方法講解
在熟悉上面的這核心信息之后，我們接下來就來依次看看幾個常用的方法是怎樣實現的。
3.1 實例構造器方法
在使用ConcurrentHashMap第一件事自然而然就是new 出來一個ConcurrentHashMap對象，一共提供了如下幾個構造器方法：
// 1. 構造一個空的map，即table數組還未初始化，初始化放在第一次插入數據時，默認大小為16ConcurrentHashMap()
// 2. 給定map的大小
ConcurrentHashMap(int initialCapacity)
// 3. 給定一個map
ConcurrentHashMap(Map<? extends K, ? extends V> m)
// 4. 給定map的大小以及加載因子
ConcurrentHashMap(int initialCapacity, float loadFactor)
// 5. 給定map大小，加載因子以及并發度（預計同時操作數據的線程）
ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel)
ConcurrentHashMap一共給我們提供了5種構造器方法，具體使用請看注釋，我們來看看第2種構造器，傳入指定大小時的情況，該構造器源碼為：
public ConcurrentHashMap(int initialCapacity) {//1. 小于0直接拋異常
? ?if (initialCapacity < 0)
? ? ? ?throw new IllegalArgumentException();
//2. 判斷是否超過了允許的最大值，超過了話則取最大值，否則再對該值進一步處理
//(>>> 1 無符號的右移一位)
? ?int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
? ? ? ? ? ? ? MAXIMUM_CAPACITY :
? ? ? ? ? ? ? tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
//3. 賦值給sizeCtl
? ?this.sizeCtl = cap;
}
這段代碼的邏輯請看注釋，很容易理解，如果小于0就直接拋出異常，如果指定值大于了所允許的最大值的話就取最大值，否則，在對指定值做進一步處理。最后將cap賦值給sizeCtl,關于sizeCtl的說明請看上面的說明，當調用構造器方法之后，sizeCtl的大小應該就代表了ConcurrentHashMap的大小，即table數組長度。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; }通過注釋就很清楚了，該方法會將調用構造器方法時指定的大小轉換成一個2的冪次方數，也就是說ConcurrentHashMap的大小一定是2的冪次方，比如，當指定大小為18時，為了滿足2的冪次方特性，實際上concurrentHashMapd的大小為2的5次方（32）。另外，需要注意的是，調用構造器方法的時候并未構造出table數組（可以理解為ConcurrentHashMap的數據容器），只是算出table數組的長度，當第一次向ConcurrentHashMap插入數據的時候才真正的完成初始化創建table數組的工作。
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. 保證只有一個線程正在進行初始化操作Thread.yield(); // lost initialization race; just spinelse if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {try {if ((tab = table) == null || tab.length == 0) {// 2. 得出數組的大小int n = (sc > 0) ? sc : DEFAULT_CAPACITY;@SuppressWarnings("unchecked")// 3. 這里才真正的初始化數組Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];table = tab = nt;// 4. 計算數組中可用的大小：實際大小n*0.75（加載因子）sc = n - (n >>> 2);}} finally {sizeCtl = sc;}break;}}return tab; }代碼的邏輯請見注釋，有可能存在一個情況是多個線程同時走到這個方法中，為了保證能夠正確初始化，在第1步中會先通過if進行判斷，若當前已經有一個線程正在初始化即sizeCtl值變為-1，這個時候其他線程在If判斷為true從而調用Thread.yield()讓出CPU時間片。正在進行初始化的線程會調用U.compareAndSwapInt方法將sizeCtl改為-1即正在初始化的狀態。另外還需要注意的事情是，在第四步中會進一步計算數組中可用的大小即為數組實際大小n乘以加載因子0.75.可以看看這里乘以0.75是怎么算的，0.75為四分之三，這里n - (n >>> 2)是不是剛好是n-(1/4)n=(3/4)n，挺有意思的吧:)。如果選擇是無參的構造器的話，這里在new Node數組的時候會使用默認大小為DEFAULT_CAPACITY（16），然后乘以加載因子0.75為12，也就是說數組的可用大小為12。
3.3 put方法
使用ConcurrentHashMap最常拍2用的也應該是put和get方法了吧，我們先來看看put方法是怎樣實現的。調用put方法時實際具體實現是putVal方法，源碼如下：
/** Implementation for put and putIfAbsent */ final V putVal(K key, V value, boolean onlyIfAbsent) {if (key == null || value == null) throw new NullPointerException();//1. 計算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. 如果當前table還沒有初始化先調用initTable方法將tab進行初始化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. 判斷當前正在擴容else if ((fh = f.hash) == MOVED)tab = helpTransfer(tab, f);else {V oldVal = null;synchronized (f) {if (tabAt(tab, i) == f) {//5. 判斷當前為鏈表，在鏈表中插入新的鍵值對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.否則當前為紅黑樹，將新的鍵值對插入到紅黑樹中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.插入完鍵值對后再根據實際大小看是否需要轉換成紅黑樹if (binCount != 0) {if (binCount >= TREEIFY_THRESHOLD)treeifyBin(tab, i);if (oldVal != null)return oldVal;break;}}}//8.對當前容量大小進行檢查，如果超過了臨界值（實際大小*加載因子）就需要擴容 addCount(1L, binCount);return null; }put方法的代碼量有點長，我們按照上面的分解的步驟一步步來看。從整體而言，為了解決線程安全的問題，ConcurrentHashMap使用了synchronzied和CAS的方式。在之前了解過HashMap以及1.8版本之前的ConcurrenHashMap都應該知道ConcurrentHashMap結構圖，為了方面下面的講解這里先直接給出，如果對這有疑問的話，可以在網上隨便搜搜即可。
?
?
如圖（圖片摘自網絡），ConcurrentHashMap是一個哈希桶數組，如果不出現哈希沖突的時候，每個元素均勻的分布在哈希桶數組中。當出現哈希沖突的時候，是標準的鏈地址的解決方式，將hash值相同的節點構成鏈表的形式，稱為“拉鏈法”，另外，在1.8版本中為了防止拉鏈過長，當鏈表的長度大于8的時候會將鏈表轉換成紅黑樹。table數組中的每個元素實際上是單鏈表的頭結點或者紅黑樹的根節點。當插入鍵值對時首先應該定位到要插入的桶，即插入table數組的索引i處。那么，怎樣計算得出索引i呢？當然是根據key的hashCode值。
spread()重哈希，以減小Hash沖突
我們知道對于一個hash表來說，hash值分散的不夠均勻的話會大大增加哈希沖突的概率，從而影響到hash表的性能。因此通過spread方法進行了一次重hash從而大大減小哈希沖突的可能性。spread方法為：
static final int spread(int h) {return (h ^ (h >>> 16)) & HASH_BITS; }該方法主要是將key的hashCode的低16位和高16位進行異或運算，這樣不僅能夠使得hash值能夠分散能夠均勻減小hash沖突的概率，另外只用到了異或運算，在性能開銷上也能兼顧，做到平衡的trade-off。
2.初始化table
緊接著到第2步，會判斷當前table數組是否初始化了，沒有的話就調用initTable進行初始化，該方法在上面已經講過了。
3.能否直接將新值插入到table數組中
從上面的結構示意圖就可以看出存在這樣一種情況，如果插入值待插入的位置剛好所在的table數組為null的話就可以直接將值插入即可。那么怎樣根據hash確定在table中待插入的索引i呢？很顯然可以通過hash值與數組的長度取模操作，從而確定新值插入到數組的哪個位置。而之前我們提過ConcurrentHashMap的大小總是2的冪次方，(n - 1) & hash運算等價于對長度n取模，也就是hash%n，但是位運算比取模運算的效率要高很多，Doug lea大師在設計并發容器的時候也是將性能優化到了極致，令人欽佩。
確定好數組的索引i后，就可以使用tabAt()方法（該方法在上面已經說明了，有疑問可以回過頭去看看）獲取該位置上的元素，如果當前Node f為null的話，就可以直接用casTabAt方法將新值插入即可。
4.當前是否正在擴容
如果當前節點不為null，且該節點為特殊節點（forwardingNode）的話，就說明當前concurrentHashMap正在進行擴容操作，關于擴容操作，下面會作為一個具體的方法進行講解。那么怎樣確定當前的這個Node是不是特殊的節點了？是通過判斷該節點的hash值是不是等于-1（MOVED）,代碼為(fh = f.hash) == MOVED，對MOVED的解釋在源碼上也寫的很清楚了：
static final int MOVED = -1; // hash for forwarding nodes5.當table[i]為鏈表的頭結點，在鏈表中插入新值
在table[i]不為null并且不為forwardingNode時，并且當前Node f的hash值大于0（fh >= 0）的話說明當前節點f為當前桶的所有的節點組成的鏈表的頭結點。那么接下來，要想向ConcurrentHashMap插入新值的話就是向這個鏈表插入新值。通過synchronized (f)的方式進行加鎖以實現線程安全性。往鏈表中插入節點的部分代碼為：
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. 在鏈表中如果找到了與待插入的鍵值對的key相同的節點，就直接覆蓋即可；2. 如果直到找到了鏈表的末尾都沒有找到的話，就直接將待插入的鍵值對追加到鏈表的末尾即可
6.當table[i]為紅黑樹的根節點，在紅黑樹中插入新值
按照之前的數組+鏈表的設計方案，這里存在一個問題，即使負載因子和Hash算法設計的再合理，也免不了會出現拉鏈過長的情況，一旦出現拉鏈過長，甚至在極端情況下，查找一個節點會出現時間復雜度為O(n)的情況，則會嚴重影響ConcurrentHashMap的性能，于是，在JDK1.8版本中，對數據結構做了進一步的優化，引入了紅黑樹。而當鏈表長度太長（默認超過8）時，鏈表就轉換為紅黑樹，利用紅黑樹快速增刪改查的特點提高ConcurrentHashMap的性能，其中會用到紅黑樹的插入、刪除、查找等算法。當table[i]為紅黑樹的樹節點時的操作為：
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中通過f instanceof TreeBin判斷當前table[i]是否是樹節點，這下也正好驗證了我們在最上面介紹時說的TreeBin會對TreeNode做進一步封裝，對紅黑樹進行操作的時候針對的是TreeBin而不是TreeNode。這段代碼很簡單，調用putTreeVal方法完成向紅黑樹插入新節點，同樣的邏輯，如果在紅黑樹中存在于待插入鍵值對的Key相同（hash值相等并且equals方法判斷為true）的節點的話，就覆蓋舊值，否則就向紅黑樹追加新節點。
7.根據當前節點個數進行調整
當完成數據新節點插入之后，會進一步對當前鏈表大小進行調整，這部分代碼為：
if (binCount != 0) {if (binCount >= TREEIFY_THRESHOLD)treeifyBin(tab, i);if (oldVal != null)return oldVal;break; }很容易理解，如果當前鏈表節點個數大于等于8（TREEIFY_THRESHOLD）的時候，就會調用treeifyBin方法將tabel[i]（第i個散列桶）拉鏈轉換成紅黑樹。
至此，關于Put方法的邏輯就基本說的差不多了，現在來做一些總結：
整體流程：
首先對于每一個放入的值，首先利用spread方法對key的hashcode進行一次重hash計算，由此來確定這個值在 table中的位置；
如果當前table數組還未初始化，先將table數組進行初始化操作；
如果這個位置是null的，那么使用CAS操作直接放入；
如果這個位置存在結點，說明發生了hash碰撞，首先判斷這個節點的類型。如果該節點fh==MOVED(代表forwardingNode,數組正在進行擴容)的話，說明正在進行擴容；
如果是鏈表節點（fh>0）,則得到的結點就是hash值相同的節點組成的鏈表的頭節點。需要依次向后遍歷確定這個新加入的值所在位置。如果遇到key相同的節點，則只需要覆蓋該結點的value值即可。否則依次向后遍歷，直到鏈表尾插入這個結點；
如果這個節點的類型是TreeBin的話(根據這個來判斷是否為紅黑樹)，直接調用紅黑樹的插入方法進行插入新的節點；
插入完節點之后再次檢查鏈表長度，如果長度大于8，就把這個鏈表轉換成紅黑樹；
對當前容量大小進行檢查，如果超過了臨界值（實際大小*加載因子）就需要擴容。
3.4 get方法
看完了put方法再來看get方法就很容易了，用逆向思維去看就好，這樣存的話我反過來這么取就好了。get方法源碼為：
public V get(Object key) {Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;// 1. 重hashint h = spread(key.hashCode());if ((tab = table) != null && (n = tab.length) > 0 &&(e = tabAt(tab, (n - 1) & h)) != null) {// 2. table[i]桶節點的key與查找的key相同，則直接返回if ((eh = e.hash) == h) {if ((ek = e.key) == key || (ek != null && key.equals(ek)))return e.val;}// 3. 當前節點hash小于0說明為樹節點，在紅黑樹中查找即可else if (eh < 0)return (p = e.find(h, key)) != null ? p.val : null;while ((e = e.next) != null) {//4. 從鏈表中查找，查找到則返回該節點的value，否則就返回null即可if (e.hash == h &&((ek = e.key) == key || (ek != null && key.equals(ek))))return e.val;}}return null; }代碼的邏輯請看注釋，首先先看當前的hash桶數組節點即table[i]是否為查找的節點，若是則直接返回；若不是，則繼續再看當前是不是樹節點？通過看節點的hash值是否為小于0，如果小于0則為樹節點。如果是樹節點在紅黑樹中查找節點；如果不是樹節點，那就只剩下為鏈表的形式的一種可能性了，就向后遍歷查找節點，若查找到則返回節點的value即可，若沒有找到就返回null。
3.5 transfer方法
當ConcurrentHashMap容量不足的時候，需要對table進行擴容。這個方法的基本思想跟HashMap是很像的，但是由于它是支持并發擴容的，所以要復雜的多。原因是它支持多線程進行擴容操作，而并沒有加鎖。我想這樣做的目的不僅僅是為了滿足concurrent的要求，而是希望利用并發處理去減少擴容帶來的時間影響。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數組，容量為之前的兩倍if (nextTab == null) { // initiatingtry {@SuppressWarnings("unchecked")Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];nextTab = nt;} catch (Throwable ex) { // try to cope with OOMEsizeCtl = Integer.MAX_VALUE;return;}nextTable = nextTab;transferIndex = n;}int nextn = nextTab.length;//2. 新建forwardingNode引用，在之后會用到ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);boolean advance = true;boolean finishing = false; // to ensure sweep before committing nextTabfor (int i = 0, bound = 0;;) {Node<K,V> f; int fh;// 3. 確定遍歷中的索引iwhile (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.將原數組中的元素復制到新數組中去//4.5 for循環退出，擴容結束修改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 當前數組中第i個元素為null，用CAS設置成特殊節點forwardingNode(可以理解成占位符)else if ((f = tabAt(tab, i)) == null)advance = casTabAt(tab, i, null, fwd);//4.2 如果遍歷到ForwardingNode節點 說明這個點已經被處理過了 直接跳過 這里是控制并發擴容的核心else if ((fh = f.hash) == MOVED)advance = true; // already processedelse {synchronized (f) {if (tabAt(tab, i) == f) {Node<K,V> ln, hn;if (fh >= 0) {//4.3 處理當前節點為鏈表的頭結點的情況，構造兩個鏈表，一個是原鏈表 另一個是原鏈表的反序排列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);elsehn = new Node<K,V>(ph, pk, pv, hn);}//在nextTable的i位置上插入一個鏈表setTabAt(nextTab, i, ln);//在nextTable的i+n的位置上插入另一個鏈表setTabAt(nextTab, i + n, hn);//在table的i位置上插入forwardNode節點 表示已經處理過該節點setTabAt(tab, i, fwd);//設置advance為true 返回到上面的while循環中 就可以執行i--操作advance = true;}//4.4 處理當前節點是TreeBin時的情況，操作和上面的類似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;elseloTail.next = p;loTail = p;++lc;}else {if ((p.prev = hiTail) == null)hi = p;elsehiTail.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;}}}}} }代碼邏輯請看注釋,整個擴容操作分為兩個部分：
第一部分是構建一個nextTable,它的容量是原來的兩倍，這個操作是單線程完成的。新建table數組的代碼為:Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1],在原容量大小的基礎上左移一位。
第二個部分就是將原來table中的元素復制到nextTable中，主要是遍歷復制的過程。 根據運算得到當前遍歷的數組的位置i，然后利用tabAt方法獲得i位置的元素再進行判斷：
如果這個位置為空，就在原table中的i位置放入forwardNode節點，這個也是觸發并發擴容的關鍵點；
如果這個位置是Node節點（fh>=0），如果它是一個鏈表的頭節點，就構造一個反序鏈表，把他們分別放在nextTable的i和i+n的位置上
如果這個位置是TreeBin節點（fh<0），也做一個反序處理，并且判斷是否需要untreefi，把處理的結果分別放在nextTable的i和i+n的位置上
遍歷過所有的節點以后就完成了復制工作，這時讓nextTable作為新的table，并且更新sizeCtl為新容量的0.75倍 ，完成擴容。設置為新容量的0.75倍代碼為 sizeCtl = (n << 1) - (n >>> 1)，仔細體會下是不是很巧妙，n<<1相當于n右移一位表示n的兩倍即2n,n>>>1左右一位相當于n除以2即0.5n,然后兩者相減為2n-0.5n=1.5n,是不是剛好等于新容量的0.75倍即2n*0.75=1.5n。最后用一個示意圖來進行總結（圖片摘自網絡）：
?
3.6 與size相關的一些方法
對于ConcurrentHashMap來說，這個table里到底裝了多少東西其實是個不確定的數量，因為不可能在調用size()方法的時候像GC的“stop the world”一樣讓其他線程都停下來讓你去統計，因此只能說這個數量是個估計值。對于這個估計值，ConcurrentHashMap也是大費周章才計算出來的。
為了統計元素個數，ConcurrentHashMap定義了一些變量和一個內部類
/*** 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; } }/******************************************/ /*** 實際上保存的是hashmap中的元素個數 利用CAS鎖進行更新但它并不用返回當前hashmap的元素個數 */ 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方法的類似 從給出的注釋來看，應該使用mappingCount代替size方法 兩個方法都沒有直接返回basecount 而是統計一次這個值，而這個值其實也是一個大概的數值，因此可能在統計的時候有其他線程正在執行插入或刪除操作。
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方法結尾處調用了addCount方法，把當前ConcurrentHashMap的元素個數+1這個方法一共做了兩件事,更新baseCount的值，檢測是否進行擴容。
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 則需要檢驗是否需要進行擴容操作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;//如果已經有其他線程在執行擴容操作if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))transfer(tab, nt);}//當前線程是唯一的或是第一個發起擴容的線程 此時nextTable=nullelse if (U.compareAndSwapInt(this, SIZECTL, sc,(rs << RESIZE_STAMP_SHIFT) + 2))transfer(tab, null);s = sumCount();}} }?
4. 總結
JDK6,7中的ConcurrentHashmap主要使用Segment來實現減小鎖粒度，分割成若干個Segment，在put的時候需要鎖住Segment，get時候不加鎖，使用volatile來保證可見性，當要統計全局時（比如size），首先會嘗試多次計算modcount來確定，這幾次嘗試中，是否有其他線程進行了修改操作，如果沒有，則直接返回size。如果有，則需要依次鎖住所有的Segment來計算。
1.8之前put定位節點時要先定位到具體的segment，然后再在segment中定位到具體的桶。而在1.8的時候摒棄了segment臃腫的設計，直接針對的是Node[] table數組中的每一個桶，進一步減小了鎖粒度。并且防止拉鏈過長導致性能下降，當鏈表長度大于8的時候采用紅黑樹的設計。
主要設計上的變化有以下幾點:
不采用segment而采用node，鎖住node來實現減小鎖粒度。
設計了MOVED狀態 當resize的中過程中 線程2還在put數據，線程2會幫助resize。
使用3個CAS操作來確保node的一些操作的原子性，這種方式代替了鎖。
sizeCtl的不同值來代表不同含義，起到了控制的作用。
采用synchronized而不是ReentrantLock
更多關于1.7版本與1.8版本的ConcurrentHashMap的實現對比，可以參考這篇文章。
參考文章
1.8版本ConcurrentHashMap
http://www.importnew.com/22007.html
http://www.jianshu.com/p/c0642afe03e0
1.8版本的HashMap
轉載于:https://www.cnblogs.com/itxiaok/p/10356586.html
總結
以上是生活随笔為你收集整理的14.并发容器之ConcurrentHashMap(JDK 1.8版本)的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 做梦梦到自己出轨被发现了是什么意思
- 下一篇： laravel-admin 开发 boo