Hashtable哮洽，Collections.SynchronizedMap和ConcurrentHashMap線程安全實現(xiàn)原理的區(qū)別以及性能測試
這三種 Map 都是 Java 中比較重要的集合類，雖然前兩個不太常用钦听，但是因為與多線程相關(guān)亚隅，所以關(guān)于這幾種 Map 的對比已經(jīng)成為了 Java 面試時的高頻考點痘儡。首先要說明的是，其中每一個單獨拎出來都足夠支撐一篇長篇大論的技術(shù)文章枢步，所以本文把重點放在了這三種集合類的線程安全實現(xiàn)原理的對比以及性能測試上沉删，其他細(xì)節(jié)不做深入探討。
一醉途、線程安全原理對比
1. Hashtable
首先必須吐槽一下這個類名矾瑰，作為官方工具類竟然不符合駝峰命名規(guī)則，怪不得被棄用了隘擎，開玩笑哈哈殴穴，主要原因還是性能低下，那 Hashtable 的性能為什么低下呢货葬，這個嘛只需要看一下它的源碼就一目了然了采幌，以下是 Hashtable 中幾個比較重要的方法：
public synchronized V put(K key, V value) {...}
public synchronized V get(Object key) {...}
public synchronized int size() {...}
public synchronized boolean remove(Object key, Object value) {...}
public synchronized boolean contains(Object value) {...}
... ...
查看源碼后可以看出，Hashtable 實現(xiàn)線程安全的原理相當(dāng)簡單粗暴震桶，直接在方法聲明上使用 synchronized 關(guān)鍵字休傍。這樣一來，不管線程執(zhí)行哪個方法蹲姐，即便只是讀取數(shù)據(jù)磨取，都需要鎖住整個 Hashtable 對象，可想而知其并發(fā)性能必然不會太好柴墩。
2. Collections.SynchronizedMap
SynchronizedMap 是 Collections 集合類的私有靜態(tài)內(nèi)部類忙厌，其定義和構(gòu)造方法如下：
private static class SynchronizedMap<K,V> implements Map<K,V>, Serializable {
private static final long serialVersionUID = 1978198479659022715L;
// 用于接收傳入的Map對象，也是類方法操作的對象
private final Map<K,V> m;
// 鎖對象
final Object mutex;
// 以下是SynchronizedMap的兩個構(gòu)造方法
SynchronizedMap(Map<K,V> m) {
this.m = Objects.requireNonNull(m);
mutex = this;
}
SynchronizedMap(Map<K,V> m, Object mutex) {
this.m = m;
this.mutex = mutex;
}
}
SynchronizedMap 一共有三個成員變量江咳，序列化ID拋開不談逢净，另外兩個分別是 Map 類型的實例變量 m，用于接收構(gòu)造方法中傳入的 Map 參數(shù)歼指，以及 Object 類型的實例變量 mutex爹土，作為鎖對象使用。
再來看構(gòu)造方法东臀，SynchronizedMap 有兩個構(gòu)造方法着饥。第一個構(gòu)造方法需要傳入一個 Map 類型的參數(shù)，這個參數(shù)會被傳遞給成員變量 m惰赋，接下來 SynchronizedMap 所有方法的操作都是針對 m 的操作宰掉，需要注意的是這個參數(shù)不能為空，否則會由 Objects 類的 requireNonNull() 方法拋出空指針異常赁濒，然后當(dāng)前的 SynchronizedMap 對象 this 會被傳遞給 mutex 作為鎖對象轨奄；第二個構(gòu)造方法有兩個參數(shù)，第一個 Map 類型的參數(shù)會被傳遞給成員變量 m拒炎，第二個 Object 類型的參數(shù)會被傳遞給 mutex 作為鎖對象挪拟。
-
最后來看 SynchronizedMap 的主要方法 (只選取了一部分)：
public int size() { synchronized (mutex) {return m.size();} } public boolean isEmpty() { synchronized (mutex) {return m.isEmpty();} } public boolean containsKey(Object key) { synchronized (mutex) {return m.containsKey(key);} } public V get(Object key) { synchronized (mutex) {return m.get(key);} } public V put(K key, V value) { synchronized (mutex) {return m.put(key, value);} } public V remove(Object key) { synchronized (mutex) {return m.remove(key);} }
從源碼可以看出，SynchronizedMap 實現(xiàn)線程安全的方法也是比較簡單的击你，所有方法都是先對鎖對象 mutex 上鎖玉组，然后再直接調(diào)用 Map 類型成員變量 m 的相關(guān)方法谎柄。這樣一來，線程在執(zhí)行方法時惯雳，只有先獲得了 mutex 的鎖才能對 m 進行操作朝巫。因此，跟 Hashtable 一樣石景，在同一個時間點劈猿，只能有一個線程對 SynchronizedMap 對象進行操作，雖然保證了線程安全潮孽，卻導(dǎo)致了性能低下揪荣。這么看來，連 Hashtable 都被棄用了往史，那性能同樣低下的 SynchronizedMap 還有什么存在的必要呢仗颈？別忘了，后者的構(gòu)造方法需要傳入一個 Map 類型的參數(shù)怠堪，也就是說它可以將非線程安全的 Map 轉(zhuǎn)化為線程安全的 Map揽乱，而這正是其存在的意義，以下是 SynchronizedMap 的用法示例 (這里并沒有演示多線程操作)：
Map<String, Integer> map = new HashMap<>();
//非線程安全操作
map.put("one", 1);
Integer one = map.get("one");
Map<String, Integer> synchronizedMap = Collections.synchronizedMap(map);
//線程安全操作
one = synchronizedMap.get("one");
synchronizedMap.put("two", 2);
Integer two = synchronizedMap.get("two");
3. ConcurrentHashMap
接下來是數(shù)據(jù)結(jié)構(gòu)和線程安全原理都最復(fù)雜的 ConcurrentHashMap粟矿。首先必須要感嘆一下凰棉，這個類的結(jié)構(gòu)之復(fù)雜（包含53個內(nèi)部類），設(shè)計之精妙（不知道怎么形容陌粹，反正就是很精妙）撒犀，真是令人嘆為觀止。說實話掏秩，要想徹底理解 ConcurrentHashMap 的各個細(xì)節(jié)或舞，還是需要相當(dāng)扎實的基礎(chǔ)并花費大量精力的。本文對于 ConcurrentHashMap 線程安全的原理只是做了宏觀的介紹蒙幻，想要深入理解的同學(xué)映凳，可以去文末的傳送門。另外邮破，本文著重介紹 JDK 1.8 版本的 ConcurrentHashMap诈豌，不過會對 JDK 1.7 版本做個簡單的回顧。
3.1 JDK 1.7 ConcurrentHashMap鎖實現(xiàn)原理回顧
如果你有一定基礎(chǔ)的話抒和，應(yīng)該會知道分段鎖這個概念矫渔。沒錯，這是JDK 1.7版本的 ConcurrentHashMap 實現(xiàn)線程安全的主要手段摧莽，具體一點就是 Segment + HashEntry + ReentrantLock庙洼。簡單來說，ConcurrentHashMap 是一個 Segment 數(shù)組（默認(rèn)長度為16），每個 Segment 又包含了一個 HashEntry 數(shù)組油够，所以可以看做一個 HashMap蚁袭， Segment 通過繼承 ReentrantLock 來進行加鎖，所以每次需要加鎖的操作鎖住的是一個 Segment叠聋，這樣只要保證每個 Segment 是線程安全的撕阎，也就實現(xiàn)了全局的線程安全。
3.2 JDK 1.8 ConcurrentHashMap 線程安全原理詳解
JDK 1.8 版本摒棄了之前版本中較為臃腫的 Segment 分段鎖設(shè)計碌补，取而代之的是 Node 數(shù)組 + CAS + synchronized + volatile 的新設(shè)計。這樣一來棉饶，ConcurrentHashMap 不僅數(shù)據(jù)結(jié)構(gòu)變得更簡單了（與JDK 1.8 的HashMap類似）厦章，鎖的粒度也更小了，鎖的單位從 Segment 變成了 Node 數(shù)組中的桶（科普：桶就是指數(shù)組中某個下標(biāo)位置上的數(shù)據(jù)集合照藻，這里可能是鏈表袜啃，也可能是紅黑樹）。說到紅黑樹幸缕，必須提一下群发，在JDK 1.8 的 HashMap 和 ConcurrentHashMap 中，如果某個數(shù)組位置上的鏈表長度過長（大于等于8）发乔，就會轉(zhuǎn)化為紅黑樹以提高查詢效率熟妓，不過這不是本文的重點。以下是 ConcurrentHashMap 線程安全原理的詳細(xì)介紹：
3.2.1 get 操作過程
? 可以發(fā)現(xiàn)發(fā)現(xiàn)源碼中完全沒有加鎖的操作栏尚，后面會說明原因
- 首先計算hash值起愈，定位到該table索引位置，如果是首節(jié)點符合就返回
- 如果遇到擴容的時候译仗，會調(diào)用標(biāo)志正在擴容節(jié)點ForwardingNode的find方法抬虽，查找該節(jié)點，匹配就返回
- 以上都不符合的話纵菌，就往下遍歷節(jié)點阐污，匹配就返回，否則最后就返回null
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
int h = spread(key.hashCode()); //計算hash
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {//讀取頭節(jié)點的Node元素
if ((eh = e.hash) == h) { //如果該節(jié)點就是首節(jié)點就返回
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
//hash值為負(fù)值表示正在擴容咱圆，這個時候查的是ForwardingNode的find方法來定位到nextTable來
//eh代表頭節(jié)點的hash值笛辟，eh=-1，說明該節(jié)點是一個ForwardingNode闷堡，正在遷移隘膘，此時調(diào)用ForwardingNode的find方法去nextTable里找。
//eh=-2杠览，說明該節(jié)點是一個TreeBin弯菊，此時調(diào)用TreeBin的find方法遍歷紅黑樹，由于紅黑樹有可能正在旋轉(zhuǎn)變色，所以find里會有讀寫鎖管钳。
//eh>=0钦铁，說明該節(jié)點下掛的是一個鏈表，直接遍歷該鏈表即可才漆。
else if (eh < 0)
return (p = e.find(h, key)) != null ? p.val : null;
while ((e = e.next) != null) {//既不是首節(jié)點也不是ForwardingNode牛曹，那就往下遍歷
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
}
可能有同學(xué)會提出疑問：為什么 get 操作不需要加鎖呢？這個嘛醇滥，也需要看一下源碼：
/**
* The array of bins. Lazily initialized upon first insertion.
* Size is always a power of two. Accessed directly by iterators.
* 這是ConcurrentHashMap的成員變量黎比，用 volatile修飾的Node數(shù)組，保證了數(shù)組在擴容時對其他線程的可見性
* 另外需要注意的是鸳玩，這個數(shù)組是延遲初始化的阅虫，會在第一次put元素時進行初始化，后面還會用到這個知識點
*/
transient volatile Node<K,V>[] table;
/**
* 這是ConcurrentHashMap靜態(tài)內(nèi)部類Node的定義不跟，可見其成員變量val和next都使用volatile修飾颓帝，可保證
* 在多線程環(huán)境下線程A修改結(jié)點的val或者新增節(jié)點的時候是對線程B可見的
*/
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 關(guān)鍵字已經(jīng)足以保證線程在讀取數(shù)據(jù)時不會讀取到臟數(shù)據(jù)，所以沒有加鎖的必要窝革。
3.2.2 put 操作過程
- 第一次 put 元素會初始化 Node 數(shù)組 (initTable)
- put 操作又分為 key (hash 碰撞) 存在時的插入和 key 不存在時的插入
- put 操作可能會引發(fā)數(shù)組擴容 (tryPresize) 和鏈表轉(zhuǎn)紅黑樹 (treeifyBin)
- 擴容會使用到數(shù)據(jù)遷移方法 (transfer)
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
// 得到 hash 值
int hash = spread(key.hashCode());
// 用于記錄相應(yīng)鏈表的長度
int binCount = 0;
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
// 如果數(shù)組"空"购城，進行數(shù)組初始化
if (tab == null || (n = tab.length) == 0)
// 表的初始化，這里不展開了虐译，核心思想是使用sizeCtl的變量和CAS操作進行控制瘪板，保證數(shù)組在擴容時
// 不會創(chuàng)建出多余的表
tab = initTable();
// 找該 hash 值對應(yīng)的數(shù)組下標(biāo)，得到第一個節(jié)點 f
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
// 如果數(shù)組該位置為空菱蔬，用一次 CAS 操作將這個新值放入其中即可篷帅，這個 put 操作差不多就結(jié)束了
// 如果 CAS 失敗，那就是有并發(fā)操作拴泌，進到下一個循環(huán)就好了(循環(huán)的意思是 CAS 在執(zhí)行失敗后會進行 // 重試)
if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null)))
break; // no lock when adding to empty bin
}
else if ((fh = f.hash) == MOVED)
// 幫助數(shù)據(jù)遷移魏身，
tab = helpTransfer(tab, f);
else { // 到這里就是說，f 是該位置的頭結(jié)點蚪腐，而且不為空
V oldVal = null;
// 獲取數(shù)組該位置的頭結(jié)點鎖對象
synchronized (f) {
if (tabAt(tab, i) == f) {
if (fh >= 0) { // 頭結(jié)點的 hash 值大于 0箭昵，說明是鏈表
// 用于累加，記錄鏈表的長度
binCount = 1;
// 遍歷鏈表
for (Node<K,V> e = f;; ++binCount) {
K ek;
// 如果發(fā)現(xiàn)了"相等"的 key回季，判斷是否要進行值覆蓋家制，然后也就可以 break 了
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;
// 調(diào)用紅黑樹的插值方法插入新節(jié)點
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key, value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
if (binCount != 0) {
// 判斷是否要將鏈表轉(zhuǎn)換為紅黑樹泡一，臨界值和 HashMap 一樣颤殴，也是 8
if (binCount >= TREEIFY_THRESHOLD)
// 這個方法和 HashMap 中稍微有一點點不同，那就是它不是一定會進行紅黑樹轉(zhuǎn)換鼻忠，
// 如果當(dāng)前數(shù)組的長度小于 64涵但，那么會選擇進行數(shù)組擴容，而不是轉(zhuǎn)換為紅黑樹
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);
return null;
}
以上是 put 方法的源碼和分析，其中涉及到的其他方法矮瘟，比如 initTable瞳脓，helpTransfer，treeifyBin 和 tryPresize 等方法不再一一展開澈侠，有興趣的同學(xué)可以去文末傳送門看詳細(xì)解析劫侧。
3.2.3 CAS 操作簡要介紹
? CAS 操作是新版本 ConcurrentHashMap 線程安全實現(xiàn)原理的精華所在，如果說其共享變量的讀取全靠 volatile 實現(xiàn)線程安全的話哨啃，那么存儲和修改過程除了使用少量的 synchronized 關(guān)鍵字外烧栋，主要是靠 CAS 操作實現(xiàn)線程安全的。不了解 CAS 操作的同學(xué)看這里 JAVA CAS原理深度分析
// CAS操作的提供者
private static final sun.misc.Unsafe U;
// 以下是put方法里用到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;
}
// tabAt方法通過Unsafe.getObjectVolatile()的方式獲取數(shù)組對應(yīng)index上的元素棘催，getObjectVolatile作用于對
// 應(yīng)的內(nèi)存偏移量上劲弦，是具備volatile內(nèi)存語義的。
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);
}
// 如果獲取的是空醇坝，嘗試用CAS的方式在數(shù)組的指定index上創(chuàng)建一個新的Node。
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);
}
在 ConcurrentHashMap 中次坡，數(shù)組初始化呼猪、插入刪除元素、擴容砸琅、數(shù)據(jù)遷移以及鏈表和紅黑樹的轉(zhuǎn)換等過程都會涉及到線程安全問題宋距，而相關(guān)的方法中實現(xiàn)線程安全的思想是一致的：對桶中的數(shù)據(jù)進行添加或修改操作時會用到 synchronized 關(guān)鍵字，也就是獲得該位置上頭節(jié)點對象的鎖症脂，保證線程安全谚赎，另外就是用到了大量的 CAS 操作。以上就是對這三種 Map 的線程安全原理的簡要介紹诱篷。
二壶唤、性能測試
直接上代碼
public class MapPerformanceTest {
private static final int THREAD_POOL_SIZE = 5;
private static Map<String, Integer> hashtableObject = null;
private static Map<String, Integer> concurrentHashMapObject = null;
private static Map<String, Integer> synchronizedMap = null;
private static void performanceTest(final Map<String, Integer> map) throws InterruptedException {
System.out.println(map.getClass().getSimpleName() + "性能測試開始了... ...");
long totalTime = 0;
// 進行五次性能測試，每次開啟五個線程棕所，每個線程對 map 進行500000次查詢操作和500000次插入操作
for (int i = 0; i < 5; i++) {
long startTime = System.nanoTime();
ExecutorService service = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
for (int j = 0; j < THREAD_POOL_SIZE; j++) {
service.execute(() -> {
for (int k = 0; k < 500000; k++) {
Integer randomNumber = (int)Math.ceil(Math.random() * 500000);
// 從map中查找數(shù)據(jù),查找結(jié)果并不會用到,這里不能用int接收返回值闸盔，因為Integer可能是
// null，賦值給int會引發(fā)空指針異常
Integer value = map.get(String.valueOf(randomNumber));
//向map中添加元素
map.put(String.valueOf(randomNumber), randomNumber);
}
});
}
//關(guān)閉線程池
service.shutdown();
service.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
long endTime = System.nanoTime();
// 單次執(zhí)行時間
long singleTime = (endTime - startTime) / 1000000;
System.out.println("第" + (i + 1) + "次測試耗時" + singleTime + "ms...");
totalTime += singleTime;
}
System.out.println("平均耗時" + totalTime / 5 + "ms...");
}
public static void main(String[] args) throws InterruptedException {
//Hashtable性能測試
hashtableObject = new Hashtable<>();
performanceTest(hashtableObject);
//SynchronizedMap性能測試
Map<String, Integer> map = new HashMap<>(500000);
synchronizedMap = Collections.synchronizedMap(map);
performanceTest(synchronizedMap);
//ConcurrentHashMap性能測試
concurrentHashMapObject = new ConcurrentHashMap<>(5000000);
performanceTest(concurrentHashMapObject);
}
}
在這里說明一點琳省，這段代碼在不同環(huán)境下運行的結(jié)果會存在差別迎吵，但是結(jié)果的數(shù)量級對比應(yīng)該是一致的，以下是我機子上的運行結(jié)果：
從運行結(jié)果可以看出针贬，在250萬這個數(shù)量級別的數(shù)據(jù)存取上击费，Hashtable 和 SynchronizedMap 的性能表現(xiàn)幾乎一致，畢竟它們的鎖實現(xiàn)原理大同小異桦他，而 ConcurrentHashMap 表現(xiàn)出了比較大的性能優(yōu)勢蔫巩，耗時只有前兩者的三分之一多一點兒。嗯... 以后面試再被問到相關(guān)問題，可以直接把數(shù)據(jù)甩給面試官... ...
三批幌、結(jié)語
好了础锐，以上就是本篇文章的全部內(nèi)容了，完結(jié)撒花荧缘，簡書上不知道怎么調(diào)圖片大小哈哈皆警。第一次寫博客，竟然嘮叨了這么多截粗，不足之處還請各位看官老爺不吝賜教信姓。另外想要進一步深入了解 ConcurrentHashMap 原理的朋友可以看一下下面兩篇文章，是我看過的講的比較詳細(xì)的绸罗。
