ConcurrentHashMap是HashMap的線程安全版本,內部也是使用(數組 + 鏈表 + 紅黑樹)的結構來存儲元素。java
相比於一樣線程安全的HashTable來講,效率等各方面都有極大地提升。算法
這裏先簡單介紹一下各類鎖,以便下文講到相關概念時能有個印象。數組
synchronized
java中的關鍵字,內部實現爲監視器鎖,主要是經過對象監視器在對象頭中的字段來代表的。
synchronized從舊版本到如今已經作了不少優化了,在運行時會有三種存在方式:偏向鎖,輕量級鎖,重量級鎖
。
偏向鎖,是指一段同步代碼一直被一個線程訪問,那麼這個線程會自動獲取鎖,下降獲取鎖的代價。
輕量級鎖,是指當鎖是偏向鎖時,被另外一個線程所訪問,偏向鎖會升級爲輕量級鎖,這個線程會經過自旋的方式嘗試獲取鎖,不會阻塞,提升性能。
重量級鎖,是指當鎖是輕量級鎖時,當自旋的線程自旋了必定的次數後,尚未獲取到鎖,就會進入阻塞狀態,該鎖升級爲重量級鎖,重量級鎖會使其餘線程阻塞,性能下降。
CAS
CAS,Compare And Swap,它是一種樂觀鎖,認爲對於同一個數據的併發操做不必定會發生修改,在更新數據的時候,嘗試去更新數據,若是失敗就不斷嘗試。
volatile(非鎖)
java中的關鍵字,當多個線程訪問同一個變量時,一個線程修改了這個變量的值,其餘線程可以當即看獲得修改的值。
volatile只保證可見性,不保證原子性,好比 volatile修改的變量 i,針對i++操做,不保證每次結果都正確,由於i++操做是兩步操做,至關於 i = i +1,先讀取,再加1,這種狀況 volatile是沒法保證的。
自旋鎖
自旋鎖,是指嘗試獲取鎖的線程不會阻塞,而是循環的方式不斷嘗試,這樣的好處是減小線程的上下文切換帶來的開鎖,提升性能,缺點是循環會消耗CPU。
分段鎖
分段鎖,是一種鎖的設計思路,它細化了鎖的粒度,主要運用在ConcurrentHashMap中,實現高效的併發操做,當操做不須要更新整個數組時,就只鎖數組中的一項就能夠了。
ReentrantLock
可重入鎖,是指一個線程獲取鎖以後再嘗試獲取鎖時會自動獲取鎖,可重入鎖的優勢是避免死鎖,synchronized也是可重入鎖。
public ConcurrentHashMap() { } public ConcurrentHashMap(int initialCapacity) { if (initialCapacity < 0) throw new IllegalArgumentException(); int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY : tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1)); this.sizeCtl = cap; } public ConcurrentHashMap(Map<? extends K, ? extends V> m) { this.sizeCtl = DEFAULT_CAPACITY; putAll(m); } public ConcurrentHashMap(int initialCapacity, float loadFactor) { this(initialCapacity, loadFactor, 1); } public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) { if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0) throw new IllegalArgumentException(); if (initialCapacity < concurrencyLevel) // Use at least as many bins initialCapacity = concurrencyLevel; // as estimated threads long size = (long)(1.0 + (long)initialCapacity / loadFactor); int cap = (size >= (long)MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : tableSizeFor((int)size); this.sizeCtl = cap; }
構造方法與HashMap對比能夠發現,沒有了HashMap中的threshold和loadFactor,而是改用了sizeCtl來控制,並且只存儲了容量在裏面,官方給出的解釋以下:安全
public V put(K key, V value) { return putVal(key, value, false); } final V putVal(K key, V value, boolean onlyIfAbsent) { // key和value都不能爲null if (key == null || value == null) throw new NullPointerException(); // 計算hash值 int hash = spread(key.hashCode()); // 要插入的元素所在桶的元素個數 int binCount = 0; // 死循環,結合CAS使用(若是CAS失敗,則會從新取整個桶進行下面的流程) for (Node<K,V>[] tab = table;;) { Node<K,V> f; int n, i, fh; if (tab == null || (n = tab.length) == 0) // 若是桶未初始化或者桶個數爲0,則初始化桶 tab = initTable(); else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { // 若是要插入的元素所在的桶尚未元素,則把這個元素插入到這個桶中 if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null))) // 若是使用CAS插入元素時,發現已經有元素了,則進入下一次循環,從新操做 // 若是使用CAS插入元素成功,則break跳出循環,流程結束 break; // no lock when adding to empty bin } else if ((fh = f.hash) == MOVED) // 若是要插入的元素所在的桶的第一個元素的hash是MOVED,則當前線程幫忙一塊兒遷移元素 tab = helpTransfer(tab, f); else { // 若是這個桶不爲空且不在遷移元素,則鎖住這個桶(分段鎖) // 並查找要插入的元素是否在這個桶中 // 存在,則替換值(onlyIfAbsent=false) // 不存在,則插入到鏈表結尾或插入樹中 V oldVal = null; synchronized (f) { // 再次檢測第一個元素是否有變化,若是有變化則進入下一次循環,從頭來過 if (tabAt(tab, i) == f) { // 若是第一個元素的hash值大於等於0(說明不是在遷移,也不是樹) // 那就是桶中的元素使用的是鏈表方式存儲 if (fh >= 0) { // 桶中元素個數賦值爲1 binCount = 1; // 遍歷整個桶,每次結束binCount加1 for (Node<K,V> e = f;; ++binCount) { K ek; if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) { // 若是找到了這個元素,則賦值了新值(onlyIfAbsent=false) // 並退出循環 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; } } } else if (f instanceof TreeBin) { // 若是第一個元素是樹節點 Node<K,V> p; // 桶中元素個數賦值爲2 binCount = 2; // 調用紅黑樹的插入方法插入元素 // 若是成功插入則返回null // 不然返回尋找到的節點 if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key, value)) != null) { // 若是找到了這個元素,則賦值了新值(onlyIfAbsent=false) // 並退出循環 oldVal = p.val; if (!onlyIfAbsent) p.val = value; } } } } // 若是binCount不爲0,說明成功插入了元素或者尋找到了元素 if (binCount != 0) { // 若是鏈表元素個數達到了8,則嘗試樹化 // 由於上面把元素插入到樹中時,binCount只賦值了2,並無計算整個樹中元素的個數 // 因此不會重複樹化 if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); // 若是要插入的元素已經存在,則返回舊值 if (oldVal != null) return oldVal; // 退出外層大循環,流程結束 break; } } } // 成功插入元素,元素個數加1(是否要擴容在這個裏面) addCount(1L, binCount); // 成功插入元素返回null return null; }
總體流程跟HashMap比較相似,大體是如下幾步:併發
添加元素操做中使用的鎖主要有(自旋鎖 + CAS + synchronized + 分段鎖)。dom
爲何使用synchronized而不是ReentrantLock?ide
由於synchronized已經獲得了極大地優化,在特定狀況下並不比ReentrantLock差。源碼分析
第一次放元素時,初始化桶數組。性能
private final Node<K,V>[] initTable() { Node<K,V>[] tab; int sc; while ((tab = table) == null || tab.length == 0) { if ((sc = sizeCtl) < 0) // 若是sizeCtl<0說明正在初始化或者擴容,讓出CPU Thread.yield(); // lost initialization race; just spin else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { // 若是把sizeCtl原子更新爲-1成功,則當前線程進入初始化 // 若是原子更新失敗則說明有其它線程先一步進入初始化了,則進入下一次循環 // 若是下一次循環時還沒初始化完畢,則sizeCtl<0進入上面if的邏輯讓出CPU // 若是下一次循環更新完畢了,則table.length!=0,退出循環 try { // 再次檢查table是否爲空,防止ABA問題 if ((tab = table) == null || tab.length == 0) { // 若是sc爲0則使用默認值16 int n = (sc > 0) ? sc : DEFAULT_CAPACITY; // 新建數組 @SuppressWarnings("unchecked") Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n]; // 賦值給table桶數組 table = tab = nt; // 設置sc爲數組長度的0.75倍 // n - (n >>> 2) = n - n/4 = 0.75n // 可見這裏裝載因子和擴容門檻都是寫死了的 // 這也正是沒有threshold和loadFactor屬性的緣由 sc = n - (n >>> 2); } } finally { // 把sc賦值給sizeCtl,這時存儲的是擴容門檻 sizeCtl = sc; } break; } } return tab; }
每次添加元素後,元素數量加1,並判斷是否達到擴容門檻,達到了則進行擴容或協助擴容。優化
private final void addCount(long x, int check) { CounterCell[] as; long b, s; // 這裏使用的思想跟LongAdder類是如出一轍的(後面會講) // 把數組的大小存儲根據不一樣的線程存儲到不一樣的段上(也是分段鎖的思想) // 而且有一個baseCount,優先更新baseCount,若是失敗了再更新不一樣線程對應的段 // 這樣能夠保證儘可能小的減小衝突 // 先嚐試把數量加到baseCount上,若是失敗再加到分段的CounterCell上 if ((as = counterCells) != null || !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) { CounterCell a; long v; int m; boolean uncontended = true; // 若是as爲空 // 或者長度爲0 // 或者當前線程所在的段爲null // 或者在當前線程的段上加數量失敗 if (as == null || (m = as.length - 1) < 0 || (a = as[ThreadLocalRandom.getProbe() & m]) == null || !(uncontended = U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) { // 強制增長數量(不管如何數量是必定要加上的,並非簡單地自旋) // 不一樣線程對應不一樣的段都更新失敗了 // 說明已經發生衝突了,那麼就對counterCells進行擴容 // 以減小多個線程hash到同一個段的機率 fullAddCount(x, uncontended); return; } if (check <= 1) return; // 計算元素個數 s = sumCount(); } if (check >= 0) { Node<K,V>[] tab, nt; int n, sc; // 若是元素個數達到了擴容門檻,則進行擴容 // 注意,正常狀況下sizeCtl存儲的是擴容門檻,即容量的0.75倍 while (s >= (long)(sc = sizeCtl) && (tab = table) != null && (n = tab.length) < MAXIMUM_CAPACITY) { // rs是擴容時的一個郵戳標識 int rs = resizeStamp(n); if (sc < 0) { // sc<0說明正在擴容中 if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || sc == rs + MAX_RESIZERS || (nt = nextTable) == null || transferIndex <= 0) // 擴容已經完成了,退出循環 // 正常應該只會觸發nextTable==null這個條件,其它條件沒看出來什麼時候觸發 break; // 擴容未完成,則當前線程加入遷移元素中 // 並把擴容線程數加1 if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) transfer(tab, nt); } else if (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2)) // 這裏是觸發擴容的那個線程進入的地方 // sizeCtl的高16位存儲着rs這個擴容郵戳 // sizeCtl的低16位存儲着擴容線程數加1,即(1+nThreads) // 因此官方說的擴容時sizeCtl的值爲 -(1+nThreads)是錯誤的 // 進入遷移元素 transfer(tab, null); // 從新計算元素個數 s = sumCount(); } } }
線程添加元素時發現正在擴容且當前元素所在的桶元素已經遷移完成了,則協助遷移其它桶的元素。
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) { Node<K,V>[] nextTab; int sc; // 若是桶數組不爲空,而且當前桶第一個元素爲ForwardingNode類型,而且nextTab不爲空 // 說明當前桶已經遷移完畢了,纔去幫忙遷移其它桶的元素 // 擴容時會把舊桶的第一個元素置爲ForwardingNode,並讓其nextTab指向新桶數組 if (tab != null && (f instanceof ForwardingNode) && (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) { int rs = resizeStamp(tab.length); // sizeCtl<0,說明正在擴容 while (nextTab == nextTable && table == tab && (sc = sizeCtl) < 0) { if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || sc == rs + MAX_RESIZERS || transferIndex <= 0) break; // 擴容線程數加1 if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) { // 當前線程幫忙遷移元素 transfer(tab, nextTab); break; } } return nextTab; } return table; }
當前桶元素遷移完成了纔去協助遷移其它桶元素;
擴容時容量變爲兩倍,並把部分元素遷移到其它桶中。
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 if (nextTab == null) { // initiating // 若是nextTab爲空,說明還沒開始遷移 // 就新建一個新桶數組 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; // 新建一個ForwardingNode類型的節點,並把新桶數組存儲在裏面 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; // 整個while循環就是在算i的值,過程太複雜,不用太關心 // i的值會從n-1依次遞減,感興趣的能夠打下斷點就知道了 // 其中n是舊桶數組的大小,也就是說i從15開始一直減到1這樣去遷移元素 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; } } if (i < 0 || i >= n || i + n >= nextn) { // 若是一次遍歷完成了 // 也就是整個map全部桶中的元素都遷移完成了 int sc; if (finishing) { // 若是所有遷移完成了,則替換舊桶數組 // 並設置下一次擴容門檻爲新桶數組容量的0.75倍 nextTable = null; table = nextTab; sizeCtl = (n << 1) - (n >>> 1); return; } if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) { // 當前線程擴容完成,把擴容線程數-1 if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT) // 擴容完成兩邊確定相等 return; // 把finishing設置爲true // finishing爲true纔會走到上面的if條件 finishing = advance = true; // i從新賦值爲n // 這樣會再從新遍歷一次桶數組,看看是否是都遷移完成了 // 也就是第二次遍歷都會走到下面的(fh = f.hash) == MOVED這個條件 i = n; // recheck before commit } } else if ((f = tabAt(tab, i)) == null) // 若是桶中無數據,直接放入ForwardingNode標記該桶已遷移 advance = casTabAt(tab, i, null, fwd); else if ((fh = f.hash) == MOVED) // 若是桶中第一個元素的hash值爲MOVED // 說明它是ForwardingNode節點 // 也就是該桶已遷移 advance = true; // already processed else { // 鎖定該桶並遷移元素 synchronized (f) { // 再次判斷當前桶第一個元素是否有修改 // 也就是可能其它線程先一步遷移了元素 if (tabAt(tab, i) == f) { // 把一個鏈表分化成兩個鏈表 // 規則是桶中各元素的hash與桶大小n進行與操做 // 等於0的放到低位鏈表(low)中,不等於0的放到高位鏈表(high)中 // 其中低位鏈表遷移到新桶中的位置相對舊桶不變 // 高位鏈表遷移到新桶中位置正好是其在舊桶的位置加n // 這也正是爲何擴容時容量在變成兩倍的緣由 Node<K,V> ln, hn; if (fh >= 0) { // 第一個元素的hash值大於等於0 // 說明該桶中元素是以鏈表形式存儲的 // 這裏與HashMap遷移算法基本相似 // 惟一不一樣的是多了一步尋找lastRun // 這裏的lastRun是提取出鏈表後面不用處理再特殊處理的子鏈表 // 好比全部元素的hash值與桶大小n與操做後的值分別爲 0 0 4 4 0 0 0 // 則最後後面三個0對應的元素確定仍是在同一個桶中 // 這時lastRun對應的就是倒數第三個節點 // 至於爲啥要這樣處理,我也沒太搞明白 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; } // 遍歷鏈表,把hash&n爲0的放在低位鏈表中 // 不爲0的放在高位鏈表中 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); // 高位鏈表的位置是原位置加n setTabAt(nextTab, i + n, hn); // 標記當前桶已遷移 setTabAt(tab, i, fwd); // advance爲true,返回上面進行--i操做 advance = true; } else if (f instanceof TreeBin) { // 若是第一個元素是樹節點 // 也是同樣,分化成兩顆樹 // 也是根據hash&n爲0放在低位樹中 // 不爲0放在高位樹中 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; // 遍歷整顆樹,根據hash&n是否爲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; } } // 若是分化的樹中元素個數小於等於6,則退化成鏈表 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); // 高位樹的位置是原位置加n setTabAt(nextTab, i + n, hn); // 標記該桶已遷移 setTabAt(tab, i, fwd); // advance爲true,返回上面進行--i操做 advance = true; } } } } } }