ConcurrentHashMap1.7和1.8的底層不一樣實現
1.Hashmap和HashTable在線程安全方面的優劣？
Hashmap多線程會致使HashMap的Entry鏈表造成環形數據結構，一旦造成環形數據結構，Entry的next節點永遠不爲空，就會產生死循環獲取Entry。java
HashTable使用synchronized來保證線程安全，但在線程競爭激烈的狀況下HashTable的效率很是低下。由於當一個線程訪問HashTable的同步方法，其餘線程也訪問HashTable的同步方法時，會進入阻塞或輪詢狀態。如線程1使用put進行元素添加，線程2不但不能使用put方法添加元素，也不能使用get方法來獲取元素，因此競爭越激烈效率越低。node
2.預備知識
Hash
散列，哈希：把任意長度的輸入經過一種算法（散列），變換成爲固定長度的輸出，這個輸出值就是散列值。屬於壓縮映射，容易產生哈希衝突。Hash算法有直接取餘法等。面試
產生哈希衝突時解決辦法：開放尋址；二、再散列；三、鏈地址法（相同hash值的元素用鏈表串起來）。算法
ConcurrentHashMap在發生hash衝突時採用了鏈地址法。數組
md4,md5,sha-hash算法也屬於hash算法，又稱摘要算法。安全
位運算
2的0次方 = 1，2的1次方=2…….，以上表格表明數字 （2的5次方+2的3次方）= 40數據結構
由上面的表格能夠看出，數字類型在數字漸漸變大時，是由低位慢慢向高位擴展的。多線程
Java實際保存int型時 正數 第31位 =0 負數：第31位=1併發
經常使用位運算有：dom
- 位與 & (1&1=1 1&0=0 0&0=0)
- 位或 | (1|1=1 1|0=1 0|0=0)
- 位非 ~ （ ~1=0 ~0=1）
- 位異或 ^ (1^1=0 1^0=1 0^0=0)
- <<有符號左移 >>有符號的右移 >>>無符號右移 例如：8 << 2 = 32 8>>2 = 2
- 取模的操做 a % (Math.pow(2,n)) 等價於 a&( Math.pow(2,n)-1)
位運算適用：權限控制，物品的屬性很是多時的保存
3. 1.7中原理和實現
-
ConcurrentHashMap中的數據結構
ConcurrentHashMap是由Segment數組結構和HashEntry數組結構組成。Segment實際繼承自可重入鎖（ReentrantLock），在ConcurrentHashMap裏扮演鎖的角色；HashEntry則用於存儲鍵值對數據。一個ConcurrentHashMap裏包含一個Segment數組，每一個Segment裏包含一個HashEntry數組，咱們稱之爲table，每一個HashEntry是一個鏈表結構的元素。
面試常問：
- ConcurrentHashMap實現原理是怎麼樣的或者問ConcurrentHashMap如何在保證高併發下線程安全的同時實現了性能提高？
答：ConcurrentHashMap容許多個修改操做併發進行，其關鍵在於使用了鎖分離技術。它使用了多個鎖來控制對hash表的不一樣部分進行的修改。內部使用段(Segment)來表示這些不一樣的部分，每一個段其實就是一個小的hash table，只要多個修改操做發生在不一樣的段上，它們就能夠併發進行。
-
初始化作了什麼事？
初始化有三個參數
initialCapacity：初始容量大小 ，默認16。
loadFactor, 擴容因子，默認0.75，當一個Segment存儲的元素數量大於initialCapacity* loadFactor時，該Segment會進行一次擴容。
concurrencyLevel 併發度，默認16。併發度能夠理解爲程序運行時可以同時更新ConccurentHashMap且不產生鎖競爭的最大線程數，實際上就是ConcurrentHashMap中的分段鎖個數，即Segment[]的數組長度。若是併發度設置的太小，會帶來嚴重的鎖競爭問題；若是併發度設置的過大，本來位於同一個Segment內的訪問會擴散到不一樣的Segment中，CPU cache命中率會降低，從而引發程序性能降低。
構造方法中部分代碼解惑：
一、
保證Segment數組的大小，必定爲2的冪，例如用戶設置併發度爲17，則實際Segment數組大小則爲32
二、
保證每一個Segment中tabel數組的大小，必定爲2的冪，初始化的三個參數取默認值時，table數組大小爲2
三、
初始化Segment數組，並實際只填充Segment數組的第0個元素。
四、
用於定位元素所在segment。segmentShift表示偏移位數，經過前面的int類型的位的描述咱們能夠得知，int類型的數字在變大的過程當中，低位老是比高位先填滿的，爲保證元素在segment級別分佈的儘可能均勻，計算元素所在segment時，老是取hash值的高位進行計算。segmentMask做用就是爲了利用位運算中取模的操做： a % (Math.pow(2,n)) 等價於 a&( Math.pow(2,n)-1)
-
在get和put操做中，是如何快速定位元素放在哪一個位置的？
對於某個元素而言，必定是放在某個segment元素的某個table元素中的，因此在定位上，
定位segment：取得key的hashcode值進行一次再散列（經過Wang/Jenkins算法），拿到再散列值後，以再散列值的高位進行取模獲得當前元素在哪一個segment上。
定位table：一樣是取得key的再散列值之後，用再散列值的所有和table的長度進行取模，獲得當前元素在table的哪一個元素上。
-
get（）方法
定位segment和定位table後，依次掃描這個table元素下的的鏈表，要麼找到元素，要麼返回null。
在高併發下的狀況下如何保證取得的元素是最新的？
答：用於存儲鍵值對數據的HashEntry，在設計上它的成員變量value等都是volatile類型的，這樣就保證別的線程對value值的修改，get方法能夠立刻看到。
-
put()方法
一、首先定位segment，當這個segment在map初始化後，還爲null，由ensureSegment方法負責填充這個segment。
- 對Segment 加鎖
三、定位所在的table元素，並掃描table下的鏈表，找到時：
沒有找到時：
-
擴容操做
Segment 不擴容，擴容下面的table數組，每次都是將數組翻倍
帶來的好處
假設原來table長度爲4，那麼元素在table中的分佈是這樣的：
| Hash值 |
15 |
23 |
34 |
56 |
77 |
| 在table中下標 |
3 = 15%4 |
3 = 23 % 4 |
2 = 34%4 |
0 = 56%4 |
1 = 77 % 4 |
擴容後table長度變爲8，那麼元素在table中的分佈變成：
| Hash值 |
56 |
|
34 |
|
|
77 |
|
15,23 |
| 下標 |
0 |
1 |
2 |
3 |
4 |
5 |
6 |
7 |
能夠看見 hash值爲34和56的下標保持不變，而15,23,77的下標都是在原來下標的基礎上+4便可，能夠快速定位和減小重排次數。
-
size方法
size的時候進行兩次不加鎖的統計，兩次一致直接返回結果，不一致，從新加鎖再次統計
-
弱一致性
get方法和containsKey方法都是經過對鏈表遍歷判斷是否存在key相同的節點以及得到該節點的value。但因爲遍歷過程當中其餘線程可能對鏈表結構作了調整，所以get和containsKey返回的多是過期的數據，這一點是ConcurrentHashMap在弱一致性上的體現。
4. 1.8中的原理和實現
-
數據結構
JDK1.8的實現已經摒棄了Segment的概念，而是直接用Node數組+鏈表+紅黑樹的數據結構來實現，併發控制使用Synchronized和CAS來操做，整個看起來就像是優化過且線程安全的HashMap，雖然在JDK1.8中還能看到Segment的數據結構，可是已經簡化了屬性，只是爲了兼容舊版本
在深刻JDK1.8的put和get實現以前要知道一些常量設計和數據結構，這些是構成ConcurrentHashMap實現結構的基礎，下面看一下基本屬性：
// node數組最大容量：2^30=1073741824 private static final int MAXIMUM_CAPACITY = 1 << 30; // 默認初始值，必須是2的幕數 private static final int DEFAULT_CAPACITY = 16; //數組可能最大值，須要與toArray（）相關方法關聯 static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8; //併發級別，遺留下來的，爲兼容之前的版本 private static final int DEFAULT_CONCURRENCY_LEVEL = 16; // 負載因子 private static final float LOAD_FACTOR = 0.75f; // 鏈表轉紅黑樹閥值,> 8 鏈表轉換爲紅黑樹 static final int TREEIFY_THRESHOLD = 8; //樹轉鏈表閥值，小於等於6（tranfer時，lc、hc=0兩個計數器分別++記錄原bin、新binTreeNode數量，<=UNTREEIFY_THRESHOLD 則untreeify(lo)） 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; // 2^15-1，help resize的最大線程數 private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1; // 32-16=16，sizeCtl中記錄size大小的偏移量 private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS; // forwarding nodes的hash值 static final int MOVED = -1; // 樹根節點的hash值 static final int TREEBIN = -2; // ReservationNode的hash值 static final int RESERVED = -3; // 可用處理器數量 static final int NCPU = Runtime.getRuntime().availableProcessors(); //存放node的數組 transient volatile Node<K,V>[] table; /*控制標識符，用來控制table的初始化和擴容的操做，不一樣的值有不一樣的含義 *當爲負數時：-1表明正在初始化，-N表明有N-1個線程正在 進行擴容 *當爲0時：表明當時的table尚未被初始化 *當爲正數時：表示初始化或者下一次進行擴容的大小*/ private transient volatile int sizeCtl;
基本屬性定義了ConcurrentHashMap的一些邊界以及操做時的一些控制，下面看一些內部的一些結構組成，這些是整個ConcurrentHashMap整個數據結構的核心
-
put操做
public V put(K key, V value) {
return putVal(key, value, false);
}
/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode()); //兩次hash，減小hash衝突，能夠均勻分佈
int binCount = 0;
for (Node<K,V>[] tab = table;;) { //對這個table進行迭代
Node<K,V> f; int n, i, fh;
//這裏就是上面構造方法沒有進行初始化，在這裏進行判斷，爲null就調用initTable進行初始化，屬於懶漢模式初始化
if (tab == null || (n = tab.length) == 0)
tab = initTable();
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {//若是i位置沒有數據，就直接無鎖插入
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)//若是在進行擴容，則先進行擴容操做
tab = helpTransfer(tab, f);
else {
V oldVal = null;
//若是以上條件都不知足，那就要進行加鎖操做，也就是存在hash衝突，鎖住鏈表或者紅黑樹的頭結點
synchronized (f) {
if (tabAt(tab, i) == f) {
if (fh >= 0) { //表示該節點是鏈表結構
binCount = 1;
for (Node<K,V> e = f;; ++binCount) {
K ek;
//這裏涉及到相同的key進行put就會覆蓋原先的value
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) { //若是鏈表的長度大於8時就會進行紅黑樹的轉換
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);//統計size，而且檢查是否須要擴容
return null;
}
這個put的過程很清晰，對當前的table進行無條件自循環直到put成功，能夠分紅如下六步流程來概述
- 若是沒有初始化就先調用initTable（）方法來進行初始化過程
- 若是沒有hash衝突就直接CAS插入
- 若是還在進行擴容操做就先進行擴容
- 若是存在hash衝突，就加鎖來保證線程安全，這裏有兩種狀況，一種是鏈表形式就直接遍歷到尾端插入，一種是紅黑樹就按照紅黑樹結構插入，
- 最後一個若是該鏈表的數量大於閾值8，就要先轉換成黑紅樹的結構，break再一次進入循環
- 若是添加成功就調用addCount（）方法統計size，而且檢查是否須要擴容
如今咱們來對每一步的細節進行源碼分析，在第一步中，符合條件會進行初始化操做，咱們來看看initTable（）方法
/**
* Initializes table, using the size recorded in sizeCtl.
*/
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {//空的table才能進入初始化操做
if ((sc = sizeCtl) < 0) //sizeCtl<0表示其餘線程已經在初始化了或者擴容了，掛起當前線程
Thread.yield(); // lost initialization race; just spin
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {//CAS操做SIZECTL爲-1，表示初始化狀態
try {
if ((tab = table) == null || tab.length == 0) {
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];//初始化
table = tab = nt;
sc = n - (n >>> 2);//記錄下次擴容的大小
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}
若是相應位置的Node還未初始化，則經過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; // no lock when adding to empty bin
}
/********************************/
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);
}
在第二步中沒有hash衝突就直接調用Unsafe的方法CAS插入該元素，進入第三步若是容器正在擴容，則會調用helpTransfer（）方法幫助擴容，如今咱們跟進helpTransfer（）方法看看
/**
* Helps transfer if a resize is in progress.
* 幫助從舊的table的元素複製到新的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) { //新的table nextTba已經存在前提下才能幫助擴容
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;
}
其實helpTransfer（）方法的目的就是調用多個工做線程一塊兒幫助進行擴容，這樣的效率就會更高，而不是隻有檢查到要擴容的那個線程進行擴容操做，其餘線程就要等待擴容操做完成才能工做
既然這裏涉及到擴容的操做，咱們也一塊兒來看看擴容方法transfer（）
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
// 每核處理的量小於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>[] nt = (Node<K,V>[])new Node<?,?>[n << 1]; //構建一個nextTable對象，其容量爲原來容量的兩倍
nextTab = nt;
} catch (Throwable ex) { // try to cope with OOME
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;
transferIndex = n;
}
int nextn = nextTab.length;
// 鏈接點指針，用於標誌位（fwd的hash值爲-1，fwd.nextTable=nextTab）
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
// 當advance == true時，代表該節點已經處理過了
boolean advance = true;
boolean finishing = false; // to ensure sweep before committing nextTab
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
// 控制 --i ,遍歷原hash表中的節點
while (advance) {
int nextIndex, nextBound;
if (--i >= bound || finishing)
advance = false;
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
// 用CAS計算獲得的transferIndex
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; // table 指向nextTable
sizeCtl = (n << 1) - (n >>> 1); // sizeCtl閾值爲原來的1.5倍
return; // 跳出死循環，
}
// CAS 更擴容閾值，在這裏面sizectl值減一，說明新加入一個線程參與到擴容操做
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
}
}
// 遍歷的節點爲null，則放入到ForwardingNode 指針節點
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
// f.hash == -1 表示遍歷到了ForwardingNode節點，意味着該節點已經處理過了
// 這裏是控制併發擴容的核心
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
else {
// 節點加鎖
synchronized (f) {
// 節點複製工做
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;
// fh >= 0 ,表示爲鏈表節點
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);
}
// 在nextTable i 位置處插上鍊表
setTabAt(nextTab, i, ln);
// 在nextTable i + n 位置處插上鍊表
setTabAt(nextTab, i + n, hn);
// 在table i 位置處插上ForwardingNode 表示該節點已經處理過了
setTabAt(tab, i, fwd);
// advance = true 能夠執行--i動做，遍歷節點
advance = true;
}
// 若是是TreeBin，則按照紅黑樹進行處理，處理邏輯與上面一致
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;
}
}
// 擴容後樹節點個數若<=6，將樹轉鏈表
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;
}
}
}
}
}
}
擴容過程有點複雜，這裏主要涉及到多線程併發擴容,ForwardingNode的做用就是支持擴容操做，將已處理的節點和空節點置爲ForwardingNode，併發處理時多個線程通過ForwardingNode就表示已經遍歷了，就日後遍歷，下圖是多線程合做擴容的過程：
介紹完擴容過程，咱們再次回到put流程，在第四步中是向鏈表或者紅黑樹里加節點，到第五步，會調用treeifyBin（）方法進行鏈表轉紅黑樹的過程
private final void treeifyBin(Node<K,V>[] tab, int index) {
Node<K,V> b; int n, sc;
if (tab != null) {
//若是整個table的數量小於64，就擴容至原來的一倍，不轉紅黑樹了
//由於這個閾值擴容能夠減小hash衝突，沒必要要去轉紅黑樹
if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
tryPresize(n << 1);
else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
synchronized (b) {
if (tabAt(tab, index) == b) {
TreeNode<K,V> hd = null, tl = null;
for (Node<K,V> e = b; e != null; e = e.next) {
//封裝成TreeNode
TreeNode<K,V> p =
new TreeNode<K,V>(e.hash, e.key, e.val,
null, null);
if ((p.prev = tl) == null)
hd = p;
else
tl.next = p;
tl = p;
}
//經過TreeBin對象對TreeNode轉換成紅黑樹
setTabAt(tab, index, new TreeBin<K,V>(hd));
}
}
}
}
}
到第六步表示已經數據加入成功了，如今調用addCount()方法計算ConcurrentHashMap的size，在原來的基礎上加一，如今來看看addCount()方法
private final void addCount(long x, int check) {
CounterCell[] as; long b, s;
//更新baseCount，table的數量，counterCells表示元素個數的變化
if ((as = counterCells) != null ||
!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
CounterCell a; long v; int m;
boolean uncontended = true;
//若是多個線程都在執行，則CAS失敗，執行fullAddCount，所有加入count
if (as == null || (m = as.length - 1) < 0 ||
(a = as[ThreadLocalRandom.getProbe() & m]) == null ||
!(uncontended =
U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
fullAddCount(x, uncontended);
return;
}
if (check <= 1)
return;
s = sumCount();
}
//check>=0表示須要進行擴容操做
if (check >= 0) {
Node<K,V>[] tab, nt; int n, sc;
while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
(n = tab.length) < MAXIMUM_CAPACITY) {
int rs = resizeStamp(n);
if (sc < 0) {
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
transferIndex <= 0)
break;
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
transfer(tab, nt);
}
//當前線程發起庫哦哦讓操做，nextTable=null
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
transfer(tab, null);
s = sumCount();
}
}
}
put的流程如今已經分析完了，你能夠從中發現，他在併發處理中使用的是樂觀鎖，當有衝突的時候才進行併發處理，並且流程步驟很清晰，可是細節設計的很複雜，畢竟多線程的場景也複雜
-
get操做
咱們如今要回到開始的例子中，咱們對我的信息進行了新增以後，咱們要獲取所新增的信息，使用String name = map.get(「name」)獲取新增的name信息，如今咱們依舊用debug的方式來分析下ConcurrentHashMap的獲取方法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()); //計算hash,再散列
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {//讀取首節點的Node元素
if ((eh = e.hash) == h) { //若是該節點就是首節點就返回
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
//hash值爲負值表示正在擴容，這個時候查的是ForwardingNode的find方法來定位到nextTable來
//查找，查找到就返回
else if (eh < 0)
return (p = e.find(h, key)) != null ? p.val : null;
while ((e = e.next) != null) {//既不是首節點也不是ForwardingNode，那就往下遍歷
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
}
ConcurrentHashMap的get操做的流程很簡單，也很清晰，能夠分爲三個步驟來描述
- 計算hash值，定位到該table索引位置，若是是首節點符合就返回
- 若是遇到擴容的時候，會調用標誌正在擴容節點ForwardingNode的find方法，查找該節點，匹配就返回
- 以上都不符合的話，就往下遍歷節點，匹配就返回，不然最後就返回null
-
size操做
最後咱們來看下例子中最後獲取size的方式int size = map.size();，如今讓咱們看下size()方法
public int size() {
long n = sumCount();
return ((n < 0L) ? 0 :
(n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
(int)n);
}
final long sumCount() {
CounterCell[] as = counterCells; CounterCell a; //變化的數量
long sum = baseCount;
if (as != null) {
for (int i = 0; i < as.length; ++i) {
if ((a = as[i]) != null)
sum += a.value;
}
}
return sum;
}
在JDK1.8版本中，對於size的計算，在擴容和addCount()方法就已經有處理了，JDK1.7是在調用size()方法纔去計算，其實在併發集合中去計算size是沒有多大的意義的，由於size是實時在變的，只能計算某一刻的大小，可是某一刻太快了，人的感知是一個時間段，因此並非很精確。
5.總結
- JDK1.8取消了segment數組，直接用table保存數據，鎖的粒度更小，減小併發衝突的機率。
- JDK1.8存儲數據時採用了鏈表+紅黑樹的形式，純鏈表的形式時間複雜度爲O(n)，紅黑樹則爲O（logn），性能提高很大。何時鏈表轉紅黑樹？當key值相等的元素造成的鏈表中元素個數超過8個的時候。
- JDK1.8的實現下降鎖的粒度，JDK1.7版本鎖的粒度是基於Segment的，包含多個HashEntry，而JDK1.8鎖的粒度就是HashEntry（首節點）
- JDK1.8版本的數據結構變得更加簡單，使得操做也更加清晰流暢，由於已經使用synchronized來進行同步，因此不須要分段鎖的概念，也就不須要Segment這種數據結構了，因爲粒度的下降，實現的複雜度也增長了
- JDK1.8使用紅黑樹來優化鏈表，基於長度很長的鏈表的遍歷是一個很漫長的過程，而紅黑樹的遍歷效率是很快的，代替必定閾值的鏈表，這樣造成一個最佳拍檔
- JDK1.8爲何使用內置鎖synchronized來代替重入鎖ReentrantLock，我以爲有如下幾點
- 由於粒度下降了，在相對而言的低粒度加鎖方式，synchronized並不比ReentrantLock差，在粗粒度加鎖中ReentrantLock可能經過Condition來控制各個低粒度的邊界，更加的靈活，而在低粒度中，Condition的優點就沒有了
- JVM的開發團隊歷來都沒有放棄synchronized，並且基於JVM的synchronized優化空間更大，使用內嵌的關鍵字比使用API更加天然
- 在大量的數據操做下，對於JVM的內存壓力，基於API的ReentrantLock會開銷更多的內存，雖然不是瓶頸，可是也是一個選擇依據
- 1. 談談ConcurrentHashMap1.7和1.8的不一樣實現
- 2. 【java基礎】ConcurrentHashMap1.7和1.8的不一樣實現
- 3. ConcurrentHashMap1.7和1.8的不同實現
- 4. ConcurrentHashMap1.7和1.8設計思想
- 5. JDK 1.8 sun.misc.Unsafe類CAS底層實現
- 6. hashMap （1.8） 底層實現原理
- 7. ConcurrentHashMap 底層具體實現JDK1.7、1.8
- 8. ConcurrentHashMap1.7和1.8解讀於理解
- 9. HashMap的底層實現(1.8前後的區別)
- 10. 解讀ConcurrentHashMap實現和應用 JDK1.7和1.8之間的不一樣
- 更多相關文章...
- • 現實生活中的 XML - XML 教程
- • ionic 頭部和底部 - ionic 教程
- • ☆基於Java Instrument的Agent實現
- • TiDB 在摩拜單車在線數據業務的應用和實踐
-
每一个你不满意的现在，都有一个你没有努力的曾经。
- 1. js中 charCodeAt
- 2. Android中通過ViewHelper.setTranslationY實現View移動控制（NineOldAndroids開源項目）
- 3. 【Android】日常記錄：BottomNavigationView自定義樣式，修改點擊後圖片
- 4. maya 文件檢查 ui和數據分離 （一）
- 5. eclipse 修改項目的jdk版本
- 6. Android InputMethod設置
- 7. Simulink中Bus Selector出現很多? ? ?
- 8. 【Openfire筆記】啓動Mac版Openfire時提示「系統偏好設置錯誤」
- 9. AutoPLP在偏好標籤中的生產與應用
- 10. 數據庫關閉的四種方式
- 1. 談談ConcurrentHashMap1.7和1.8的不一樣實現
- 2. 【java基礎】ConcurrentHashMap1.7和1.8的不一樣實現
- 3. ConcurrentHashMap1.7和1.8的不同實現
- 4. ConcurrentHashMap1.7和1.8設計思想
- 5. JDK 1.8 sun.misc.Unsafe類CAS底層實現
- 6. hashMap （1.8） 底層實現原理
- 7. ConcurrentHashMap 底層具體實現JDK1.7、1.8
- 8. ConcurrentHashMap1.7和1.8解讀於理解
- 9. HashMap的底層實現(1.8前後的區別)
- 10. 解讀ConcurrentHashMap實現和應用 JDK1.7和1.8之間的不一樣