Java集合-ConcurrentHashMap工作原理和實現JDK7
概述
本文學習知識點
1.ConcurrentHashMap與HashMap的區別。
2.資料儲存結構。
3.如何提高併發讀寫效能。
4.put和get方法原始碼實現分析。
5.size方法如何實現。
與HashMap的區別
1.ConcurrentHashMap和HashMap都是Map的實現，提供key，value的讀寫。
2.都繼承自AbstractMap類，但實現的介面不同，如下圖：
ConcurrentHashMap與HashMap類圖
3.ConcurrentHashMap是執行緒安全的，HashMap非執行緒安全。
4.兩者採用的資料結構不同。ConcurrentHashMap採用segement+entry陣列的方式，HashMap採用Entry陣列的方式
資料結構
如下圖描述了ConcurrentHashMap的儲存結構，一目瞭然：
ConcurrentHashMap資料儲存結構
根據這個結構圖，我們再看看原始碼中的宣告：
public class ConcurrentHashMap<K, V> extends AbstractMap<K, V> implements ConcurrentMap<K, V>, Serializable { // 預設segment個數 static final int DEFAULT_INITIAL_CAPACITY = 16; // 預設的更新map的併發級別，即預設最多允許16個執行緒進行併發寫， // 這個級別與segment的個數一致 static final int DEFAULT_CONCURRENCY_LEVEL = 16; /** * The segments, each of which is a specialized hash table. */ final Segment<K,V>[] segments; // 其他省略 }
這裡的segments陣列中，每個segments其實是一個hash表，即HashEntry陣列，如下，採用內部類實現方式：
static final class Segment<K,V> extends ReentrantLock implements Serializable { // Hash table，預設初始容量為2，特別注意，這裡table用了volatile修飾， // 保證多執行緒讀寫時的可見性 transient volatile HashEntry<K,V>[] table; // segment中hashtable的元素數量計數器，用於size方法中，分段計算彙總 transient int count; // 執行更新操作時，獲取segment鎖的重試次數，多核CPU重試64次，單核CPU重試1次 static final int MAX_SCAN_RETRIES = Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1; // 其他省略 }
Segment類，繼承自ReentrantLock，因此每個Segment元素都是一個鎖，鎖的粒度從Hashtable類的整個表，降低到單個Segment元素，以支援分段鎖，提供更高的併發讀寫能力，再來看下segments及其內部HashEntry陣列怎麼初始化，以及初始化容量大小，構造方法如下：
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;
// Find power-of-two sizes best matching arguments
int sshift = 0;
int ssize = 1;// segments陣列的大小
// 根據併發級別算出segments的大小以及2的指數
while (ssize < concurrencyLevel) {
++sshift;
ssize <<= 1;
}
this.segmentShift = 32 - sshift;
this.segmentMask = ssize - 1;
if (initialCapacity > MAXIMUM_CAPACITY)
initialCapacity = MAXIMUM_CAPACITY;
int c = initialCapacity / ssize;
if (c * ssize < initialCapacity)
++c;
int cap = MIN_SEGMENT_TABLE_CAPACITY;// segment內部HashEntry的預設容量，為2
while (cap < c)
cap <<= 1;
// create segments and segments[0]，即建立segments陣列，
// 並初始化segments[0]元素的HashEntry陣列
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為sun.misc.Unsafe物件，使用CAS操作，
// 將segments[0]的元素替換為已經初始化的s0，保證原子性。
// Unsafe類採用C++語言實現，底層實現CPU的CAS指令操作，保證原子性。
UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
this.segments = ss;
}
從構造方法可以看出，ConcurrentHashMap初始化時，先根據併發級別數，計算出Segment陣列的大小ssize和每個Segment中HashEntry陣列的大小cap，並初始化Segment陣列的第一個元素；Segment陣列的大小為2的冪次方，預設為16，每個Segment內部HashEntry陣列大小也是2的冪次方，最小值為2，也是預設的初始大小。
put方法實現
public V put(K key, V value) {
Segment<K,V> s;
if (value == null)
throw new NullPointerException();
int hash = hash(key);
// 根據hash值、segmentShift,segmentMask(段的長度)計算定位到段的索引下標
int j = (hash >>> segmentShift) & segmentMask;
// (j << SSHIFT) + SBASE)這段程式碼也是為了定位段下標。
// 如果通過Unsafe類的CAS讀取段下標元素，元素沒有初始化，
// 則呼叫ensureSegment進行初始化
if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck
(segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment
s = ensureSegment(j);
return s.put(key, hash, value, false);
}
private int hash(Object k) {
int h = hashSeed;
if ((0 != h) && (k instanceof String)) {
return sun.misc.Hashing.stringHash32((String) k);
}
h ^= k.hashCode();
// Spread bits to regularize both segment and index locations,
// using variant of single-word Wang/Jenkins hash.
h += (h << 15) ^ 0xffffcd7d;
h ^= (h >>> 10);
h += (h << 3);
h ^= (h >>> 6);
h += (h << 2) + (h << 14);
return h ^ (h >>> 16);
}
put過程：
1.通過key的hashCode值再計算hash值。
2.通過hash值找到segments陣列的下標。
3.檢查下標segment是否已經初始化，如果沒有初始化，則呼叫ensureSegment進行初始化，內部用了CAS操作進行替換，達到初始化效果。初始化的過程進行了雙重檢查，UNSAFE.getObjectVolatile，通過這個方法執行了兩次，以檢查segment是否已經初始化，以及用UNSAFE.compareAndSwapObject進行CAS替換，CAS的替換有失敗的可能，因此原始碼中還加了自旋重試的操作，保證最終CAS操作的成功。ensureSegment的實現原始碼就不貼出來了，讀者自行看下。
4.再呼叫segment類的put方法，將元素放入HashEntry陣列中。Segment類的put方法是操作的核心，內部回有加鎖等機制。原始碼如下：
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
// 呼叫父類ReentrantLock的tryLock()方法嘗試鎖住segment，
// 獲取鎖失敗，則呼叫scanAndLockForPut方法自旋重試獲取鎖
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);
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;
}
當有執行緒A和執行緒B在相同segment物件上put物件時，執行過程如下：
1、執行緒A執行tryLock()方法獲取鎖。
2、執行緒B獲取鎖失敗，則執行scanAndLockForPut()方法，在scanAndLockForPut方法中，會通過重複執行tryLock()方法嘗試獲取鎖，在多處理器環境下，重複次數為64，單處理器重複次數為1，當執行tryLock()方法的次數超過上限時，則執行lock()方法掛起執行緒B；這樣設計目的是為了讓執行緒切換和自旋消耗的CPU的時間達到平衡，不至於白白浪費CPU，也不會過於平凡切換執行緒導致更多的CPU浪費。
3、獲得鎖之後，根據hash值定位到HashEntry陣列的下標，更新或插入元素，在插入過程中，如果HashEntry陣列元素個數容量超過負載比例，則進行rehash操作擴容，擴容為原來的兩倍（rehash請對比Java集合-HashMap原始碼實現深入解析中的邏輯，自行分析，基本一模一樣）；
4、在插入後，還會更新segment的count計數器，用於size方法中計算map元素個數時不用對每個segment內部HashEntry遍歷重新計算，提高效能。
5、當執行緒A執行完插入操作後，會通過unlock()方法釋放鎖，接著喚醒執行緒B繼續執行；
get方法
get方法比較簡單，值得注意的是，get方法不加鎖，並用Unsafe.getObjectVolatile方法讀取元素，這個方法保證讀取物件永遠是最新的
public V get(Object key) {
Segment<K,V> s; // manually integrate access methods to reduce overhead
HashEntry<K,V>[] tab;
int h = hash(key);
long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
(tab = s.table) != null) {
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;
if ((k = e.key) == key || (e.hash == h && key.equals(k)))
return e.value;
}
}
return null;
}
size方法實現
public int size() {
// Try a few times to get accurate count. On failure due to
// continuous async changes in table, resort to locking.
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;
// 彙總每個segment的count計數器
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 {
// 釋放所有segment上的鎖
if (retries > RETRIES_BEFORE_LOCK) {
for (int j = 0; j < segments.length; ++j)
segmentAt(segments, j).unlock();
}
}
return overflow ? Integer.MAX_VALUE : size;
}
由於ConcurrentHashMap支援併發讀寫，所以在準確計算元素時存在一定的難度，思路如下：
1、重複彙總計算segments元素個數，即將每個segment的計數器count累加彙總。
2、如果連續兩次計算的sum值相同，則結束計算，認為併發過程中計算的值正確，並返回。
3、如果重複三次，計算的結果都不相同，則強制鎖住全部segment後，重新計算值。
雖然採用重複計算和最後加鎖的方式再次計算，但size方法仍不能保證結果的準確性，例如，兩次計算結果相等，在返回之前，又有新的執行緒插入新值，則此時的結果就是不準確的。所以，在併發讀寫環境下，size方法進行精確計算的意義並不大，只能作為一個大概的計算結果，因此也決定了在使用size方法時，要評估清楚自己的需求是什麼，是精確計算，還是粗略計算。
作者：Misout
連結：https://www.jianshu.com/p/47c1be88a88e
來源：簡書
簡書著作權歸作者所有，任何形式的轉載都請聯絡作者獲得授權並註明出處。