高併發下ConcurrentHashMap究竟是怎麼實現線程安全的？
前言
這幾天一直在看Java併發相關的知識，又學習了不少新的知識。看見大佬們一個個更博速度如此之快，可想知識淵博如滔滔江水。這一週就學習一些併發知識，搭建了一個ftp服務器，不知道小可愛們這周學習了那些知識呢？微信公衆號【JustKeepCoding】
拋出正題：ConcurrentHashMap是如何實現線程安全，這裏面又隱含那些玄機呢？java
上一節講述了Hashtable這個結構雖然線程安全，可是效率不高，就是由於他的每一個操做都使用了synchronized同步塊。由於synchronized同步塊的線程若是拿到了鎖，就會阻塞其餘線程，阻塞線程會使操做系統從用戶態轉爲阻塞態，從而加大性能消耗。node
因此在高併發的場景下，使用ConcurrentHashMap是最合適不過的了。那麼問題來了，爲何都喜歡使用ConcurrentHashMap這個結構，它究竟是怎麼實現線程安全和併發度的呢？程序員
ConcurrentHashMap的底層究竟是怎麼實現的？
ConcurrentHashMap是面試必問的知識點，裏面涵蓋的知識也比較多，爲此咱們仍是經過源碼分析再切入到Doug Lea（做者）使用的其餘技術。面試
JDK 1.7的ConcurrentHashMap用了哪些技術？
1.7函數構造和變量等
public class ConcurrentHashMap<K, V> extends AbstractMap<K, V>
implements ConcurrentMap<K, V>, Serializable {
static final int DEFAULT_INITIAL_CAPACITY = 16;//默認容量，segments的個數
static final float DEFAULT_LOAD_FACTOR = 0.75f;//負載因子
static final int DEFAULT_CONCURRENCY_LEVEL = 16;//
static final int MAXIMUM_CAPACITY = 1 << 30;//最大容量，仍是左移30位
static final int MIN_SEGMENT_TABLE_CAPACITY = 2;//最小的segment容量爲2
final Segment<K,V>[] segments;//重要組成
}
採用分段鎖技術，將ConcurrentHashMap的內部分紅多個Segment段，每一個Segment能夠編程
Segment組成
static final class Segment<K,V> extends ReentrantLock implements Serializable {
static final int MAX_SCAN_RETRIES =
Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;
transient volatile HashEntry<K,V>[] table;//相似於HashMap中的Entry
transient int modCount;//修改次數
transient int threshold;//閾值
final float loadFactor;//加載因子
}
segment內部是一個HashEntry<K,V>[] table，跟JDK1.7版本的HashMap中的Entry相似，都是數組加鏈表的形式。可是這裏多了一個volatile關鍵字修飾，這樣作到底有什麼好處呢？數組
volatile關鍵字
要弄清楚volatile關鍵字的前因後果須要具有Java內存模型、CPU緩存模型等知識。那麼首先從併發編程的三個重要特性講起：原子性、有序性和可見性。緩存
- 原子性
原子性就是指在一次操做或屢次操做中，要麼全部的操做同時成功，那麼同時失敗。這個在上一篇MySQL的基礎講到過,好比我向淘寶買牙刷，我支出20，那麼淘寶就要收到20。要麼我轉不出來，淘寶也收不到轉。要麼我支出成功，淘寶也收到成功。不可能存在，我支出但淘寶收不到的操做。安全
- 有序性
有序性就是指代碼執行過程當中的前後順序，因爲Java在編譯期間的優化，致使代碼執行順序未必是順序執行，由於JVM可能會指令重排序。服務器
- 可見性
可見性是指當一個線程對一個變量進行修改，另外一個線程能夠當即看到修改的最新值。這就相似於MySQL的讀取未提交的事務隔離級別，防止讀取髒數據。微信
那麼JVM是如何來保證可見性的呢？
- 經過volatile關鍵字，對於共享資源的讀操做會直接在主內存中進行，對於共享數據的寫操做固然是先修改工做內存，可是修改後會主動刷新到主內存中。
- 經過synchronized關鍵字，synchronized只能保證同一時刻只有一個線程得到鎖，而後執行同步方法，而且還會保證在鎖釋放以前，會將對變量的修改會主動刷新到主內存。
- 經過JUC提供的顯式鎖Lock也能保證，Lock的lock方法可以保證在同一時刻只有一個線程得到鎖而後執行同步方法，並保證在鎖釋放以前（Lock的unlock）將修改的變量刷新到主內存。
咱們假設一個場景，當一個讀線程自身有個Init_value的值，每次都須要跟當前值比較，若是值被修改了就打印輸出，沒修改就不輸出。當一個修改線程，每次都會修改當前的值，修改會會進行sleep睡眠10毫秒，用於等待讀線程輸出，即便其交替輸出。（由於代碼太長就不貼了）
你猜結果是怎樣的？
結果是修改線程一直輸出，讀線程在控制檯沒有任何打印。這是由於，共享數據開始是緩存到CPU的cache快裏，即讀線程的本地內存中。即便修改線程修改數據，讀線程也是從本地取，而本地內存值沒有被修改，因此不會輸出。那麼使用volatile關鍵字，就能夠強制讀線程從內存刷新到本地線程內存（CPU的cache中）。
volatile關鍵字語義：
- 保證可見性
- 保證順序性
- 不保證原子性
因此回到HashEntry這裏，採用volatile關鍵字就是爲了併發環境下保證HashEntry的有序性和可見性。
ReentrantLock 的使用與理解
從源碼能夠看到Segment繼承ReentrantLock，ReentrantLock 又是一個很重要的知識點，咱們都知道能夠用互斥同步來保證線程的安全性，而最基本的同步互斥手段就是synchronized和java.util.concurrent(J.U.C)包下的ReentrantLock 來實現同步。在基本用法上，ReentrantLock 和synchronized很類似，但在JDK1.5以前ReentrantLock 的功能是比較豐富的，在JDK1.8中運用普遍。
JDK 1.8的ConcurrentHashMap又採用了哪些技術呢？
悲觀鎖策略（阻塞同步）
互斥同步最主要的問題就是線程阻塞和喚醒帶來的性能問題，所以這種同步也稱阻塞同步。從處理方式來說，互斥同步是一種悲觀的併發策略，老是認爲只要不去作正確的的同步措施（列入加鎖），那確定會出現問題，不管共享數據是否出現競爭，都須要加鎖。（須要線程掛起）
樂觀鎖策略（非阻塞同步）
基於衝突檢測的樂觀併發策略就是先進行操做，若是沒有其餘線程爭用共享資源，那麼操做就成功了。若是共享數據有爭用，出現衝突，那就再採起其餘的不就措施。（不須要線程掛起）
put操做
public V put(K key, V value) {//定位到segment
Segment<K,V> s;
if (value == null)
throw new NullPointerException();//若是爲NULL則會出現異常
int hash = hash(key);
int j = (hash >>> segmentShift) & segmentMask;
if ((s = (Segment<K,V>)UNSAFE.getObject
(segments, (j << SSHIFT) + SBASE)) == null)
s = ensureSegment(j);
return s.put(key, hash, value, false);//再從HashEntry中插入
}
/**
**segment數組中的put操做，相似於HashMap中的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;//計算segment[i]中的table的具體位置,這些就相似hashmap的操做了
HashEntry<K,V> first = entryAt(tab, index);//尋找table其中一個頭結點
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節點去獲取鎖（自旋鎖）
**/
private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
HashEntry<K,V> first = entryForHash(this, hash);//獲取插入值當前鏈表頭結點
HashEntry<K,V> e = first;
HashEntry<K,V> node = null;//新的節點
int retries = -1; // 嘗試獲取鎖的次數，初始爲-1
while (!tryLock()) {//沒有獲取到鎖的話
HashEntry<K,V> f; // to recheck first below
if (retries < 0) {//進入第一種狀況
if (e == null) {//頭結點爲空，建立新的節點
if (node == null) // speculatively create node
node = new HashEntry<K,V>(hash, key, value, null);//爲了少作一點事，之後直接使用
retries = 0;
}
else if (key.equals(e.key))
retries = 0;
else
e = e.next;
}
else if (++retries > MAX_SCAN_RETRIES) {//第二種狀況，超過自旋次數加以阻塞
lock();//獲取不到鎖，則阻塞等待
break;
}
else if ((retries & 1) == 0 &&
(f = entryForHash(this, hash)) != first) {
//由於當前沒有獲取鎖，可能頭結點被更另外一個線程更改，須要判斷是不是以前的頭結點。若是不是則從新循環判斷
e = first = f; // re-traverse if entry changed
retries = -1;
}
}
return node;
}
tryLock()會嘗試獲取鎖，獲取不到就返回false，獲取的到就返回true，作其餘事，而且不會阻塞。Lock()方法若是獲取不到鎖就會阻塞。
流程圖：
擴容操做
/**
**擴容操做，不須要rehash，直接使用以前的hash
**/
private void rehash(HashEntry<K,V> node) {
HashEntry<K,V>[] oldTable = table;//舊的hash表
int oldCapacity = oldTable.length;//以前的hash表長
int newCapacity = oldCapacity << 1;//擴容兩倍
threshold = (int)(newCapacity * loadFactor);
HashEntry<K,V>[] newTable =
(HashEntry<K,V>[]) new HashEntry[newCapacity];//擴容後的hash表
int sizeMask = newCapacity - 1;//爲了計算數組下標
for (int i = 0; i < oldCapacity ; i++) {
HashEntry<K,V> e = oldTable[i];
if (e != null) {
HashEntry<K,V> next = e.next;
int idx = e.hash & sizeMask;
if (next == null) // Single node on list
newTable[idx] = e;
else { // Reuse consecutive sequence at same slot
HashEntry<K,V> lastRun = e;//每次記錄最後一串index相同的節點，從新賦值
int lastIdx = idx;
for (HashEntry<K,V> last = next;
last != null;
last = last.next) {//遍歷節點，記錄最後index值相同的一串節點
int k = last.hash & sizeMask;
if (k != lastIdx) {//最後的節點的index不跟數組的index值相同時
lastIdx = k;//更新到最後節點的index
lastRun = last;
}
}
newTable[lastIdx] = lastRun;
// Clone remaining nodes
for (HashEntry<K,V> p = e; p != lastRun; p = p.next) {//將最後一串index值相同的從新移到新的數組下標裏
V v = p.value;
int h = p.hash;
int k = h & sizeMask;
HashEntry<K,V> n = newTable[k];
newTable[k] = new HashEntry<K,V>(h, p.key, v, n);
}
}
}
}
int nodeIndex = node.hash & sizeMask; // 對新增的節點進行獲取index
node.setNext(newTable[nodeIndex]);//頭插法
newTable[nodeIndex] = node;
table = newTable;
}
get操做
get 就比較簡單啦，它是將 Key 經過 Hash 以後定位到具體的 Segment ，再經過一次 Hash 定位到具體的元素上。
即便多個線程對HashEntry修改，有volatile修飾的HashEntry，每次都會get到最新值。
結論
爲何線程安全呢，就是由於在1.7版本它的每一個方法都會加鎖，put採用自旋鎖不會使線程阻塞（操做狀態切換會消耗性能），從而性能比hashtable好不少。並且它的每一個HashEntry[i]都是被volatile修飾，能夠保證線程操做的可見性，即每次不會髒讀，即便其餘線程修改了值，都會強制刷新到本地內存。
那爲何併發度高呢？
由於是對單個segment[i]進行加鎖，意思就是segment若是有16個，那麼能夠同時有16個線程修改並且仍是線程安全的。相對於Hashtable的鎖，是鎖定整個Hashtable對象，那麼多個線程訪問就須要被阻塞。
JDK1.8的ConcurrentHashMap
1.8版本取消了1.7中的分段鎖策略，使用了CAS + synchronized來保證安全性。這跟1.8版本的HashMap也很類似，引入了紅黑樹，在鏈表節點數大於8的時候會轉換成紅黑樹。
1.8源碼構造
public class ConcurrentHashMap<K,V> extends AbstractMap<K,V>
implements ConcurrentMap<K,V>, Serializable {
private static final int MAXIMUM_CAPACITY = 1 << 30;//最大容量
private static final int DEFAULT_CAPACITY = 16;//默認容量
private static final float LOAD_FACTOR = 0.75f;//負載因子
static final int TREEIFY_THRESHOLD = 8;//鏈表節點數大於8 轉紅黑樹
static final int UNTREEIFY_THRESHOLD = 6;//刪除紅黑樹節點小於6轉鏈表
transient volatile Node<K,V>[] table;//節點桶，volatile修飾
private transient volatile Node<K,V>[] nextTable;//下一張表
}
1.8的put方法
final V putVal(K key, V value, boolean onlyIfAbsent) {//put中調用putVal
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;
if (tab == null || (n = tab.length) == 0)
tab = initTable();//初始化表
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {//獲取table[i]的頭結點是空
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))//CAS操做，寫入node
break; // no lock when adding to empty bin
}
else if ((fh = f.hash) == MOVED)//當此時另外一個線程正在擴容
tab = helpTransfer(tab, f);//當前線程幫助擴容，增長效率
else {//進入table節點，判斷當前節點是鏈表節點仍是紅黑樹節點
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)//鏈表節點大於8
treeifyBin(tab, i);//轉紅黑樹
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);//統計節點數
return null;
}
TreeBin保證了即便紅黑樹自平衡，也不會釋放鎖資源。由於若是使用HashMap的紅黑樹結構的話，當插入節點會自平衡使頭結點更換，而頭結點是獲取到鎖了，若是頭結點更換，其餘線程此時就會獲取到該節點的鎖，形成插入的異常。
put流程：
- 1.先判斷key和value是否爲空，爲空拋出異常
-
2.計算出hash值，遍歷表結構，進入下面三種狀況
- 是否須要初始化表，空則初始化，再進入2
- 獲取table[i]的頭節點，空則用
CAS插入(key,value),進入2 - 看是否當前有線程正在擴容，有則幫忙擴容
helpTransfer，進入2 -
synchronized獲鎖資源，判斷是鏈表節點仍是紅黑樹節點，插入
- 3.統計節點數
- 返回空
比較並替換（Compare-and-swap，簡稱爲CAS）
樂觀鎖策略須要CAS指令來保證進行，就是實現樂觀鎖的一種方式，CAS操做是由sun.misc.Unsafe類中的compareAndSwapInt()和compareAndSwapLong()等幾個方法包裝提供，是一種輕量級鎖，J.U.C中不少工具類就是基於CAS實現的，好比CompareAndSet和getAndIncrement()等都使用了Unsafe類的CAS操做。
CAS操做流程就是線程在讀取數據時不進行加鎖，在準備寫回數據時，比較原值是否修改，若未被其餘線程修改則寫回，若已被修改，則從新執行讀取流程。
1.8 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) {//hash桶存在，且當前的桶不爲空
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;
}
總結：
1.8在1.7上作的改動仍是挺大的，ConcurrentHashMap這個數據結構的知識仍是挺多的，好比CAS的操做和理解、1.8中synchronized關鍵字的優化、1.8中的擴容方法插入方法、紅黑樹自平衡等等。都是Java程序員面試的必備知識點，這節又寫了11000多字，挺辛苦的，因此可以給我點個贊嗎？
最後拋出兩個問題：
- 1.8中的ConcurrentHashMap爲何沒有使用
Lastrun（還記得1.7中擴容的Lastrun嗎） - unsafe類你屬於嗎？Java程序中最牛逼的類，爲何最牛逼呢？
你的點贊就是我最大的努力！關注個人公衆號【JustKeepCoding】，每週兩篇爲你推送不同的知識！
- 1. 高併發下ConcurrentHashMap究竟是怎麼實現線程安全的？
- 2. 高併發下的線程安全實現——互斥同步
- 3. ConcurrentHashMap實現線程安全的原理
- 4. ConcurrentHashMap：實現線程安全的HashMap
- 5. 《剖析Java線程究竟是並行仍是併發》
- 6. 線程安全究竟是什麼意思？
- 7. Java併發之volatile究竟是什麼?
- 8. C++怎麼實現線程安全
- 9. 多線程——探索 ConcurrentHashMap 高併發性的實現機制
- 10. 5G的調製方式，究竟是怎麼實現的？
- 更多相關文章...
- • Rust 併發編程 - RUST 教程
- • ASP.NET MVC - 安全 - ASP.NET 教程
- • ☆基於Java Instrument的Agent實現
- • Spring Cloud 微服務實戰(三) - 服務註冊與發現
-
每一个你不满意的现在，都有一个你没有努力的曾经。
- 1. 跳槽面試的幾個實用小技巧，不妨看看！
- 2. Mac實用技巧 |如何使用Mac系統中自帶的預覽工具將圖片變成黑白色?
- 3. Mac實用技巧 |如何使用Mac系統中自帶的預覽工具將圖片變成黑白色?
- 4. 如何使用Mac系統中自帶的預覽工具將圖片變成黑白色?
- 5. Mac OS非兼容Windows軟件運行解決方案——「以VMware & Microsoft Access爲例「
- 6. 封裝 pyinstaller -F -i b.ico excel.py
- 7. 數據庫作業三ER圖待完善
- 8. nvm安裝使用低版本node.js(非命令安裝)
- 9. 如何快速轉換圖片格式
- 10. 將表格內容分條轉換爲若干文檔