ConcurrentHashMap源碼分析_JDK1.8版本

時間 2019-11-12

標籤 concurrenthashmap 源碼分析 jdk1.8 jdk 版本欄目 Java 简体版

原文原文鏈接

在jdk1.8中主要作了2方面的改進java

改進一：取消segments字段，直接採用transient volatile HashEntry<K,V>[] table保存數據，採用table數組元素做爲鎖，從而實現了對每一行數據進行加鎖，進一步減小併發衝突的機率。算法

改進二：將原先table數組＋單向鏈表的數據結構，變動爲table數組＋單向鏈表＋紅黑樹的結構。對於hash表來講，最核心的能力在於將key hash以後能均勻的分佈在數組中。若是hash以後散列的很均勻，那麼table數組中的每一個隊列長度主要爲0或者1。但實際狀況並不是老是如此理想，雖然ConcurrentHashMap類默認的加載因子爲0.75，可是在數據量過大或者運氣不佳的狀況下，仍是會存在一些隊列長度過長的狀況，若是仍是採用單向列表方式，那麼查詢某個節點的時間複雜度爲O(n)；所以，對於個數超過8(默認值)的列表，jdk1.8中採用了紅黑樹的結構，那麼查詢的時間複雜度能夠下降到O(logN)，能夠改進性能。編程

一、jdk1.8的ConcurrentHashMap再也不使用Segment代理Map操做這種設計，總體結構變爲HashMap這種結構，可是依舊保留分段鎖的思想。以前版本是每一個Segment都持有一把鎖，1.8版本改成鎖住剛好裝在一個hash桶自己位置上的節點，也就是hash桶的第一個節點 tabAt(table, i)，後面直接叫第一個節點。它多是Node鏈表的頭結點、保留節點ReservationNode、或者是TreeBin節點（TreeBin節點持有紅黑樹的根節點）。還有，1.8的節點變成了4種，這個後面細說，是個重要的知識。

二、能夠多線程併發來完成擴容這個耗時耗力的操做。在以前的版本中若是Segment正在進行擴容操做，其餘寫線程都會被阻塞，jdk1.8改成一個寫線程觸發了擴容操做，其餘寫線程進行寫入操做時，能夠幫助它來完成擴容這個耗時的操做。多線程併發擴容這部分後面細說。

三、由於多線程併發擴容的存在，致使的其餘操做的實現上會有比較大的改動，常見的get/put/remove/replace/clear，以及迭代操做，都要考慮併發擴容的影響。

四、使用新的計數方法。不使用Segment時，若是直接使用一個volatile類變量計數，由於每次讀寫volatile變量的開銷很大，高併發時效率不如以前版本的使用Segment時的計數方式。jdk1.8新增了一個用與高併發狀況的計數工具類java.util.concurrent.atomic.LongAdder，此類是基本思想和1.7及之前的ConcurrentHashMap同樣，使用了一層中間類，叫作Cell（相似Segment這個類）的計數單元，來實現分段計數，最後合併統計一次。由於不一樣的計數單元能夠承擔不一樣的線程的計數要求，減小了線程之間的競爭，在1.8的ConcurrentHashMap基本結果改變時，繼續保持和分段計數同樣的併發計數效率。

關於這個LongAdder，專門寫了一篇，能夠看下這裏。

五、同1.8版本的HashMap，當一個hash桶中的hash衝突節點太多時，把鏈表變爲紅黑樹，提升衝突時的查找效率。

六、一些小的改進，具體見後面的源碼上我寫的註釋。

七、函數式編程、Stream api相關的新功能，佔據了1.8的大概40%的代碼，這部分這裏就先不說了。

JDK1.6版本

ConcurrentHashMap結構

在JDK1.6中，ConcurrentHashMap將數據分紅一段一段存儲，給每一段數據配一把鎖，當一個線程得到鎖互斥訪問一個段數據時，其餘段的數據也可被其餘線程訪問；每一個Segment擁有一把可重入鎖，所以ConcurrentHashMap的分段鎖數目即爲Segment數組長度。ConcurrentHashMap結構：每個segment都是一個HashEntry<K,V>[] table， table中的每個元素本質上都是一個HashEntry的單向隊列（單向鏈表實現）。每個segment都是一個HashEntry<K,V>[] table， table中的每個元素本質上都是一個HashEntry的單向隊列。api

