當前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

并发编程（十六）——java7 深入并发包 ConcurrentHashMap 源码解析

發布時間：2023/12/19 编程问答 23 豆豆

生活随笔收集整理的這篇文章主要介紹了并发编程（十六）——java7 深入并发包 ConcurrentHashMap 源码解析小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

以前寫過介紹HashMap的文章，文中提到過HashMap在put的時候，插入的元素超過了容量（由負載因子決定）的范圍就會觸發擴容操作，就是rehash，這個會重新將原數組的內容重新hash到新的擴容數組中，在多線程的環境下，存在同時其他的元素也在進行put操作，如果hash值相同，可能出現同時在同一數組下用鏈表表示，造成閉環，導致在get時會出現死循環，所以HashMap是線程不安全的。

JDK1.7的實現

整個 ConcurrentHashMap 由一個個 Segment 組成，Segment 代表”部分“或”一段“的意思，所以很多地方都會將其描述為分段鎖。注意，行文中，我很多地方用了“槽”來代表一個 segment。

簡單理解就是，ConcurrentHashMap 是一個 Segment 數組，Segment 通過繼承 ReentrantLock 來進行加鎖，所以每次需要加鎖的操作鎖住的是一個 segment，這樣只要保證每個 Segment 是線程安全的，也就實現了全局的線程安全。

concurrencyLevel：并行級別、并發數、Segment 數。默認是 16，也就是說 ConcurrentHashMap 有 16 個 Segments，所以理論上，這個時候，最多可以同時支持 16 個線程并發寫，只要它們的操作分別分布在不同的 Segment 上。這個值可以在初始化的時候設置為其他值，但是一旦初始化以后，它是不可以擴容的。

再具體到每個 Segment 內部，其實每個 Segment 很像之前介紹的 HashMap，不過它要保證線程安全，所以處理起來要麻煩些。

初始化

initialCapacity：初始容量，這個值指的是整個 ConcurrentHashMap 的初始容量，實際操作的時候需要平均分給每個 Segment。

loadFactor：負載因子，之前我們說了，Segment 數組不可以擴容，所以這個負載因子是給每個 Segment 內部使用的。

1 public ConcurrentHashMap(int initialCapacity, 2 float loadFactor, int concurrencyLevel) { 3 if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0) 4 throw new IllegalArgumentException(); 5 if (concurrencyLevel > MAX_SEGMENTS) 6 concurrencyLevel = MAX_SEGMENTS; 7 // Find power-of-two sizes best matching arguments 8 int sshift = 0; 9 int ssize = 1; 10 // 計算并行級別 ssize，因為要保持并行級別是 2 的 n 次方 11 while (ssize < concurrencyLevel) { 12 ++sshift; 13 ssize <<= 1; 14 } 15 // 我們這里先不要那么燒腦，用默認值，concurrencyLevel 為 16，sshift 為 4 16 // 那么計算出 segmentShift 為 28，segmentMask 為 15，后面會用到這兩個值 17 this.segmentShift = 32 - sshift; 18 this.segmentMask = ssize - 1; 19 20 if (initialCapacity > MAXIMUM_CAPACITY) 21 initialCapacity = MAXIMUM_CAPACITY; 22 23 // initialCapacity 是設置整個 map 初始的大小， 24 // 這里根據 initialCapacity 計算 Segment 數組中每個位置可以分到的大小 25 // 如 initialCapacity 為 64，那么每個 Segment 或稱之為"槽"可以分到 4 個 26 int c = initialCapacity / ssize; 27 if (c * ssize < initialCapacity) 28 ++c; 29 // 默認 MIN_SEGMENT_TABLE_CAPACITY 是 2，這個值也是有講究的，因為這樣的話，對于具體的槽上， 30 // 插入一個元素不至于擴容，插入第二個的時候才會擴容 31 int cap = MIN_SEGMENT_TABLE_CAPACITY; 32 while (cap < c) 33 cap <<= 1; 34 35 // 創建 Segment 數組， 36 // 并創建數組的第一個元素 segment[0] 37 Segment<K,V> s0 = 38 new Segment<K,V>(loadFactor, (int)(cap * loadFactor), 39 (HashEntry<K,V>[])new HashEntry[cap]); 40 Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize]; 41 // 往數組寫入 segment[0] 42 UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0] 43 this.segments = ss; 44 }

