ConcurrentHashMap & HashMap最清晰的底層原理分析(基於JDK1.7跟1.8比較)
前言
Map 這樣的 Key Value 在軟體開發中是非常經典的結構,常用於在記憶體中存放資料。
本篇主要想討論 ConcurrentHashMap 這樣一個併發容器,在正式開始之前我覺得有必要談談 HashMap,沒有它就不會有後面的 ConcurrentHashMap。
HashMap
眾所周知 HashMap 底層是基於 陣列 + 連結串列 組成的,不過在 jdk1.7 和 1.8 中具體實現稍有不同。
Base 1.7
1.7 中的資料結構圖:
先來看看 1.7 原始碼實現:
這是 HashMap 中比較核心的幾個成員變數;看看分別是什麼意思?
- 初始化桶大小,因為底層是陣列,所以這是陣列預設的大小。
- 桶最大值。
- 預設的負載因子(0.75)
- table 真正存放資料的陣列。
- Map 存放數量的大小。
- 桶大小,可在初始化時顯式指定。
- 負載因子,可在初始化時顯式指定。
重點解釋下負載因子:
由於給定的 HashMap 的容量大小是固定的,比如預設初始化:
public HashMap() { this(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR); } public HashMap(int initialCapacity, float loadFactor) { if (initialCapacity < 0) throw new IllegalArgumentException("Illegal initial capacity: " + initialCapacity); if (initialCapacity > MAXIMUM_CAPACITY) initialCapacity = MAXIMUM_CAPACITY; if (loadFactor <= 0 || Float.isNaN(loadFactor)) throw new IllegalArgumentException("Illegal load factor: " + loadFactor); this.loadFactor = loadFactor; threshold = initialCapacity; init(); }
給定的預設容量為 16,負載因子為 0.75。Map 在使用過程中不斷的往裡面存放資料,當數量達到了 16 * 0.75 = 12 就需要將當前 16 的容量進行擴容,而擴容這個過程涉及到 rehash、複製資料等操作,所以非常消耗效能。
因此通常建議能提前預估 HashMap 的大小最好,儘量的減少擴容帶來的效能損耗。
根據程式碼可以看到其實真正存放資料的是
transient Entry<K,V>[] table = (Entry<K,V>[]) EMPTY_TABLE;
這個陣列,那麼它又是如何定義的呢?
Entry 是 HashMap 中的一個內部類,從他的成員變數很容易看出:
- key 就是寫入時的鍵。
- value 自然就是值。
- 開始的時候就提到 HashMap 是由陣列和連結串列組成,所以這個 next 就是用於實現連結串列結構。
- hash 存放的是當前 key 的 hashcode。
知曉了基本結構,那來看看其中重要的寫入、獲取函式:
put 方法
public V put(K key, V value) {
if (table == EMPTY_TABLE) {
inflateTable(threshold);
}
if (key == null)
return putForNullKey(value);
int hash = hash(key);
int i = indexFor(hash, table.length);
for (Entry<K,V> e = table[i]; e != null; e = e.next) {
Object k;
if (e.hash == hash && ((k = e.key) == key || key.equals(k))) {
V oldValue = e.value;
e.value = value;
e.recordAccess(this);
return oldValue;
}
}
modCount++;
addEntry(hash, key, value, i);
return null;
}
再結合圖看看 put 的流程:
- 判斷當前陣列是否需要初始化。
- 如果 key 為空,則 put 一個空值進去。
- 根據 key 計算出 hashcode。
- 根據計算出的 hashcode 定位出所在桶。
- 如果桶是一個連結串列則需要遍歷判斷裡面的 hashcode、key 是否和傳入 key 相等,如果相等則進行覆蓋,並返回原來的值。
- 如果桶是空的,說明當前位置沒有資料存入;新增一個 Entry 物件寫入當前位置。
void addEntry(int hash, K key, V value, int bucketIndex) {
if ((size >= threshold) && (null != table[bucketIndex])) {
resize(2 * table.length);
hash = (null != key) ? hash(key) : 0;
bucketIndex = indexFor(hash, table.length);
}
createEntry(hash, key, value, bucketIndex);
}
void createEntry(int hash, K key, V value, int bucketIndex) {
Entry<K,V> e = table[bucketIndex];
table[bucketIndex] = new Entry<>(hash, key, value, e);
size++;
}
當呼叫 addEntry 寫入 Entry 時需要判斷是否需要擴容。
如果需要就進行兩倍擴充,並將當前的 key 重新 hash 並定位。
而在 createEntry
中會將當前位置的桶傳入到新建的桶中,如果當前桶有值就會在位置形成連結串列。
get 方法
public V get(Object key) {
if (key == null)
return getForNullKey();
Entry<K,V> entry = getEntry(key);
return null == entry ? null : entry.getValue();
}
final Entry<K,V> getEntry(Object key) {
if (size == 0) {
return null;
}
int hash = (key == null) ? 0 : hash(key);
for (Entry<K,V> e = table[indexFor(hash, table.length)];
e != null;
e = e.next) {
Object k;
if (e.hash == hash &&
((k = e.key) == key || (key != null && key.equals(k))))
return e;
}
return null;
}
再結合圖看看 get的流程:
- 首先也是根據 key 計算出 hashcode,然後定位到具體的桶中。
- 判斷該位置是否為連結串列。
- 不是連結串列就根據
key、key 的 hashcode
是否相等來返回值。 - 為連結串列則需要遍歷直到 key 及 hashcode 相等時候就返回值。
- 啥都沒取到就直接返回 null 。
Base 1.8
不知道 1.7 的實現大家看出需要優化的點沒有?其實一個很明顯的地方就是:
當 Hash 衝突嚴重時,在桶上形成的連結串列會變的越來越長,這樣在查詢時的效率就會越來越低;時間複雜度為
O(N)
。
因此 1.8 中重點優化了這個查詢效率。
1.8 HashMap 結構圖:
先來看看幾個核心的成員變數:
static final int DEFAULT_INITIAL_CAPACITY = 1 << 4; // aka 16
/**
* The maximum capacity, used if a higher value is implicitly specified
* by either of the constructors with arguments.
* MUST be a power of two <= 1<<30.
*/
static final int MAXIMUM_CAPACITY = 1 << 30;
/**
* The load factor used when none specified in constructor.
*/
static final float DEFAULT_LOAD_FACTOR = 0.75f;
static final int TREEIFY_THRESHOLD = 8;
transient Node<K,V>[] table;
/**
* Holds cached entrySet(). Note that AbstractMap fields are used
* for keySet() and values().
*/
transient Set<Map.Entry<K,V>> entrySet;
/**
* The number of key-value mappings contained in this map.
*/
transient int size;
和 1.7 大體上都差不多,還是有幾個重要的區別:
TREEIFY_THRESHOLD 用於判斷是否需要將連結串列轉換為紅黑樹的閾值。
- HashEntry 修改為 Node。
- Node 的核心組成其實也是和 1.7 中的 HashEntry 一樣,存放的都是 key value hashcode next 等資料。
再來看看核心方法。
put 方法
看似要比 1.7 的複雜,我們一步步拆解:
- 判斷當前桶是否為空,空的就需要初始化(resize 中會判斷是否進行初始化)。
- 根據當前 key 的 hashcode 定位到具體的桶中並判斷是否為空,為空表明沒有 Hash 衝突就直接在當前位置建立一個新桶即可。
- 如果當前桶有值( Hash 衝突),那麼就要比較當前桶中的 key、key 的 hashcode 與寫入的 key 是否相等,相等就賦值給 e,在第 8 步的時候會統一進行賦值及返回。
- 如果當前桶為紅黑樹,那就要按照紅黑樹的方式寫入資料。
- 如果是個連結串列,就需要將當前的 key、value 封裝成一個新節點寫入到當前桶的後面(形成連結串列)。
- 接著判斷當前連結串列的大小是否大於預設的閾值,大於時就要轉換為紅黑樹。
- 如果在遍歷過程中找到 key 相同時直接退出遍歷。
- 如果 e != null 就相當於存在相同的 key,那就需要將值覆蓋。
- 最後判斷是否需要進行擴容。
get 方法
public V get(Object key) {
Node<K,V> e;
return (e = getNode(hash(key), key)) == null ? null : e.value;
}
final Node<K,V> getNode(int hash, Object key) {
Node<K,V>[] tab; Node<K,V> first, e; int n; K k;
if ((tab = table) != null && (n = tab.length) > 0 &&
(first = tab[(n - 1) & hash]) != null) {
if (first.hash == hash && // always check first node
((k = first.key) == key || (key != null && key.equals(k))))
return first;
if ((e = first.next) != null) {
if (first instanceof TreeNode)
return ((TreeNode<K,V>)first).getTreeNode(hash, key);
do {
if (e.hash == hash &&
((k = e.key) == key || (key != null && key.equals(k))))
return e;
} while ((e = e.next) != null);
}
}
return null;
}
get 方法看起來就要簡單許多了。
- 首先將 key hash 之後取得所定位的桶。
- 如果桶為空則直接返回 null 。
- 否則判斷桶的第一個位置(有可能是連結串列、紅黑樹)的 key 是否為查詢的 key,是就直接返回 value。
- 如果第一個不匹配,則判斷它的下一個是紅黑樹還是連結串列。
- 紅黑樹就按照樹的查詢方式返回值。
- 不然就按照連結串列的方式遍歷匹配返回值。
從這兩個核心方法(get/put)可以看出 1.8 中對大連結串列做了優化,修改為紅黑樹之後查詢效率直接提高到了 O(logn)。
但是 HashMap 原有的問題也都存在,比如在併發場景下使用時容易出現死迴圈。
final HashMap<String, String> map = new HashMap<String, String>();
for (int i = 0; i < 1000; i++) {
new Thread(new Runnable() {
@Override
public void run() {
map.put(UUID.randomUUID().toString(), "");
}
}).start();
}
但是為什麼呢?簡單分析下。
看過上文的還記得在 HashMap 擴容的時候會呼叫 resize() 方法,就是這裡的併發操作容易在一個桶上形成環形連結串列;這樣當獲取一個不存在的 key 時,計算出的 index 正好是環形連結串列的下標就會出現死迴圈。
如下圖:
遍歷方式
還有一個值得注意的是 HashMap 的遍歷方式,通常有以下幾種:
Iterator<Map.Entry<String, Integer>> entryIterator = map.entrySet().iterator();
while (entryIterator.hasNext()) {
Map.Entry<String, Integer> next = entryIterator.next();
System.out.println("key=" + next.getKey() + " value=" + next.getValue());
}
Iterator<String> iterator = map.keySet().iterator();
while (iterator.hasNext()){
String key = iterator.next();
System.out.println("key=" + key + " value=" + map.get(key));
}
強烈建議使用第一種 EntrySet 進行遍歷。
第一種可以把 key value 同時取出,第二種還得需要通過 key 取一次 value,效率較低。
簡單總結下 HashMap:無論是 1.7 還是 1.8 其實都能看出 JDK 沒有對它做任何的同步操作,所以併發會出問題,甚至出現死迴圈導致系統不可用。
因此 JDK 推出了專項專用的 ConcurrentHashMap ,該類位於 java.util.concurrent 包下,專門用於解決併發問題。
堅持看到這裡的朋友算是已經把 ConcurrentHashMap 的基礎已經打牢了,下面正式開始分析。
ConcurrentHashMap
ConcurrentHashMap 同樣也分為 1.7 、1.8 版,兩者在實現上略有不同。
Base 1.7
先來看看 1.7 的實現,下面是他的結構圖:
如圖所示,是由 Segment 陣列、HashEntry 組成,和 HashMap 一樣,仍然是陣列加連結串列。
它的核心成員變數:
/**
* Segment 陣列,存放資料時首先需要定位到具體的 Segment 中。
*/
final Segment<K,V>[] segments;
transient Set<K> keySet;
transient Set<Map.Entry<K,V>> entrySet;
Segment 是 ConcurrentHashMap 的一個內部類,主要的組成如下:
static final class Segment<K,V> extends ReentrantLock implements Serializable {
private static final long serialVersionUID = 2249069246763182397L;
// 和 HashMap 中的 HashEntry 作用一樣,真正存放資料的桶
transient volatile HashEntry<K,V>[] table;
transient int count;
transient int modCount;
transient int threshold;
final float loadFactor;
}
看看其中 HashEntry 的組成:
和 HashMap 非常類似,唯一的區別就是其中的核心資料如 value ,以及連結串列都是 volatile 修飾的,保證了獲取時的可見性。
原理上來說:ConcurrentHashMap 採用了分段鎖技術,其中 Segment 繼承於 ReentrantLock。不會像 HashTable 那樣不管是 put 還是 get 操作都需要做同步處理,理論上 ConcurrentHashMap 支援 CurrencyLevel (Segment 陣列數量)的執行緒併發。每當一個執行緒佔用鎖訪問一個 Segment 時,不會影響到其他的 Segment。
下面也來看看核心的 put get 方法。
put 方法
public V put(K key, V value) {
Segment<K,V> s;
if (value == null)
throw new NullPointerException();
int hash = hash(key);
int j = (hash >>> segmentShift) & segmentMask;
if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck
(segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment
s = ensureSegment(j);
return s.put(key, hash, value, false);
}
首先是通過 key 定位到 Segment,之後在對應的 Segment 中進行具體的 put。
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
HashEntry<K,V> node = tryLock() ? null :
scanAndLockForPut(key, hash, value);
V oldValue;
try {
HashEntry<K,V>[] tab = table;
int index = (tab.length - 1) & hash;
HashEntry<K,V> first = entryAt(tab, index);
for (HashEntry<K,V> e = first;;) {
if (e != null) {
K k;
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) {
oldValue = e.value;
if (!onlyIfAbsent) {
e.value = value;
++modCount;
}
break;
}
e = e.next;
}
else {
if (node != null)
node.setNext(first);
else
node = new HashEntry<K,V>(hash, key, value, first);
int c = count + 1;
if (c > threshold && tab.length < MAXIMUM_CAPACITY)
rehash(node);
else
setEntryAt(tab, index, node);
++modCount;
count = c;
oldValue = null;
break;
}
}
} finally {
unlock();
}
return oldValue;
}
雖然 HashEntry 中的 value 是用 volatile 關鍵詞修飾的,但是並不能保證併發的原子性,所以 put 操作時仍然需要加鎖處理。
首先第一步的時候會嘗試獲取鎖,如果獲取失敗肯定就有其他執行緒存在競爭,則利用 scanAndLockForPut()
自旋獲取鎖。
- 嘗試自旋獲取鎖。
- 如果重試的次數達到了
MAX_SCAN_RETRIES
則改為阻塞鎖獲取,保證能獲取成功。
再結合圖看看 put 的流程:
- 將當前 Segment 中的 table 通過 key 的 hashcode 定位到 HashEntry。
- 遍歷該 HashEntry,如果不為空則判斷傳入的 key 和當前遍歷的 key 是否相等,相等則覆蓋舊的 value。
- 不為空則需要新建一個 HashEntry 並加入到 Segment 中,同時會先判斷是否需要擴容。
- 最後會解除在 1 中所獲取當前 Segment 的鎖。
get 方法
public V get(Object key) {
Segment<K,V> s; // manually integrate access methods to reduce overhead
HashEntry<K,V>[] tab;
int h = hash(key);
long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
(tab = s.table) != null) {
for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile
(tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);
e != null; e = e.next) {
K k;
if ((k = e.key) == key || (e.hash == h && key.equals(k)))
return e.value;
}
}
return null;
}
get 邏輯比較簡單:
只需要將 Key 通過 Hash 之後定位到具體的 Segment ,再通過一次 Hash 定位到具體的元素上。
由於 HashEntry 中的 value 屬性是用 volatile 關鍵詞修飾的,保證了記憶體可見性,所以每次獲取時都是最新值。
ConcurrentHashMap 的 get 方法是非常高效的,因為整個過程都不需要加鎖。
Base 1.8
1.7 已經解決了併發問題,並且能支援 N 個 Segment 這麼多次數的併發,但依然存在 HashMap 在 1.7 版本中的問題。
那就是查詢遍歷連結串列效率太低。
因此 1.8 做了一些資料結構上的調整。
首先來看下底層的組成結構:
看起來是不是和 1.8 HashMap 結構類似?
其中拋棄了原有的 Segment 分段鎖,而採用了 CAS + synchronized
來保證併發安全性。
也將 1.7 中存放資料的 HashEntry 改為 Node,但作用都是相同的。
其中的 val next
都用了 volatile 修飾,保證了可見性。
put 方法
重點來看看 put 函式:
再結合圖看看 put 的流程:
- 根據 key 計算出 hashcode 。
- 判斷是否需要進行初始化。
- f 即為當前 key 定位出的 Node,如果為空表示當前位置可以寫入資料,利用 CAS 嘗試寫入,失敗則自旋保證成功。
- 如果當前位置的 hashcode == MOVED == -1,則需要進行擴容。
- 如果都不滿足,則利用 synchronized 鎖寫入資料。
- 如果數量大於 TREEIFY_THRESHOLD 則要轉換為紅黑樹。
get 方法
再結合圖看看 get的流程:
- 根據計算出來的 hashcode 定址,如果就在桶上那麼直接返回值。
- 如果是紅黑樹那就按照樹的方式獲取值。
- 就不滿足那就按照連結串列的方式遍歷獲取值。
1.8 在 1.7 的資料結構上做了大的改動,採用紅黑樹之後可以保證查詢效率(
O(logn)
),甚至取消了 ReentrantLock 改為了 synchronized,這樣可以看出在新版的 JDK 中對 synchronized 優化是很到位的。
總結
看完了整個 HashMap 和 ConcurrentHashMap 在 1.7 和 1.8 中不同的實現方式相信大家對他們的理解應該會更加到位。
其實這塊也是面試的重點內容,通常的套路是:
談談你理解的 HashMap,講講其中的 get put 過程。
- 1.8 做了什麼優化?
- 是執行緒安全的嘛?
- 不安全會導致哪些問題?
- 如何解決?有沒有執行緒安全的併發容器?
- ConcurrentHashMap 是如何實現的? 1.7、1.8 實現有何不同?為什麼這麼做?
除了面試會問到之外平時的應用其實也蠻多,像之前談到的 Guava 中 Cache 的實現就是利用 ConcurrentHashMap 的思想。 --------------------- 作者:crossoverJie 來源:CSDN 原文:https://blog.csdn.net/qq_18661793/article/details/81161193