鎖分離實現

當一個線程訪問Node/鍵值對數據時，必須得到與它對應的segment鎖，其餘線程能夠訪問其餘Segment中的數據（鎖分離）；數組

ConcurrentHashMap聲明

public class ConcurrentHashMap<K,V> extends AbstractMap<K,V> implements ConcurrentMap<K,V>, Serializable安全

無鎖算法：CAS

樂觀鎖與悲觀鎖

悲觀鎖好比synchronized鎖，爲確保其餘線程不會干擾當前線程工做，所以掛起其餘須要鎖的線程，等待持有鎖的線程釋放；數據結構
樂觀鎖老是假設沒有衝突發生去作操做，若是檢測到衝突就失敗重試，知道成功爲止；多線程

CAS算法

CAS(Compare And Swap)：CAS算法包含三個參數CAS(V, E, N)，判斷預期值E和內存舊值是否相同(Compare)，若是相等用新值N覆蓋舊值V(Swap)，不然失敗；
當多個線程嘗試使用CAS同時更新同一個變量時，只有其中一個線程能更新變量的值，其餘線程失敗（失敗線程不會被阻塞，而是被告知「失敗」，能夠繼續嘗試）；
CAS在硬件層面能夠被編譯爲機器指令執行，所以性能高於基於鎖佔有方式實現線程安全；併發

ConcurrentHashMap結構圖示

與JDK1.6對比

JDK 1.8取消類segments字段，直接用table數組存儲鍵值對，JDK1.6中每一個bucket中鍵值對組織方式是單向鏈表，查找複雜度是O(n)，JDK1.8中當鏈表長度超過TREEIFY_THRESHOLD時，鏈表轉換爲紅黑樹，查詢複雜度能夠下降到O(log n)，改進性能；app

鎖分離

JDK1.8中，一個線程每次對一個桶（鏈表 or 紅黑樹）進行加鎖，其餘線程仍然能夠訪問其餘桶；

線程安全

ConcurrentHashMap底層數據結構與HashMap相同，仍然採用table數組+鏈表+紅黑樹結構；
一個線程進行put/remove操做時，對桶（鏈表 or 紅黑樹）加上synchronized獨佔鎖；
ConcurrentHashMap採用CAS算法保證線程安全；

ConcurrentHashMap基本數據結構

transient volatile Node<K,V>[] table：鍵值對桶數組
private transient volatile Node<K,V>[] nextTable： rehash擴容時用到的新鍵值對數組
private transient volatile long baseCount：<span id = "jump1"></span>記錄當前鍵值對總數，經過CAS更新，對全部線程可見
private transient volatile int sizeCtl
sizeCtl表示鍵值對總數閾值，經過CAS更新, 對全部線程可見
- 當sizeCtl < 0時，表示多個線程在等待擴容；
- 當sizeCtl = 0時，默認值；
- 當sizeCtl > 0時，表示擴容的閾值；
private transient volatile int cellBusy：自旋鎖；
private transient volatile CounterCell[] counterCells: counter cell表，長度總爲2的冪次；
static class Segment<K,V>：在JDK1.8中，Segment類僅僅在序列化和反序列化時發揮做用；

// 視圖 private transient KeySetView<K,V> keySet private transient ValuesView<K,V> values private transient EntrySetView<K,V> entrySet

描述鍵值對：Node<K, V>

static class Node<K,V> implements Map.Entry<K,V> { final int hash; final K key; // 鍵值對的value和next均爲volatile類型 volatile V val; volatile Node<K,V> next; ... }

ConcurrentHashMap重要方法分析

構造函數

ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel)

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

該構造器會根據輸入的initialCapacity肯定一個 >= initialCapacity的最小2的次冪；

concurrentLevel：在JDK1.8以前本質是ConcurrentHashMap分段鎖總數，表示同時更新ConcurrentHashMap且不產生鎖競爭的最大線程數；在JDK1.8中，僅在構造器中確保初始容量>=concurrentLevel，爲兼容舊版本而保留；

添加/更新鍵值對：putVal