初始化完成，我們得到了一個 Segment 數組。

我們就當是用 new ConcurrentHashMap() 無參構造函數進行初始化的，那么初始化完成后：

Segment 數組長度為 16，不可以擴容
Segment[i] 的默認大小為 2，負載因子是 0.75，得出初始閾值為 1.5，也就是以后插入第一個元素不會觸發擴容，插入第二個會進行第一次擴容
這里初始化了 segment[0]，其他位置還是 null，至于為什么要初始化 segment[0]，后面的代碼會介紹
當前 segmentShift 的值為 32 - 4 = 28，segmentMask 為 16 - 1 = 15，姑且把它們簡單翻譯為移位數和掩碼，這兩個值馬上就會用到

Segment?

1 static class Segment<K,V> extends ReentrantLock implements Serializable { 2 3 transient volatile HashEntry<K,V>[] table; 4 5 transient int count; 6 7 transient int modCount; 8 9 }

從上Segment的繼承體系可以看出，Segment實現了ReentrantLock,也就帶有鎖的功能,table使用volatile修飾，保證了內存可見性。

put 過程分析

我們先看 put 的主流程，對于其中的一些關鍵細節操作，后面會進行詳細介紹。

1 public V put(K key, V value) { 2 Segment<K,V> s; 3 if (value == null) 4 throw new NullPointerException(); 5 // 1. 計算 key 的 hash 值 6 int hash = hash(key); 7 // 2. 根據 hash 值找到 Segment 數組中的位置 j 8 // hash 是 32 位，無符號右移 segmentShift(28) 位，剩下高 4 位， 9 // 然后和 segmentMask(15) 做一次與操作，也就是說 j 是 hash 值的高 4 位，也就是槽的數組下標 10 int j = (hash >>> segmentShift) & segmentMask; 11 // 剛剛說了，初始化的時候初始化了 segment[0]，但是其他位置還是 null， 12 // ensureSegment(j) 對 segment[j] 進行初始化 13 if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck 14 (segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment 15 s = ensureSegment(j); 16 // 3. 插入新值到槽 s 中 17 return s.put(key, hash, value, false); 18 }

初始化槽: ensureSegment

ConcurrentHashMap 初始化的時候會初始化第一個槽 segment[0]，對于其他槽來說，在插入第一個值的時候進行初始化。

這里需要考慮并發，因為很可能會有多個線程同時進來初始化同一個槽 segment[k]，不過只要有一個成功了就可以。

1 private Segment<K,V> ensureSegment(int k) { 2 final Segment<K,V>[] ss = this.segments; 3 long u = (k << SSHIFT) + SBASE; // raw offset 4 Segment<K,V> seg; 5 if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) { 6 // 這里看到為什么之前要初始化 segment[0] 了， 7 // 使用當前 segment[0] 處的數組長度和負載因子來初始化 segment[k] 8 // 為什么要用“當前”，因為 segment[0] 可能早就擴容過了 9 Segment<K,V> proto = ss[0]; 10 int cap = proto.table.length; 11 float lf = proto.loadFactor; 12 int threshold = (int)(cap * lf); 13 14 // 初始化 segment[k] 內部的數組 15 HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap]; 16 if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) 17 == null) { // 再次檢查一遍該槽是否被其他線程初始化了。 18 19 Segment<K,V> s = new Segment<K,V>(lf, threshold, tab); 20 // 使用 while 循環，內部用 CAS，當前線程成功設值或其他線程成功設值后，退出,如果其他線程成功設置后，這里獲取到直接返回 21 while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) 22 == null) { 23 if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s)) 24 break; 25 } 26 } 27 } 28 return seg; 29 }

總的來說，ensureSegment(int k) 比較簡單，對于并發操作使用 CAS 進行控制。

第一層很簡單，根據 hash 值很快就能找到相應的 Segment，之后就是 Segment 內部的 put 操作了。

Segment 內部是由?數組+鏈表?組成的。

1 final V put(K key, int hash, V value, boolean onlyIfAbsent) { 2 // 在往該 segment 寫入前，需要先獲取該 segment 的獨占鎖 3 // 先看主流程，后面還會具體介紹這部分內容 4 HashEntry<K,V> node = tryLock() ? null : 5 scanAndLockForPut(key, hash, value); 6 V oldValue; 7 try { 8 // 這個是 segment 內部的數組 9 HashEntry<K,V>[] tab = table; 10 // 再利用 hash 值，求應該放置的數組下標 11 int index = (tab.length - 1) & hash; 12 // first 是數組該位置處的鏈表的表頭 13 HashEntry<K,V> first = entryAt(tab, index); 14 15 // 下面這串 for 循環雖然很長，不過也很好理解，想想該位置沒有任何元素和已經存在一個鏈表這兩種情況 16 for (HashEntry<K,V> e = first;;) { 17 if (e != null) { 18 K k; 19 if ((k = e.key) == key || 20 (e.hash == hash && key.equals(k))) { 21 oldValue = e.value; 22 if (!onlyIfAbsent) { 23 // 覆蓋舊值 24 e.value = value; 25 ++modCount; 26 } 27 break; 28 } 29 // 繼續順著鏈表走 30 e = e.next; 31 } 32 else { 33 // node 到底是不是 null，這個要看獲取鎖的過程，不過和這里都沒有關系。 34 // 如果不為 null，那就直接將它設置為鏈表表頭；如果是null，初始化并設置為鏈表表頭。 35 if (node != null) 36 node.setNext(first); 37 else 38 node = new HashEntry<K,V>(hash, key, value, first); 39 40 int c = count + 1; 41 // 如果超過了該 segment 的閾值，這個 segment 需要擴容 42 if (c > threshold && tab.length < MAXIMUM_CAPACITY) 43 rehash(node); // 擴容后面也會具體分析 44 else 45 // 沒有達到閾值，將 node 放到數組 tab 的 index 位置， 46 // 其實就是將新的節點設置成原鏈表的表頭 47 setEntryAt(tab, index, node); 48 ++modCount; 49 count = c; 50 oldValue = null; 51 break; 52 } 53 } 54 } finally { 55 // 解鎖 56 unlock(); 57 } 58 return oldValue; 59 }

整體流程還是比較簡單的，由于有獨占鎖的保護，所以 segment 內部的操作并不復雜。至于這里面的并發問題，我們稍后再進行介紹。

到這里 put 操作就結束了，接下來，我們說一說其中幾步關鍵的操作。

獲取寫入鎖: scanAndLockForPut

前面我們看到，在往某個 segment 中 put 的時候，首先會調用 node = tryLock() ? null : scanAndLockForPut(key, hash, value)，也就是說先進行一次 tryLock() 快速獲取該 segment 的獨占鎖，如果失敗，那么進入到 scanAndLockForPut 這個方法來獲取鎖。

下面我們來具體分析這個方法中是怎么控制加鎖的。

1 private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) { 2 HashEntry<K,V> first = entryForHash(this, hash); 3 HashEntry<K,V> e = first; 4 HashEntry<K,V> node = null; 5 int retries = -1; // negative while locating node 6 7 // 循環獲取鎖 8 while (!tryLock()) { 9 HashEntry<K,V> f; // to recheck first below 10 if (retries < 0) { 11 if (e == null) { 12 if (node == null) // speculatively create node 13 // 進到這里說明數組該位置的鏈表是空的，沒有任何元素 14 // 當然，進到這里的另一個原因是 tryLock() 失敗，所以該槽存在并發，不一定是該位置 15 node = new HashEntry<K,V>(hash, key, value, null); 16 retries = 0; 17 } 18 else if (key.equals(e.key)) 19 retries = 0; 20 else 21 // 順著鏈表往下走 22 e = e.next; 23 } 24 // 重試次數如果超過 MAX_SCAN_RETRIES（單核1多核64），那么不搶了，進入到阻塞隊列等待鎖 25 // lock() 是阻塞方法，直到獲取鎖后返回 26 else if (++retries > MAX_SCAN_RETRIES) { 27 lock(); 28 break; 29 } 30 else if ((retries & 1) == 0 && 31 // 這個時候是有大問題了，那就是有新的元素進到了鏈表，成為了新的表頭 32 // 所以這邊的策略是，相當于重新走一遍這個 scanAndLockForPut 方法 33 (f = entryForHash(this, hash)) != first) { 34 e = first = f; // re-traverse if entry changed 35 retries = -1; 36 } 37 } 38 return node; 39 }

這個方法有兩個出口，一個是 tryLock() 成功了，循環終止，另一個就是重試次數超過了 MAX_SCAN_RETRIES，進到 lock() 方法，此方法會阻塞等待，直到成功拿到獨占鎖。

這個方法就是看似復雜，但是其實就是做了一件事，那就是獲取該 segment 的獨占鎖，如果需要的話順便實例化了一下 node。

獲取鎖時，并不直接使用lock來獲取，因為該方法獲取鎖失敗時會掛起。事實上，它使用了自旋鎖，如果tryLock獲取鎖失敗，說明鎖被其它線程占用，此時通過循環再次以tryLock的方式申請鎖。如果在循環過程中該Key所對應的鏈表頭被修改，則重置retry次數。如果retry次數超過一定值，則使用lock方法申請鎖。

這里使用自旋鎖是因為自旋鎖的效率比較高，但是它消耗CPU資源比較多，因此在自旋次數超過閾值時切換為互斥鎖。

擴容: rehash

重復一下，segment 數組不能擴容，擴容是 segment 數組某個位置內部的數組 HashEntry\<k,v>[] 進行擴容，擴容后，容量為原來的 2 倍。

首先，我們要回顧一下觸發擴容的地方，put 的時候，如果判斷該值的插入會導致該 segment 的元素個數超過閾值，那么先進行擴容，再插值，讀者這個時候可以回去 put 方法看一眼。

該方法不需要考慮并發，因為到這里的時候，是持有該 segment 的獨占鎖的。

1 // 方法參數上的 node 是這次擴容后，需要添加到新的數組中的數據。 2 private void rehash(HashEntry<K,V> node) { 3 HashEntry<K,V>[] oldTable = table; 4 int oldCapacity = oldTable.length; 5 // 2 倍 6 int newCapacity = oldCapacity << 1; 7 threshold = (int)(newCapacity * loadFactor); 8 // 創建新數組 9 HashEntry<K,V>[] newTable = 10 (HashEntry<K,V>[]) new HashEntry[newCapacity]; 11 // 新的掩碼，如從 16 擴容到 32，那么 sizeMask 為 31，對應二進制 ‘000...00011111’ 12 int sizeMask = newCapacity - 1; 13 14 // 遍歷原數組，老套路，將原數組位置 i 處的鏈表拆分到新數組位置 i 和 i+oldCap 兩個位置 15 for (int i = 0; i < oldCapacity ; i++) { 16 // e 是鏈表的第一個元素 17 HashEntry<K,V> e = oldTable[i]; 18 if (e != null) { 19 HashEntry<K,V> next = e.next; 20 // 計算應該放置在新數組中的位置， 21 // 假設原數組長度為 16，e 在 oldTable[3] 處，那么 idx 只可能是 3 或者是 3 + 16 = 19 22 int idx = e.hash & sizeMask; 23 if (next == null) // 該位置處只有一個元素，那比較好辦 24 newTable[idx] = e; 25 else { // Reuse consecutive sequence at same slot 26 // e 是鏈表表頭 27 HashEntry<K,V> lastRun = e; 28 // idx 是當前鏈表的頭結點 e 的新位置 29 int lastIdx = idx; 30 31 // 下面這個 for 循環會找到一個 lastRun 節點，這個節點之后的所有元素是將要放到一起的 32 for (HashEntry<K,V> last = next; 33 last != null; 34 last = last.next) { 35 int k = last.hash & sizeMask; 36 if (k != lastIdx) { 37 lastIdx = k; 38 lastRun = last; 39 } 40 } 41 // 將 lastRun 及其之后的所有節點組成的這個鏈表放到 lastIdx 這個位置 42 newTable[lastIdx] = lastRun; 43 // 下面的操作是處理 lastRun 之前的節點， 44 // 這些節點可能分配在另一個鏈表中，也可能分配到上面的那個鏈表中 45 for (HashEntry<K,V> p = e; p != lastRun; p = p.next) { 46 V v = p.value; 47 int h = p.hash; 48 int k = h & sizeMask; 49 HashEntry<K,V> n = newTable[k]; 50 newTable[k] = new HashEntry<K,V>(h, p.key, v, n); 51 } 52 } 53 } 54 } 55 // 將新來的 node 放到新數組中剛剛的兩個鏈表之一的頭部 56 int nodeIndex = node.hash & sizeMask; // add the new node 57 node.setNext(newTable[nodeIndex]); 58 newTable[nodeIndex] = node; 59 table = newTable; 60 }

總結一下put的流程：

當執行put操作時，會進行第一次key的hash來定位Segment的位置，如果該Segment還沒有初始化，即通過CAS操作進行賦值，然后進行第二次hash操作，找到相應的HashEntry的位置，這里會利用繼承過來的鎖的特性，在將數據插入指定的HashEntry位置時（鏈表的尾端），會通過繼承ReentrantLock的tryLock（）方法嘗試去獲取鎖，如果獲取成功就直接插入相應的位置，如果已經有線程獲取該Segment的鎖，那當前線程會以自旋的方式去繼續的調用tryLock（）方法去獲取鎖，超過指定次數就掛起，等待喚醒。

get 過程分析

相對于 put 來說，get 真的不要太簡單。

計算 hash 值，找到 segment 數組中的具體位置，或我們前面用的“槽”

槽中也是一個數組，根據 hash 找到數組中具體的位置

到這里是鏈表了，順著鏈表進行查找即可

1 public V get(Object key) { 2 Segment<K,V> s; // manually integrate access methods to reduce overhead 3 HashEntry<K,V>[] tab; 4 // 1. hash 值 5 int h = hash(key); 6 long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE; 7 // 2. 根據 hash 找到對應的 segment 8 if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null && 9 (tab = s.table) != null) { 10 // 3. 找到segment 內部數組相應位置的鏈表，遍歷 11 for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile 12 (tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE); 13 e != null; e = e.next) { 14 K k; 15 if ((k = e.key) == key || (e.hash == h && key.equals(k))) 16 return e.value; 17 } 18 } 19 return null; 20 }

size操作

put、remove和get操作只需要關心一個Segment，而size操作需要遍歷所有的Segment才能算出整個Map的大小。一個簡單的方案是，先鎖住所有Sgment，計算完后再解鎖。但這樣做，在做size操作時，不僅無法對Map進行寫操作，同時也無法進行讀操作，不利于對Map的并行操作。

為更好支持并發操作，ConcurrentHashMap會在不上鎖的前提逐個Segment計算3次size，如果某相鄰兩次計算獲取的所有Segment的更新次數（每個Segment都與HashMap一樣通過modCount跟蹤自己的修改次數，Segment每修改一次其modCount加一）相等，說明這兩次計算過程中無更新操作，則這兩次計算出的總size相等，可直接作為最終結果返回。如果這三次計算過程中Map有更新，則對所有Segment加鎖重新計算Size。該計算方法代碼如下

1 public int size() { 2 final Segment<K,V>[] segments = this.segments; 3 int size; 4 boolean overflow; // true if size overflows 32 bits 5 long sum; // sum of modCounts 6 long last = 0L; // previous sum 7 int retries = -1; // first iteration isn't retry 8 try { 9 for (;;) { 10 if (retries++ == RETRIES_BEFORE_LOCK) { 11 for (int j = 0; j < segments.length; ++j) 12 ensureSegment(j).lock(); // force creation 13 } 14 sum = 0L; 15 size = 0; 16 overflow = false; 17 for (int j = 0; j < segments.length; ++j) { 18 Segment<K,V> seg = segmentAt(segments, j); 19 if (seg != null) { 20 sum += seg.modCount; 21 int c = seg.count; 22 if (c < 0 || (size += c) < 0) 23 overflow = true; 24 } 25 } 26 if (sum == last) 27 break; 28 last = sum; 29 } 30 } finally { 31 if (retries > RETRIES_BEFORE_LOCK) { 32 for (int j = 0; j < segments.length; ++j) 33 segmentAt(segments, j).unlock(); 34 } 35 } 36 return overflow ? Integer.MAX_VALUE : size; 37 }

ConcurrentHashMap的Size方法是一個嵌套循環，大體邏輯如下：

1.遍歷所有的Segment。

2.把Segment的元素數量累加起來。

3.把Segment的修改次數累加起來。

4.判斷所有Segment的總修改次數是否大于上一次的總修改次數。如果大于，說明統計過程中有修改，重新統計，嘗試次數+1；如果不是。說明沒有修改，統計結束。

5.如果嘗試次數超過閾值，則對每一個Segment加鎖，再重新統計。

6.再次判斷所有Segment的總修改次數是否大于上一次的總修改次數。由于已經加鎖，次數一定和上次相等。

7.釋放鎖，統計結束。

并發問題分析

現在我們已經說完了 put 過程和 get 過程，我們可以看到 get 過程中是沒有加鎖的，那自然我們就需要去考慮并發問題。

添加節點的操作 put 和刪除節點的操作 remove 都是要加 segment 上的獨占鎖的，所以它們之間自然不會有問題，我們需要考慮的問題就是 get 的時候在同一個 segment 中發生了 put 或 remove 操作。

put 操作的線程安全性。

初始化槽，這個我們之前就說過了，使用了 CAS 來初始化 Segment 中的數組。

添加節點到鏈表的操作是插入到表頭的，所以，如果這個時候 get 操作在鏈表遍歷的過程已經到了中間，是不會影響的。當然，另一個并發問題就是 get 操作在 put 之后，需要保證剛剛插入表頭的節點被讀取，這個依賴于 setEntryAt 方法中使用的 UNSAFE.putOrderedObject。

擴容。擴容是新創建了數組，然后進行遷移數據，最后面將 newTable 設置給屬性 table。所以，如果 get 操作此時也在進行，那么也沒關系，如果 get 先行，那么就是在舊的 table 上做查詢操作；而 put 先行，那么 put 操作的可見性保證就是 table 使用了 volatile 關鍵字。

remove 操作的線程安全性。

remove 操作我們沒有分析源碼，所以這里說的讀者感興趣的話還是需要到源碼中去求實一下的。

get 操作需要遍歷鏈表，但是 remove 操作會"破壞"鏈表。

如果 remove 破壞的節點 get 操作已經過去了，那么這里不存在任何問題。

如果 remove 先破壞了一個節點，分兩種情況考慮。 1、如果此節點是頭結點，那么需要將頭結點的 next 設置為數組該位置的元素，table 雖然使用了 volatile 修飾，但是 volatile 并不能提供數組內部操作的可見性保證，所以源碼中使用了 UNSAFE 來操作數組，請看方法 setEntryAt。2、如果要刪除的節點不是頭結點，它會將要刪除節點的后繼節點接到前驅節點中，這里的并發保證就是 next 屬性是 volatile 的。

總結

以上是生活随笔為你收集整理的并发编程（十六）——java7 深入并发包 ConcurrentHashMap 源码解析的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇： java简述垃圾回收原理及算法_Java
下一篇： dart 替代java_Dart与Jav