併發編程—6ConcurrentHashMap1.7 & 1.8
[TOC]前端
6 ConcurrentHashMap jdk1.7
- hash算法的介紹
- 構造方法作了什麼
- get方法作了什麼
- put方法作了什麼
- 動態擴容邏輯
6.1 預備知識
hash算法: 就是把任意長度的輸入，經過散列算法，變換成固定長度的輸出，該輸出就是散列值。這種轉換是一種壓縮映射，也就是，散列值的空間一般遠小於輸入的空間，不一樣的輸入可能會散列成相同的輸出，因此不可能從散列值來惟一的肯定輸入值。簡單的說就是一種將任意長度的消息壓縮到某一固定長度的消息摘要的函數。在jdk1.7爲了使得hash出來的值更加均勻，在concurrentHashMap裏面使用了Wang/Jenkins hash算法再作了一次hash。 HashMap和HashTable的缺陷： Hashmap多線程會致使HashMap的Entry鏈表造成環形數據結構，一旦造成環形數據結構，Entry的next節點永遠不爲空，就會產生死循環獲取Entry。 HashTable使用synchronized來保證線程安全，但在線程競爭激烈的狀況下HashTable的效率很是低下。由於當一個線程訪問HashTable的同步方法，其餘線程也訪問HashTable的同步方法時，會進入阻塞或輪詢狀態。如線程1使用put進行元素添加，線程2不但不能使用put方法添加元素，也不能使用get方法來獲取元素，因此競爭越激烈效率越低。 putIfAbsent() ：沒有這個值則放入map，有這個值則返回key原本對應的值。
6.2 jdk1.7原理和實現
ConcurrentHashMap是由Segment數組結構和HashEntry數組結構組成。Segment實際繼承自可重入鎖（ReentrantLock），在ConcurrentHashMap裏扮演鎖的角色；HashEntry則用於存儲鍵值對數據。一個ConcurrentHashMap裏包含一個Segment數組，每一個Segment裏包含一個HashEntry數組，咱們稱之爲table，每一個HashEntry是一個鏈表結構的元素。node
面試常問： ConcurrentHashMap實現原理是怎麼樣的或者問ConcurrentHashMap如何在保證高併發下線程安全的同時實現了性能提高？ 答：ConcurrentHashMap容許多個修改操做併發進行，其關鍵在於使用了鎖分離技術。它使用了多個鎖來控制對hash表的不一樣部分進行的修改。內部使用段(Segment)來表示這些不一樣的部分，每一個段其實就是一個小的hash table，只要多個修改操做發生在不一樣的段上，它們就能夠併發進行。
6.3 源碼
6.3.1 構造方法
initialCapacity：初始容量大小 ，默認16。
loadFactor, 擴容因子，默認0.75，當一個Segment存儲的元素數量大於initialCapacity* loadFactor時，該Segment會進行一次擴容。
concurrencyLevel 併發度，默認16。併發度能夠理解爲程序運行時可以同時更新ConccurentHashMap且不產生鎖競爭的最大線程數，實際上就是ConcurrentHashMap中的分段鎖個數，即Segment[]的數組長度。若是併發度設置的太小，會帶來嚴重的鎖競爭問題；若是併發度設置的過大，本來位於同一個Segment內的訪問會擴散到不一樣的Segment中，CPU cache命中率會降低，從而引發程序性能降低。
public ConcurrentHashMap() {
this(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL);
}
//點擊this進去
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {
//參數校驗
if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
if (concurrencyLevel > MAX_SEGMENTS)
concurrencyLevel = MAX_SEGMENTS;
//設置ssize的值，ssize是segments數組的大小，這裏取的是大於等於concurrencyLevel的2^n的一個值。
int sshift = 0;
int ssize = 1;
while (ssize < concurrencyLevel) {
++sshift;
ssize <<= 1;
}
//設置segmentShift和segmentMask的值
//sshift就是上面描述中的n值，默認狀況下concurrencyLevel等於16，sshift就等於4
//所以默認狀況下segmentShift的值就是28，這個值會參與hash運算。
//segmentMask是hash運算的掩碼，默認狀況下等於16-1=15，相似於網絡中的子網掩碼，segmentMask的二進制最後幾位都是1，最大值是末尾16個1（65535）。
this.segmentShift = 32 - sshift;
this.segmentMask = ssize - 1;
if (initialCapacity > MAXIMUM_CAPACITY)
initialCapacity = MAXIMUM_CAPACITY;
int c = initialCapacity / ssize;
if (c * ssize < initialCapacity)
++c;
//初始化segment，其中cap是segment裏面的HashEntry數組的長度。它取的是大於等於c（Map容量/ssize）的2^N的一個值
int cap = MIN_SEGMENT_TABLE_CAPACITY;
while (cap < c)
cap <<= 1;
//建立segments和segments[0]（這裏面只初始化了一個segments數組中的第0個元素）。
Segment<K,V> s0 =
new Segment<K,V>(loadFactor, (int)(cap * loadFactor),
(HashEntry<K,V>[])new HashEntry[cap]);
Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];
UNSAFE.putOrderedObject(ss, SBASE, s0);
this.segments = ss;
}
執行流程小結：：
--初始化currentHashMap時，初始化map的容量，負載因子，併發等級等信息
--默認會在map裏面新建一個segment數組，而且會只初始化segments數組中的第0個元素。
6.2.2 get方法
get操做是先定位到segment，而後再到segment中去獲取對應的value值 定位segment和定位table後，依次掃描這個table元素下的的鏈表，要麼找到元素，要麼返回null。 在高併發下的狀況下如何保證取得的元素是最新的？ 答：用於存儲鍵值對數據的HashEntry，在設計上它的成員變量value等都是volatile類型的，這樣就保證別的線程對value值的修改，get方法能夠立刻看到。面試
public V get(Object key) {
Segment<K,V> s;
HashEntry<K,V>[] tab;
int h = hash(key);
// 根據Segment的索引（(h >>> segmentShift) & segmentMask）算出在Segment[]上的偏移量
// 默認狀況下segmentShift爲28，segmentMask爲15（低位有1111），從而能夠獲得h的高四位的值。做爲segment數組的索引。
long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
(tab = s.table) != null) {
//若segment存在則繼續查找segment上面的table[]的索引位置
//根據table的索引（(tab.length - 1) & h）算出在table[]上的偏移量，循環鏈表找出結果
for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile
(tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);
e != null; e = e.next) {
K k;
//對象相等
//或者
//hash相等且key相等時斷定同一個key。返回value
if ((k = e.key) == key || (e.hash == h && key.equals(k)))
return e.value;
}
}
return null;
}
執行流程小結：
-- 1。定位segment的位置，經過segment的索引（h >>> segmentShift) & segmentMask）即經過hash的高位算出Segment[]上的偏移量。
-- 2。 根據table的索引((tab.length - 1) & h)算出table[]上的偏移量。
-- 3。 循環HashEntry鏈表直到找到結果。
-- 4。由於沒有加鎖若是在get的時候，同時有線程修改了hashEntry的值可能會出現獲取不到真實的值。出現弱一致性的問題。
6.2.3 put方法
對於put()操做，前面的定位Segment的操做都是和put()相同的。找到Segment之後，而後對整個Segment加鎖，而後再進行後續的操做算法
1。put方法
public V put(K key, V value) {
Segment<K,V> s;
//1.校驗
if (value == null)
throw new NullPointerException();
int hash = hash(key);
// 2. 定位Segment，並判斷其是否存在
int j = (hash >>> segmentShift) & segmentMask;
if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck
(segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment
s = ensureSegment(j);//若是Segment不存在，則新建。
return s.put(key, hash, value, false);// 若是Segment存在，提交給Segment去處理。調用Segment.put方法
}
執行流程小結：
-- a 根據key算出的hash值，肯定是屬於哪一個segement。斷定segment是否存在。
-- b 若是對應的腳標segement存在，則提交Segment.put()去處理。
-- c 若是對應的腳標未存在，新建一個，經過cas操做設置到segement數組中去。
//2 ensureSegment 分析
@SuppressWarnings("unchecked")
private Segment<K,V> ensureSegment(int k) {
final Segment<K,V>[] ss = this.segments;
long u = (k << SSHIFT) + SBASE; // segment數組的偏移量
Segment<K,V> seg;
if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {
Segment<K,V> proto = ss[0]; // 使用 segment 0 做爲原型，能夠省略一些參數的計算
int cap = proto.table.length;
float lf = proto.loadFactor;
int threshold = (int)(cap * lf);
HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];
if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
== null) { // recheck
Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);//建立Segment。使用cas建立直到成功
while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
== null) {
if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
break;
}
}
}
return seg;
}
執行流程小結：
-- 根據segment數組的偏移量去獲取看是否返回非null元素
-- 若是seg非null，直接返回
-- 若是seg爲null，則使用segments數組第0個元素做爲原型建立一個新的segment數組元素。並使用cas操做設置到segments數組
//3 Segment.put()方法
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
//嘗試加鎖。首先嚐試tryLock()，屢次失敗之後使用lock()；同時會查找HashEntry，若是沒有找到，建立並返回一個
HashEntry<K,V> node = tryLock() ? null :
scanAndLockForPut(key, hash, value);
V oldValue;
try {
HashEntry<K,V>[] tab = table;
int index = (tab.length - 1) & hash;
HashEntry<K,V> first = entryAt(tab, index);
// 循環定位鏈表中的HashEntry位置，而後執行變動
for (HashEntry<K,V> e = first;;) {
if (e != null) {// 查找key，若找到直接修改value值。若是找不到繼續找下一個節點e.next
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若是不等於null，說明以前已經預熱完成，能夠直接插入
node.setNext(first);
else //
node = new HashEntry<K,V>(hash, key, value, first);
int c = count + 1;
//threshold 通常等於 (int)(capacity *loadFactor)，默認是16 * 0.75 = 12。當table裏面的鏈表長度大於12時，須要動態擴容。rehash()
if (c > threshold && tab.length < MAXIMUM_CAPACITY)
rehash(node);
else
setEntryAt(tab, index, node);
++modCount;
count = c;
oldValue = null;
break;
}
}
} finally {
unlock();
}
return oldValue;
}
執行流程小結：
-- a 由於put是修改操做，這裏須要加鎖。會先嚐試加鎖，若是加鎖次數過多，則直接lock
-- b 先根據(tab.length - 1) & hash定位在table裏面的位置
-- c 循環鏈表，若是存在相同的key值，就修改oldvalue，放回oldevalue。
-- c 若是不存在相同的key值，則設置。
// 4.scanAndLockForPut分析
private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
//經過給定的segment和hash，定位table裏hashentry的位置
HashEntry<K,V> first = entryForHash(this, hash);
HashEntry<K,V> e = first;
HashEntry<K,V> node = null;
int retries = -1; // 定位node時爲負值
while (!tryLock()) {// 這裏首先嚐試使用tryLock()，達到最大重試次數MAX_SCAN_RETRIES(2次或者65次)後，轉爲lock()的阻塞操做
HashEntry<K,V> f; // to recheck first below
if (retries < 0) {
if (e == null) { //定位node的位置，若找不到則建立一個HashEntry。代表table裏面一個HaseEntry都沒有
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) {//MAX_SCAN_RETRIES 1 or 64
lock();
break;
}
else if ((retries & 1) == 0 &&
(f = entryForHash(this, hash)) != first) {
// 若是加鎖過程當中，node有新增，則從新遍歷鏈表，（這裏能夠解釋對於鏈表的插入位置老是head的問題了）
e = first = f; // re-traverse if entry changed
retries = -1;
}
}
return node;
}
執行流程小結：
-- a 嘗試tryLock().達到最大重試次數MAX_SCAN_RETRIES後，轉爲lock()的阻塞操做.
-- b 經過給定的segment和hash，定位table的位置。返回第一個entry。
若是第一entry爲null，則新建一個HashEntry。
若是第一個entry不爲null。
4。size方法
ConcurrentHashMap的size()操做須要統計全部的Segment中的HashEntry數量，最大爲Integer.MAX_VALUE。由於在統計個過程當中，有可能出現多線程修改的問題。即使如此，ConcurrentHashMap首先會用無鎖嘗試3次，若是統計失敗，再加鎖統計。代碼以下
估計的大概數量，不是精確數量。
public int size() {
// Try a few times to get accurate count. On failure due to
// continuous async changes in table, resort to locking.
// 首先嚐試3次無鎖的統計，若是失敗，再進入加鎖統計
final Segment<K,V>[] segments = this.segments;
int size;
boolean overflow; // true if size overflows 32 bits
long sum; // sum of modCounts
long last = 0L; // previous sum
int retries = -1; // first iteration isn't retry
try {
for (;;) {
if (retries++ == RETRIES_BEFORE_LOCK) {// 加鎖統計
for (int j = 0; j < segments.length; ++j)
ensureSegment(j).lock(); // force creation
}
sum = 0L;
size = 0;
overflow = false;
// 嘗試3次無鎖統計，這裏面經過統計先後的modCount值的和 變化，這個值在每一個Segment中，每一次變動操做都會遞增，相似於Segment的版本號
for (int j = 0; j < segments.length; ++j) {
Segment<K,V> seg = segmentAt(segments, j);
if (seg != null) {
sum += seg.modCount;
int c = seg.count;
if (c < 0 || (size += c) < 0)
overflow = true;
}
}
if (sum == last)
break;
last = sum;
}
} finally {
if (retries > RETRIES_BEFORE_LOCK) {
for (int j = 0; j < segments.length; ++j)
segmentAt(segments, j).unlock();
}
}
return overflow ? Integer.MAX_VALUE : size;
}
5。動態擴容 transfer??
> rehash也就是擴容操做，擴容以後的容量是以前的兩倍，因此擴容以後的newCapacity也是2^n的一個值
private void rehash(HashEntry<K,V> node) {
/*
* Reclassify nodes in each list to new table. Because we
* are using power-of-two expansion, the elements from
* each bin must either stay at same index, or move with a
* power of two offset. We eliminate unnecessary node
* creation by catching cases where old nodes can be
* reused because their next fields won't change.
* Statistically, at the default threshold, only about
* one-sixth of them need cloning when a table
* doubles. The nodes they replace will be garbage
* collectable as soon as they are no longer referenced by
* any reader thread that may be in the midst of
* concurrently traversing table. Entry accesses use plain
* array indexing because they are followed by volatile
* table write.
*/
/*
* 將table中每一個節點從新分配到新的table中去。由於使用的是 *2的方式擴容,
* 每一個元素在table中的索引要麼爲i（不變）,要麼是i+oldCapacity。
* 如：擴容前容量是16，當前HashEntry在table[]中的索引爲3，則新的索引可能爲3或者19。
* 在節點拷貝的過程當中，有一些節點的next節點是不用調整的，就直接利用了。
* 據統計，在默認的threshold值時, 擴容只須要1/6的節點須要拷貝。
* 那些被替換掉的節點，在沒有任何線程引用的時候，將會被GC回收。
* Entry accesses use plain array indexing because they are followed by volatile table write.
*/
HashEntry<K,V>[] oldTable = table;
int oldCapacity = oldTable.length;
int newCapacity = oldCapacity << 1;
threshold = (int)(newCapacity * loadFactor);
HashEntry<K,V>[] newTable =
(HashEntry<K,V>[]) new HashEntry[newCapacity];
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 // 重複利用一些擴容後，next不變的節點，這些節點在原先鏈表的尾部
HashEntry<K,V> lastRun = e;
int lastIdx = idx;
for (HashEntry<K,V> last = next;
last != null;
last = last.next) {
int k = last.hash & sizeMask;
if (k != lastIdx) {
lastIdx = k;
lastRun = last;
}
}
newTable[lastIdx] = lastRun;
// Clone remaining nodes // 對於next變化的節點從新計算hash（鏈表前面部分節點），而後從新插入
for (HashEntry<K,V> p = e; p != lastRun; p = p.next) {
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; // add the new node // 將須要put的新node插
node.setNext(newTable[nodeIndex]);
newTable[nodeIndex] = node;
table = newTable;
}
-- 首先計算出newCapacity的容量；
-- 而後循環table[]，從新分配每條鏈表上面的元素。由於使用的是 *2的方式擴容，每一個元素在table中的索引要麼爲i（不變）,要麼是i+oldCapacity。如：擴容前容量是16，當前HashEntry在table[]中的索引爲3，則新的索引可能爲3或者19。
-- 拷貝過程當中，若是爲單鏈表則直接賦值；在節點拷貝的過程當中，有一些節點的next節點是不用調整的（鏈表後端部分片斷），就直接利用了；對於前端部分的片斷，則從新hash，而後插入到對應的鏈表中。
-- 最後再將須要put進來的node，在擴容後的結構中插入。
6.弱一致性問題
8.ConcurrentHashMap jdk1.8
8.1與1.7相比的重大變化
- 取消了segment數組，直接用table保存數據，鎖的粒度更小，減小併發衝突的機率。
- 存儲數據時採用了鏈表+紅黑樹的形式，純鏈表的形式時間複雜度爲O(n)，紅黑樹則爲O（logn），性能提高很大。何時鏈表轉紅黑樹？當key值相等的元素造成的鏈表中元素個數超過8個的時候。
8.2 主要數據結構和關鍵變量
Node類存放實際的key和value值。後端
8.3sizeCtl：
負數：表示進行初始化或者擴容,-1表示正在初始化，-N，表示有N-1個線程正在進行擴容 正數：0 表示尚未被初始化，>0的數，初始化或者是下一次進行擴容的閾值 TreeNode 用在紅黑樹，表示樹的節點, TreeBin是實際放在table數組中的，表明了這個紅黑樹的根。數組
8.4 源碼
8.4.1 初始化作了什麼事？
只是給成員變量賦值，put時進行實際數組的填充安全
8.4.2 在get和put操做中，是如何快速定位元素放在哪一個位置的？
get（）方法網絡
put()方法 數組的實際初始化數據結構
8.4.3 擴容操做
transfer()方法進行實際的擴容操做，table大小也是翻倍的形式，有一個併發擴容的機制。多線程
相關文章
- 1. Java併發編程的藝術(十三)——併發容器之ConcurrentHashMap(JDK 1.8版本)
- 2. 併發編程之 ConcurrentHashMap（JDK 1.8） putVal 源碼分析
- 3. Java併發——ConcurrentHashMap(JDK 1.8）
- 4. Scala併發編程
- 5. 併發編程
- 更多相關文章...
- • Rust 併發編程 - RUST 教程
- • XML 編輯器 - XML 教程
- • 算法總結-歸併排序
- • PHP開發工具
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
