Java集合，ConcurrentHashMap底層實現和原理（經常使用於併發編程）
概述
ConcurrentHashMap經常使用於併發編程，這裏就從源碼上來分析一下ConcurrentHashMap數據結構和底層原理。java
在開始以前先介紹一個算法， 這個算法和Concurrent的實現是分不開的。
CAS算法：node
- CAS是英文單詞Compare And Swap的縮寫，翻譯過來就是比較並替換。
- CAS機制當中使用了3個基本操做數：內存地址V，舊的預期值A，要修改的新值B。
- 更新一個變量的時候，只有當變量的預期值A和內存地址V當中的實際值相同時，纔會將內存地址V對應的值修改成B
從思想上來講，Synchronized屬於悲觀鎖，悲觀地認爲程序中的併發狀況嚴重，因此嚴防死守。CAS屬於樂觀鎖，樂觀地認爲程序中的併發狀況不那麼嚴重，因此讓線程不斷去嘗試更新。算法
ConcurrentHashMap是一個線程安全的Map集合，能夠應對高併發的場景，保證線程安全。相比較HashTable，它的鎖粒度更加的細化，由於HashTable的方法都是用Synchronized修飾的，效率灰常的底下。編程
1.8以前ConcurrentHashMap使用鎖分段技術，將數據分紅一段段的存儲，每個數據段配置一把鎖，相互之間不影響，而1.8以後摒棄了Segment（鎖段）的概念，啓用了全新的實現，也就是利用CAS+Synchronized來保證併發更新的安全，底層採用的依然是數組+鏈表+紅黑樹。數組
本篇文章是基於JDK1.8 。安全
數據結構
繼承關係
public class ConcurrentHashMap<K,V>
extends AbstractMap<K,V>
implements ConcurrentMap<K,V>, Serializable
ConcurrentHashMap 繼承了AbstractMap ,而且實現了ConcurrentMap接口。數據結構
與HashMap比對：
- 相同點：都集成了AbstractMap接口
- 不一樣點：HashMap實現了Map接口，ConcurrentHashMap實現了ConcurrentMap接口，而ConcurrentMap繼承了Map接口，使用default關鍵字定義了一些方法 。
從繼承關係上看ConcurrentHashMap與HashMap並無太大的區別。多線程
基本屬性
private static final int MAXIMUM_CAPACITY = 1 << 30; //最大容量2的30次方 private static final int DEFAULT_CAPACITY = 16; //默認容量 1<<4 private static final float LOAD_FACTOR = 0.75f; //負載因子 static final int TREEIFY_THRESHOLD = 8; //鏈表轉爲紅黑樹 static final int UNTREEIFY_THRESHOLD = 6; //樹轉列表 static final int MIN_TREEIFY_CAPACITY = 64; // private static final int MIN_TRANSFER_STRIDE = 16; private static int RESIZE_STAMP_BITS = 16; private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1; private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS; static final int MOVED = -1; // forwarding nodes 的hash值 static final int TREEBIN = -2; // roots of trees 的hash值 static final int RESERVED = -3; // transient reservations 的hash值 static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash static final int NCPU = Runtime.getRuntime().availableProcessors(); //可用處理器數量
重點說一下 sizeCtrl 屬性，這個屬性在 ConcurrentHashMap 中扮演者重要的角色。併發
//表初始化或者擴容的一個控制標識位 //負數表明正在進行初始化或者擴容的操做 // -1 表明初始化 // -N 表明有n-1個線程在進行擴容操做 //正數或者0表示沒有進行初始化操做，這個數值表示初始化或者下一次要擴容的大小。 //transient 修飾的屬性不會被序列化，volatile保證可見性 private transient volatile int sizeCtl;
構造方法
//無參構造方法，沒有進行任何操做
public ConcurrentHashMap() {}
//指定初始化大小構造方法，判斷參數的合法性，並建立了計算初始化的大小
public ConcurrentHashMap(int initialCapacity) {}
//將指定的集合轉化爲ConcurrentHashMap
public ConcurrentHashMap(Map<? extends K, ? extends V> m) {}
//指定初始化大小和負載因子的構造方法
public ConcurrentHashMap(int initialCapacity, float loadFactor) {
this(initialCapacity, loadFactor, 1);
}
//指定初始化大小，負載因子和concurrentLevel併發更新線程的數量，也能夠理解爲segment的個數
public ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel) {}
ConcurrentHashMap的構造方法並沒作太多的工做，主要是進行了參數的合法性校驗，和初始值大小的轉換。這個方法 tableSizeFor()說明一下， 主要的功能就是將指定的初始化參數轉換爲2的冪次方形式， 若是初始化參數爲9 ，轉換後初始大小爲16 。ide
內部數據結構
Node
首當其衝，由於它是ConcurrentHashMap的核心，它包裝了key-value的鍵值對，全部插入的數據都包裝在這裏面，與HashMap很類似，可是有一些差異：
static class Node<K,V> implements Map.Entry<K,V> {
final int hash;
final K key;
volatile V val;
volatile Node<K,V> next;
Node(int hash, K key, V val, Node<K,V> next) {
this.hash = hash;
this.key = key;
this.val = val;
this.next = next;
}
}
value 和 next使用了volatile修飾，保證了線程之間的可見性。也不容許調用setValue()方法直接改變Node的值。並增長了find()方法輔助map.get()方法。
TreeNode
樹節點類，另一個核心的數據結構。當鏈表長度過長的時候，會轉換爲TreeNode。可是與HashMap不相同的是，它並非直接轉換爲紅黑樹，而是把這些結點包裝成TreeNode放在TreeBin對象中，由TreeBin完成對紅黑樹的包裝。並且TreeNode在ConcurrentHashMap集成自Node類，而並不是HashMap中的集成自LinkedHashMap.Entry類，也就是說TreeNode帶有next指針，這樣作的目的是方便基於TreeBin的訪問。
TreeBin
這個類並不負責包裝用戶的key、value信息，而是包裝的不少TreeNode節點。它代替了TreeNode的根節點，也就是說在實際的ConcurrentHashMap「數組」中，存放的是TreeBin對象，而不是TreeNode對象，這是與HashMap的區別。另外這個類還帶有了讀寫鎖。
ForwardingNode
一個用於鏈接兩個table的節點類。它包含一個nextTable指針，用於指向下一張表。並且這個節點的key value next指針所有爲null，它的hash值爲-1. 這裏面定義的find的方法是從nextTable裏進行查詢節點，而不是以自身爲頭節點進行查找
ConcurrentHashMap經常使用方法
initTable 初始化方法
初始化方法是很重要的一個方法，由於在ConcurrentHashMap的構造方法中只是簡單的進行了一些參數校驗和參數轉換的操做。整個Map的初始化是在插入元素的時候觸發的。這一點在下面的put方法中會進行說明。
//執行初始化操做，單線程操做
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
if ((sc = sizeCtl) < 0)
//sizeCtl < 0 表示有線程正在進行初始化操做，從運行狀態變爲就緒狀態。
Thread.yield(); // lost initialization race; just spin
//設置SIZECTL的值爲-1，阻塞其餘線程的操做
//該方法有四個參數
//第一個參數：須要改變的對象
//第二個參數：偏移量
//第三個參數：期待的值
//第四個參數：更新後的值
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")
//初始化Node對象數組
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;
//sc的值設置爲n的0.75倍
sc = n - (n >>> 2); //至關於n*0.75
}
} finally {
sizeCtl = sc; //更改sizeCtl的值
}
break; //中斷循壞返回
}
}
return tab; //返回初始化的值
}
擴容方法
當ConcurrentHashMap 容量不足的時候，須要對table進行擴容，這個方法是支持多個線程併發擴容的，咱們所說的擴容，從本質上來講，無非是從一個數組到另一個數組的拷貝。
擴容方法分爲兩個部分：
- 建立擴容後的新數組，容量變爲原來的兩倍 ，新數組的建立時單線程完成
- 將原來的數組元素複製到新的數組中，這個是多線程操做。
//幫助擴容
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
Node<K,V>[] nextTab; int sc;
if (tab != null && (f instanceof ForwardingNode) &&(nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {
int rs = resizeStamp(tab.length);
while (nextTab == nextTable && table == tab &&(sc = sizeCtl) < 0) {
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || transferIndex <= 0)
break;
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
transfer(tab, nextTab);
break;
}
}
return nextTab;
}
return table;
}
//tab = table ,nextTab 一個Node<Key,Value>[]類型的變量
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
//n 是tab的長度 ， stride 初始值爲0
int n = tab.length, stride;
//判斷cpu處理多線程的能力，若是小於16就直接賦值爲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> 類型數組
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; //將數組長度賦值給transferIndex
}
int nextn = nextTab.length; //獲取新數組的長度
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab); //建立fwd節點
boolean advance = true;
boolean finishing = false; // to ensure sweep before committing nextTab
//使用for循環來處理每一個槽位中的鏈表元素，CAS設置transferIndex屬性值，並初始化i和bound值
// i 指當前的槽位序號，bound值須要處理的邊界，先處理槽位爲15的節點
for (int i = 0, bound = 0;;) {
//建立兩個變量，一個爲Node<K,V> 類型，一個爲int類型
Node<K,V> f; int fh;
while (advance) {
int nextIndex, nextBound;
if (--i >= bound || finishing)
advance = false;
//將transferIndex的值賦值給 nextIndex ,並判斷nextIndex的值是否小於等於0
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
//更新nextIndex的值
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;
//若是table已經複製結束
if (finishing) {
nextTable = null; //清空nextTable
table = nextTab; //把nextTab 賦值給 table
sizeCtl = (n << 1) - (n >>> 1); //閾值設置爲容量的1.5倍
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
}
}
//CAS算法獲取某個數組節點，爲空就設置爲fwd
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
//若是某個節點的hash爲-1，跳過
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) { // fh 當前節點的hash值 若 >= 0
int runBit = fh & n;
Node<K,V> lastRun = f; //將當前節點賦值給 lastRun 節點
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);
}
//在nextTab 的i 位置上放置ln節點
setTabAt(nextTab, i, ln);
//在nextTab 的 i+n 位置上放置 hn節點
setTabAt(nextTab, i + n, hn);
//在tab節點i位置上插入插入forwardNode節點，表示該節點已經處理
setTabAt(tab, i, fwd);
advance = true;
}
//對TreeBin對象進行處理，過程與上面有些相似
//也把節點分類，分別插入到lo和hi爲頭節點的鏈表中
//
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;
}
}
//若是擴容後 不在須要tree結構，反向轉換成鏈表結構
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;
}
}
}
}
}
}
put方法
put操做是最長用的方法，接下來看一下put()方法的具體實現：
- put()要求鍵值都不能爲空
- 須要通過兩次散列， 是數據均勻分散，減小碰撞的次數
- 判斷tab是否進行了初始化，沒有則調用initTable進行初始化操做（單線程）
- 數組i的位置沒有元素存在，直接放入
- 若是i的位置在進行MOVE操做，也就是在進行擴容操做，則多線程幫助擴容
- 若是i的位置有元素存在，則在該節點加鎖Synchronized，判斷是鏈表仍是紅黑樹，按照相應的插入規則插入
final V putVal(K key, V value, boolean onlyIfAbsent) {
//key|value == null 拋出異常
//ConcurrentHashMap不容許鍵或者值爲null的這種狀況發生
//這一點和HashMap有區別
if (key == null || value == null) throw new NullPointerException();
//散列在散列， 讓數據均勻分佈，減小碰撞次數
int hash = spread(key.hashCode()); -->static final int spread(int h) {return (h ^ (h >>> 16)) & HASH_BITS;}
int binCount = 0;
//死循環 至關於while(true) ,將table賦值給 tab
for (Node<K,V>[] tab = table;;) {
//建立一個Node類型的變量f , int 類型的變量 n i fh
Node<K,V> f; int n, i, fh;
//判斷tab是否爲null ,是否進行了初始化操做，若是沒有執行初始化，執行初始化操做
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//tabAt 獲取值
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//添加到table中
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
break; //退出循環 // no lock when adding to empty bin
}
//node的hash值爲 -1
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;
//key 相等，使用新值替換舊值
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;
}
Get方法
Get方法也是最長用的方法，元素放入了，總要取出來
- 根據傳入的key,獲取相應的hash值
- 而後判斷當前的table數組是否爲空
- 計算指定的key在table中存儲的位置
- 鏈表或者紅黑樹轉換相依的方法處理
- 不存在則返回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());
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;
}
//eh< 0 表示紅黑樹節點
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;
}
總結
JDK6，7中的ConcurrentHashmap主要使用Segment來實現減少鎖粒度，把HashMap分割成若干個Segment，在put的時候須要鎖住Segment，get時候不加鎖，使用volatile來保證可見性，當要統計全局時（好比size），首先會嘗試屢次計算modcount來肯定，這幾回嘗試中，是否有其餘線程進行了修改操做，若是沒有，則直接返回size。若是有，則須要依次鎖住全部的Segment來計算。
jdk7中ConcurrentHashmap中，當長度過長，碰撞會很頻繁，鏈表的增改刪查操做都會消耗很長的時間，影響性能，因此jdk8 中徹底重寫了concurrentHashmap,代碼量從原來的1000多行變成了 6000多 行，實現上也和原來的分段式存儲有很大的區別。
主要設計上的變化有如下幾點:
- 不採用segment而採用node，鎖住node來實現減少鎖粒度。
- 設計了MOVED狀態 當resize的中過程當中 線程2還在put數據，線程2會幫助resize。
- 使用3個CAS操做來確保node的一些操做的原子性，這種方式代替了鎖。
- sizeCtl的不一樣值來表明不一樣含義，起到了控制的做用。
參考：
相關文章
- 1. Java集合 --- ConcurrentHashMap底層實現和原理（經常使用於併發編程）
- 2. Java併發編程系列-(8) JMM和底層實現原理
- 3. 徹底掌握JMM和底層實現原理--併發編程
- 4. java集合 底層實現原理---hashMap
- 5. hashMap和concurrentHashMap底層實現原理
- 6. Java併發編程-synchronized底層原理
- 7. Java併發之底層實現原理
- 8. java併發編程筆記2-底層實現原理-volatile
- 9. Java併發編程之synchronized底層實現原理
- 10. Java併發編程——底層實現原理
- 更多相關文章...
- • Rust 併發編程 - RUST 教程
- • 使用Redis和Lua的原子性實現搶紅包功能 - 紅包項目實戰
- • Java Agent入門實戰（三）-JVM Attach原理與使用
- • ☆基於Java Instrument的Agent實現
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. 字節跳動21屆秋招運營兩輪面試經驗分享
- 2. Java 3 年，25K 多嗎？
- 3. mysql安裝部署
- 4. web前端開發中父鏈和子鏈方式實現通信
- 5. 3.1.6 spark體系之分佈式計算-scala編程-scala中trait特性
- 6. dataframe2
- 7. ThinkFree在線
- 8. 在線畫圖
- 9. devtools熱部署
- 10. 編譯和鏈接
歡迎關注本站公眾號,獲取更多信息
相關文章