更新時間:2022-08-24 來源:黑馬程序員 瀏覽量:
一、糟糕的面試
面試官：小王，你說說HashMap的是線程安全的嗎?
小王：HashMap不安全，在多線程下，會出現線程安全問題。他兄弟HashTable
線程是安全的，但是出于性能考慮，我們往往會選擇ConcurrentHashMap。
面試官：HashMap線程不安全的原因是什么?
小王：這個...暫時忘記了
面試官：為什么HashTable線程安全，為什么性能低?
小王：這個...
面試官：ConcurrentHashMap是怎么實現線程安全的?性能為什么較高?
小王：...
面試官：回答的很不錯，回去等通知吧。
二、hashMap
2.1 暴露問題
大家都知道，HashMap在多線程下會存在線程安全問題，如下：
JavaEE
鴻蒙應用開發
集成電路應用開發
人工智能開發
AI+Linux云計算運維
Python+大數據開發
AI+設計
軟件測試
新媒體+短視頻
HTML&JS+前端
```java
public class Demo2 {
public static void main(String[] args) {
//shift+ctrl+alt+u
HashMap<String, String> map = new HashMap<>();
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
for (int i = 0; i <= 10; i++) {
map.put(i+"",i+"");
}
}
});
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
for (int i = 11; i <= 20; i++) {
map.put(i+"",i+"");
}
}
});
t1.start();
t2.start();
//確保兩個子線程執行完畢之后，主線程再來打印hashmap
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
//遍歷hashMap
for (int i = 1; i <= 20; i++) {
System.out.println(map.get(i + ""));
}
}
}
```控制臺：
```
null
2
null
null
null
6
7
8
9
10
null
null
13
null
null
null
17
18
19
20
```
以上例子證明了,HashMap確實存在線程安全問題。
2.2 源碼追蹤
翻閱源碼(1.8)如下：
```java
final V putVal(int hash, K key, V value, boolean onlyIfAbsent,
boolean evict) {
Node<K,V>[] tab; Node<K,V> p; int n, i;
if ((tab = table) == null || (n = tab.length) == 0)
n = (tab = resize()).length;
//此處線程不安全
if ((p = tab[i = (n - 1) & hash]) == null)
tab[i] = newNode(hash, key, value, null);
else {
Node<K,V> e; K k;
if (p.hash == hash &&
((k = p.key) == key || (key != null && key.equals(k))))
e = p;
else if (p instanceof TreeNode)
e = ((TreeNode<K,V>)p).putTreeVal(this, tab, hash, key, value);
else {
for (int binCount = 0; ; ++binCount) {
if ((e = p.next) == null) {
p.next = newNode(hash, key, value, null);
if (binCount >= TREEIFY_THRESHOLD - 1) // -1 for 1st
treeifyBin(tab, hash);
break;
}
if (e.hash == hash &&
((k = e.key) == key || (key != null && key.equals(k))))
break;
p = e;
}
}
if (e != null) { // existing mapping for key
V oldValue = e.value;
if (!onlyIfAbsent || oldValue == null)
e.value = value;
afterNodeAccess(e);
return oldValue;
}
}
++modCount;
//此處線程不安全。
if (++size > threshold)
resize();
afterNodeInsertion(evict);
return null;
}
```(1)代碼一
```java if ((p = tab[i = (n - 1) & hash]) == null) tab[i] = newNode(hash, key, value, null); ```
是否Hash沖突,沒沖突就直接賦值給數組當前索引。
線程A判斷通過，進入方法，切換B線程，判斷通過，進入方法，切換A線程，賦值成功，切換B線程賦值成功，B線程的值覆蓋了A線程的值，發生了數據覆蓋，用戶感受到是數據丟失。
(2) 代碼二
```java if (++size > threshold) resize(); ```
當元素個數size大于擴容閾值，則擴容，這里會有兩個問題。
- 成員的size變量沒有保證原子性，因此多線程下size自增是存在原子性問題。即添加了兩個元素，但是size只增加了1。
- 兩個線程如果都通過上面閾值的判斷，就會發生擴容兩次的情況，這也是一種安全問題。
三、HashTable
3.1 線程安全演示
我們可以使用HashTable來解決上面的安全問題。
看下面的代碼：
```java
public class Demo2 {
public static void main(String[] args) {
Hashtable<String, String> map = new Hashtable<>();
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
for (int i = 0; i <= 10; i++) {
map.put(i+"",i+"");//{i,i}
}
}
});
//20,20 21,21 ... 39,39
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
for (int i = 11; i <= 20; i++) {
map.put(i+"",i+"");//{i,i}
}
}
});
t1.start();
t2.start();
//確保兩個子線程執行完畢之后，主線程再來打印hashmap
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
for (int i = 0; i <= 20; i++) {
System.out.println(map.get(i + ""));
}
}
}
```控制臺：
```java
0
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
```
以上代碼，說明了Hashtable的確是線程安全的。
3.2 翻看源碼
Hashtable源碼：
```java
public synchronized V put(K key, V value) {
// Make sure the value is not null
if (value == null) {
throw new NullPointerException();
}
// Makes sure the key is not already in the hashtable.
Entry<?,?> tab[] = table;
int hash = key.hashCode();
int index = (hash & 0x7FFFFFFF) % tab.length;
@SuppressWarnings("unchecked")
Entry<K,V> entry = (Entry<K,V>)tab[index];
for(; entry != null ; entry = entry.next) {
if ((entry.hash == hash) && entry.key.equals(key)) {
V old = entry.value;
entry.value = value;
return old;
}
}
addEntry(hash, key, value, index);
return null;
}
public synchronized V get(Object key) {
Entry<?,?> tab[] = table;
int hash = key.hashCode();
int index = (hash & 0x7FFFFFFF) % tab.length;
for (Entry<?,?> e = tab[index] ; e != null ; e = e.next) {
if ((e.hash == hash) && e.key.equals(key)) {
return (V)e.value;
}
}
return null;
}
public synchronized int size() {
return count;
}
```通過閱讀源碼可以發現，Hashtable每個操作數據的方法，都是使用了重量級鎖synchronized。線程A在操作數據的時候，線程B只能阻塞。保證了整個Hash表只能線程串行化執行，從而解決了多線程產生的安全問題。
因為Hashtable是對整個哈希表進行加鎖，加鎖粒度過大，發生線程阻塞的概率非常大，雖然synchronized有自己的鎖優化機制，但是也很快就會升級成重量級鎖。而當synchronized成為了重量級鎖，就會請求底層系統鎖，跳出jvm級別，頻繁涉及用戶態和內核態的切換，性能開銷比較大。
所以在今天已經不推薦使用HashTable了。
四、ConcurrentHashMap
4.1 線程安全演示
以上兩個章節我們發現，在Map集合中HashMap是最常用的集合對象。但是多線程操作HashMap會有線程安全問題，解決方式就是使用HashTable，但是HashTable會全表加鎖性能犧牲很大。
JDK1.5以后所提供了ConcurrentHashMap，使用它既能解決線程安全問題，性能又比HashTable高很多，所以這是目前主流的折中方案。
代碼如下：
```java
public class Demo3 {
public static void main(String[] args) {
ConcurrentHashMap<String, String> map = new ConcurrentHashMap<>();
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
for (int i = 0; i <= 10; i++) {
map.put(i + "", i + "");//{i,i}
}
}
});
//20,20 21,21 ... 39,39
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
for (int i = 11; i <= 20; i++) {
map.put(i + "", i + "");//{i,i}
}
}
});
t1.start();
t2.start();
//確保兩個子線程執行完畢之后，主線程再來打印hashmap
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
for (int i = 0; i <= 20; i++) {
System.out.println(map.get(i + ""));
}
}
}
```控制臺：
```java
0
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
```
線程安全得到保證。
總結 :
1 ，HashMap是線程不安全的。多線程環境下會有數據安全問題
2 ，Hashtable是線程安全的，但是會將整張表鎖起來，效率低下
3，ConcurrentHashMap也是線程安全的，效率較高。 在JDK7和JDK8中，底層原理不一樣。
4.2 jdk7原理解析
1.ConcurrentHashMap集合底層是一個默認長度為16，加載因子為0.75的大數組 Segment數組。大數組通常是創建之后長度就固定的，而擴容是指小數組擴容。
2.默認情況下還會創建一個長度為2的小數組，把地址賦值給0索引處，其他索引此時的元素仍為null。
`(Segment` 繼承 `ReentrantLock` 鎖,用于存放數組 `HashEntry[]`。)
如下圖
3.調用put方法時，此時會根據key的哈希值來計算出在大數組中存儲的索引位置。
如果這個索引此時為null，則按照0素引的模板小數組來創建小數組。創建完畢后會**二次哈希**計算出key在小數組中存儲的位置，然后把鍵值對對象存儲小數組的該索引位。
如下圖，先根據key的哈希算出來在大數組的4索引，創建小數組掛在4索引。接著繼續使用key的hash算出存在小數組的0素引。
4.調用put方法時，此時會根據key的哈希值來計算出在大數組中存儲的索引位置。
如果該位置不為null，就會根據記錄的地址值找到小數組。二次哈希計算出小數組的索引位置。
如果需要擴容就把小數組擴容2倍。
如果不需要擴容，則會判斷小數組該索引是否有元素
如果沒有元素，就直接存
如果有元素，就調用equals方法比較key是否相同
比較發現沒有重復，就會在小數組上掛鏈表。
如下圖
線程一來訪問索引4，此時就會對索引4的Segment進行加鎖。其他線程訪問索引4就會阻塞，訪問其他索引就可以訪問，這種技術叫分段鎖，將數據拆成一段一段的進行加鎖。
在當前例子中，我們沒有指定大數組的長度，因此長度默認是 16。在理想情況下，最多可以支持16個線程同時操作不同的segment對象，達到了并發的目的。但是如果多個線程同時操作同一個segment，就會阻塞，串行化執行。
關鍵字：分段鎖、二次哈希、Segment數組不能擴容、HashEntry數組可以擴容
總結：
ConcurrentHashMap1.7使用Segment+HashEntry數組實現的。本質上是一個 Segment 數組，Segment 繼承 ReentrantLock ，同時具備了加鎖和釋放鎖的功能。每個Segment都線程安全，全局也就安全了。把Hashtable的鎖全表，變成了鎖一段段的數據，粒度細提高性能。
補充：ConcurrentHashMap1.8則完全不同，放棄了Segment。數據結構使用synchronized+CAS+紅黑樹。鎖的粒度也從段鎖縮小為結點鎖，粒度更細，同時數組支持擴容，并發能力更高。使用synchronized其實也是因為1.6jdk對synchronized的優化有關。
4.3 jdk8 原理解析
在1.8中，ConcurrentHashMap可以說發生翻天覆地的變化，底層數據結構不再采用segment數組，也不再采用分段鎖。而是采用 數組+鏈表+紅黑樹來實現，鎖也從分段鎖提升成了節點鎖，粒度更細。使用CAS+synchronized來保證線程安全。
底層結構：數組+鏈表+紅黑樹
CAS + synchronized同步代碼塊 保證線程安全
初始化數組源碼如下：
```java
//假設多線程來擴容，concurrentHashMap為了線程安全，只能讓一個線程成功初始化數組，其他線程均失敗。
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
//所有線程進入循環，去搶著初始化數組
while ((tab = table) == null || tab.length == 0) {
if ((sc = sizeCtl) < 0)
Thread.yield(); //讓線程讓出cpu，以至于把cpu更多的可能讓給初始化操作的線程
//CAS操作，保證一個線程進入下面的邏輯，其他線程最終只能執行 Thread.yield();
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
if ((tab = table) == null || tab.length == 0) {
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
//這個就是初始化數組，默認長度n為16
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}
```通過源碼發現，這里使用了 自旋+CAS+線程讓出cpu。
其中 自旋+Cas：初始化操作必須且只會由一個線程執行一次，不會初始化多個數組。
線程讓出cpu：提高性能，讓初始化操作更快執行
put操作源碼如下：
```java
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;
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
//如果數組是null,說明第一次put，那就初始化數組
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//根據key找到索引，如果此索引為null，則使用CAS將node直接賦給當前索引
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
}
//如果當前索引位置的元素的hash==MOVED說明此時正在發生數組擴容的數據遷移操作，
//當前線程幫助完成數據遷移
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
//當前索引既不是null，也沒有在數據遷移。此時索引位置存儲的要么鏈表要么紅黑樹
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;
}
}
}
addCount(1L, binCount);
return null;
}
```通過源碼發現:
put操作如果沒有發生hash沖突，則CAS直接賦值到索引
如果發生了hash沖突，判斷此時是否正在擴容數據遷移，是就加入幫忙數據遷移
如果此時是鏈表或者紅黑樹，就加節點鎖，保證當前索引操作的線程串行化執行。
擴容源碼如下：
```java
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
//計算步長，算法的目的是，cpu核數越多，步長越小。
//步長最少為16，最多為數組的長度
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
if (nextTab == null) { // initiating
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<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 (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) {
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
}
}
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
else {
synchronized (f) {
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;
if (fh >= 0) {
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);
else
hn = new Node<K,V>(ph, pk, pv, hn);
}
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
setTabAt(tab, i, fwd);
advance = true;
}
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;
else
loTail.next = p;
loTail = p;
++lc;
}
else {
if ((p.prev = hiTail) == null)
hi = p;
else
hiTail.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;
}
}
}
}
}
}
```通過源碼發現：
ConcurrentHashMap采用的是多線程擴容，來提高擴容的效率。總體思想是，cpu核數越多線程越多，每個線程分得數據遷移任務量越小。
步長：單個線程負責遷移的桶數量。
下面做一個模擬擴容的流程：
假設現在數組長度有512，cpu核數2，步長32。
線程一：負責遷移索引(512-步長-1，512-1)，即數組[479] - 數組[511]
線程二：負責遷移索引 數組[446] 數組[478]
如果線程一和線程二都執行完畢，兩個線程就會通過CAS去搶下一個任務
如果線程二搶到了，數組[413] 數組[445]
線程一失敗了，自旋繼續CAS搶，搶到了
線程一 數組[381] 數組[412]
4.4 對比區別
(1)1.7用的是segment+hashentry數組實現的分段鎖。只要線程沒同時訪問同一個分段數組，就可以并行訪問默認長度16，segment數組不可以擴容(大數組)，hashentry數組可以擴容。
(2)1.8用的是CAS+synchronized+voletile 實現的，底層是數組+鏈表+紅黑樹。對比1.7 鎖的粒度更細，鎖到了節點級別。
(3)1.8為什么synchronized替換segment?ReentrantLock：park unpark 用戶態 - 內核態 性能開銷比較大。synchronized：1.6之后優化了，偏向鎖 輕量級鎖 。此時加鎖粒度非常小，比1.7小。轉成重量級鎖概率極小。
五、總結
通過上面的學習得知，hashmap在多線程情況下初始化數組和擴容的時候均會出現線程安全問題。我們可以通過HashTable來解決，HashTable是對整個hash表加鎖，相當于線程串行化操作hash表，在解決問題的同時也會導致性能極低。最終我們可以使用ConcurrentHashMap將鎖的粒度控制到最小，將性能影響控制到最低，同時擴容的時候ConcurrentHashMap還支持多線程擴容。可以說ConcurrentHashMap是多線程操作hashmap場景的不二之選，比如SpringCache框架中就使用了ConcurrentHashMap來作為本地緩存。
.jpg)