putVal方法分析

final V putVal(K key, V value, boolean onlyIfAbsent) { if (key == null || value == null) throw new NullPointerException(); int hash = spread(key.hashCode()); int binCount = 0; // 不斷CAS探測，若是其餘線程正在修改tab，CAS嘗試失敗，直到成功爲止 for (Node<K,V>[] tab = table;;) { Node<K,V> f; int n, i, fh; // 空表，對tab進行初始化 if (tab == null || (n = tab.length) == 0) tab = initTable(); /** * CAS探測空桶 * 計算key所在bucket表中數組索引: i = (n - 1) & hash) */ else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { // CAS添加新鍵值對 if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null))) break; // no lock when adding to empty bin } // 檢測到tab[i]桶正在進行rehash, else if ((fh = f.hash) == MOVED) tab = helpTransfer(tab, f); else { V oldVal = null; // 對桶的首元素上鎖獨佔 synchronized (f) { if (tabAt(tab, i) == f) { // 桶中鍵值對組織形式是鏈表 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; } } } // 桶中鍵值對組織形式是紅黑樹 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; } } } } // 檢查桶中鍵值對總數 if (binCount != 0) { if (binCount >= TREEIFY_THRESHOLD) // 鏈表轉換爲紅黑樹 treeifyBin(tab, i); if (oldVal != null) return oldVal; break; } } } // 更新baseCount addCount(1L, binCount); return null; }

synchronized (f) {}操做經過對桶的首元素 = 鏈表表頭 Or 紅黑樹根節點加鎖，從而實現對整個桶進行加鎖，有鎖分離思想的體現；

獲取鍵值對：get

public V get(Object key) { Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek; int h = spread(key.hashCode()); if ((tab = table) != null && (n = tab.length) > 0 && (e = tabAt(tab, (n - 1) & h)) != null) { if ((eh = e.hash) == h) { if ((ek = e.key) == key || (ek != null && key.equals(ek))) return e.val; } else if (eh < 0) return (p = e.find(h, key)) != null ? p.val : null; while ((e = e.next) != null) { if (e.hash == h && ((ek = e.key) == key || (ek != null && key.equals(ek)))) return e.val; } } return null; }

get方法經過CAS保證鍵值對的原子性，當tab[i]被鎖住，CAS失敗並不斷重試，保證get不會出錯；

刪除鍵值對：remove

擴容機制

transfer

當baseCount超過sizeCtl，將table中全部bin內的鍵值對拷貝到nextTable；
待補充；

helpTransfer

待補充；

table原子操做方法

獲取tab[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); }

tabAt方法原子讀取table[i]；調用Unsafe對象的getObjectVolatile方法獲取tab[i]，因爲對volatile寫操做happen-before於volatile讀操做，所以其餘線程對table的修改均對get讀取可見；
((long)i << ASHIFT) + ABASE)計算i元素的地址

CAS算法更新鍵值對：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); }

casTabAt經過compareAndSwapObject方法比較tabp[i]和v是否相等，相等就用c更新tab[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); }

僅在synchronized同步塊中被調用，更新鍵值對；

CAS更新baseCount

addCountaddCount

private final void addCount(long x, int check) { CounterCell[] as; long b, s; // s = b + x，完成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))) { // 多線程CAS發生失敗時執行 fullAddCount(x, uncontended); return; } if (check <= 1) return; s = sumCount(); } if (check >= 0) { Node<K,V>[] tab, nt; int n, sc; // 當更新後的鍵值對總數baseCount >= 閾值sizeCtl時，進行rehash while (s >= (long)(sc = sizeCtl) && (tab = table) != null && (n = tab.length) < MAXIMUM_CAPACITY) { int rs = resizeStamp(n); // sc < 0 表示其餘線程已經在rehash if (sc < 0) { if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || sc == rs + MAX_RESIZERS || (nt = nextTable) == null || transferIndex <= 0) break; // 其餘線程的rehash操做已經完成，當前線程能夠進行rehash if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) transfer(tab, nt); } // sc >= 0 表示只有當前線程在進行rehash操做，調用輔助擴容方法transfer else if (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2)) transfer(tab, null); s = sumCount(); } } }