文章推薦

• 精選java等全套學(xué)習(xí)資源
• 精選java電子圖書資源
• 精選大數(shù)據(jù)學(xué)習(xí)資源
• java項(xiàng)目練習(xí)精選

###Java7 ConcurrentHashMap

ConcurrentHashMap 和 HashMap 思路是差不多的，但是因?yàn)樗С植l(fā)操作，所以要復(fù)雜一些。

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

簡(jiǎn)單理解就是，ConcurrentHashMap 是一個(gè) Segment 數(shù)組，Segment 通過(guò)繼承 ReentrantLock 來(lái)進(jìn)行加鎖，所以每次需要加鎖的操作鎖住的是一個(gè) segment，這樣只要保證每個(gè) Segment 是線程安全的，也就實(shí)現(xiàn)了全局的線程安全。

**concurrencyLevel：**并行級(jí)別、并發(fā)數(shù)、Segment 數(shù)，怎么翻譯不重要，理解它。默認(rèn)是 16，也就是說(shuō) ConcurrentHashMap 有 16 個(gè) Segments，所以理論上，這個(gè)時(shí)候，最多可以同時(shí)支持 16 個(gè)線程并發(fā)寫，只要它們的操作分別分布在不同的 Segment 上。這個(gè)值可以在初始化的時(shí)候設(shè)置為其他值，但是一旦初始化以后，它是不可以擴(kuò)容的。

再具體到每個(gè) Segment 內(nèi)部，其實(shí)每個(gè) Segment 很像之前介紹的 HashMap，不過(guò)它要保證線程安全，所以處理起來(lái)要麻煩些。
##初始化

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

• loadFactor：負(fù)載因子，之前我們說(shuō)了，Segment 數(shù)組不可以擴(kuò)容，所以這個(gè)負(fù)載因子是給每個(gè) Segment 內(nèi)部使用的。

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 argumentsint sshift = 0;int ssize = 1;// 計(jì)算并行級(jí)別 ssize，因?yàn)橐３植⑿屑?jí)別是 2 的 n 次方while (ssize < concurrencyLevel) {++sshift;ssize <<= 1;}// 我們這里先不要那么燒腦，用默認(rèn)值，concurrencyLevel 為 16，sshift 為 4// 那么計(jì)算出 segmentShift 為 28，segmentMask 為 15，后面會(huì)用到這兩個(gè)值this.segmentShift = 32 - sshift;this.segmentMask = ssize - 1;if (initialCapacity > MAXIMUM_CAPACITY)initialCapacity = MAXIMUM_CAPACITY;// initialCapacity 是設(shè)置整個(gè) map 初始的大小，// 這里根據(jù) initialCapacity 計(jì)算 Segment 數(shù)組中每個(gè)位置可以分到的大小// 如 initialCapacity 為 64，那么每個(gè) Segment 或稱之為"槽"可以分到 4 個(gè)int c = initialCapacity / ssize;if (c * ssize < initialCapacity)++c;// 默認(rèn) MIN_SEGMENT_TABLE_CAPACITY 是 2，這個(gè)值也是有講究的，因?yàn)檫@樣的話，對(duì)于具體的槽上，// 插入一個(gè)元素不至于擴(kuò)容，插入第二個(gè)的時(shí)候才會(huì)擴(kuò)容int cap = MIN_SEGMENT_TABLE_CAPACITY; while (cap < c)cap <<= 1;// 創(chuàng)建 Segment 數(shù)組，// 并創(chuàng)建數(shù)組的第一個(gè)元素 segment[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];// 往數(shù)組寫入 segment[0]UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]this.segments = ss; }

初始化完成，我們得到了一個(gè) Segment 數(shù)組。

我們就當(dāng)是用 new ConcurrentHashMap() 無(wú)參構(gòu)造函數(shù)進(jìn)行初始化的，那么初始化完成后：

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

##put 過(guò)程分析

我們先看 put 的主流程，對(duì)于其中的一些關(guān)鍵細(xì)節(jié)操作，后面會(huì)進(jìn)行詳細(xì)介紹。

public V put(K key, V value) {Segment<K,V> s;if (value == null)throw new NullPointerException();// 1. 計(jì)算 key 的 hash 值int hash = hash(key);// 2. 根據(jù) hash 值找到 Segment 數(shù)組中的位置 j// hash 是 32 位，無(wú)符號(hào)右移 segmentShift(28) 位，剩下低 4 位，// 然后和 segmentMask(15) 做一次與操作，也就是說(shuō) j 是 hash 值的最后 4 位，也就是槽的數(shù)組下標(biāo)int j = (hash >>> segmentShift) & segmentMask;// 剛剛說(shuō)了，初始化的時(shí)候初始化了 segment[0]，但是其他位置還是 null，// ensureSegment(j) 對(duì) segment[j] 進(jìn)行初始化if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck(segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegments = ensureSegment(j);// 3. 插入新值到 槽 s 中return s.put(key, hash, value, false); }

第一層皮很簡(jiǎn)單，根據(jù) hash 值很快就能找到相應(yīng)的 Segment，之后就是 Segment 內(nèi)部的 put 操作了。

Segment 內(nèi)部是由 數(shù)組+鏈表 組成的。

final V put(K key, int hash, V value, boolean onlyIfAbsent) {// 在往該 segment 寫入前，需要先獲取該 segment 的獨(dú)占鎖// 先看主流程，后面還會(huì)具體介紹這部分內(nèi)容HashEntry<K,V> node = tryLock() ? null :scanAndLockForPut(key, hash, value);V oldValue;try {// 這個(gè)是 segment 內(nèi)部的數(shù)組HashEntry<K,V>[] tab = table;// 再利用 hash 值，求應(yīng)該放置的數(shù)組下標(biāo)int index = (tab.length - 1) & hash;// first 是數(shù)組該位置處的鏈表的表頭HashEntry<K,V> first = entryAt(tab, index);// 下面這串 for 循環(huán)雖然很長(zhǎng)，不過(guò)也很好理解，想想該位置沒有任何元素和已經(jīng)存在一個(gè)鏈表這兩種情況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;}// 繼續(xù)順著鏈表走e = e.next;}else {// node 到底是不是 null，這個(gè)要看獲取鎖的過(guò)程，不過(guò)和這里都沒有關(guān)系。// 如果不為 null，那就直接將它設(shè)置為鏈表表頭；如果是null，初始化并設(shè)置為鏈表表頭。if (node != null)node.setNext(first);elsenode = new HashEntry<K,V>(hash, key, value, first);int c = count + 1;// 如果超過(guò)了該 segment 的閾值，這個(gè) segment 需要擴(kuò)容if (c > threshold && tab.length < MAXIMUM_CAPACITY)rehash(node); // 擴(kuò)容后面也會(huì)具體分析else// 沒有達(dá)到閾值，將 node 放到數(shù)組 tab 的 index 位置，// 其實(shí)就是將新的節(jié)點(diǎn)設(shè)置成原鏈表的表頭setEntryAt(tab, index, node);++modCount;count = c;oldValue = null;break;}}} finally {// 解鎖unlock();}return oldValue; }

整體流程還是比較簡(jiǎn)單的，由于有獨(dú)占鎖的保護(hù)，所以 segment 內(nèi)部的操作并不復(fù)雜。至于這里面的并發(fā)問(wèn)題，我們稍后再進(jìn)行介紹。

到這里 put 操作就結(jié)束了，接下來(lái)，我們說(shuō)一說(shuō)其中幾步關(guān)鍵的操作。
##初始化槽: ensureSegment

ConcurrentHashMap 初始化的時(shí)候會(huì)初始化第一個(gè)槽 segment[0]，對(duì)于其他槽來(lái)說(shuō)，在插入第一個(gè)值的時(shí)候進(jìn)行初始化。

這里需要考慮并發(fā)，因?yàn)楹芸赡軙?huì)有多個(gè)線程同時(shí)進(jìn)來(lái)初始化同一個(gè)槽 segment[k]，不過(guò)只要有一個(gè)成功了就可以。

private Segment<K,V> ensureSegment(int k) {final Segment<K,V>[] ss = this.segments;long u = (k << SSHIFT) + SBASE; // raw offsetSegment<K,V> seg;if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {// 這里看到為什么之前要初始化 segment[0] 了，// 使用當(dāng)前 segment[0] 處的數(shù)組長(zhǎng)度和負(fù)載因子來(lái)初始化 segment[k]// 為什么要用“當(dāng)前”，因?yàn)?segment[0] 可能早就擴(kuò)容過(guò)了Segment<K,V> proto = ss[0];int cap = proto.table.length;float lf = proto.loadFactor;int threshold = (int)(cap * lf);// 初始化 segment[k] 內(nèi)部的數(shù)組HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))== null) { // 再次檢查一遍該槽是否被其他線程初始化了。Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);// 使用 while 循環(huán)，內(nèi)部用 CAS，當(dāng)前線程成功設(shè)值或其他線程成功設(shè)值后，退出while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))== null) {if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))break;}}}return seg; }

總的來(lái)說(shuō)，ensureSegment(int k) 比較簡(jiǎn)單，對(duì)于并發(fā)操作使用 CAS 進(jìn)行控制。

我沒搞懂這里為什么要搞一個(gè) while 循環(huán)，CAS 失敗不就代表有其他線程成功了嗎，為什么要再進(jìn)行判斷？

##獲取寫入鎖: scanAndLockForPut

前面我們看到，在往某個(gè) segment 中 put 的時(shí)候，首先會(huì)調(diào)用 node = tryLock() ? null : scanAndLockForPut(key, hash, value)，也就是說(shuō)先進(jìn)行一次 tryLock() 快速獲取該 segment 的獨(dú)占鎖，如果失敗，那么進(jìn)入到 scanAndLockForPut 這個(gè)方法來(lái)獲取鎖。

下面我們來(lái)具體分析這個(gè)方法中是怎么控制加鎖的。

private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {HashEntry<K,V> first = entryForHash(this, hash);HashEntry<K,V> e = first;HashEntry<K,V> node = null;int retries = -1; // negative while locating node// 循環(huán)獲取鎖while (!tryLock()) {HashEntry<K,V> f; // to recheck first belowif (retries < 0) {if (e == null) {if (node == null) // speculatively create node// 進(jìn)到這里說(shuō)明數(shù)組該位置的鏈表是空的，沒有任何元素// 當(dāng)然，進(jìn)到這里的另一個(gè)原因是 tryLock() 失敗，所以該槽存在并發(fā)，不一定是該位置node = new HashEntry<K,V>(hash, key, value, null);retries = 0;}else if (key.equals(e.key))retries = 0;else// 順著鏈表往下走e = e.next;}// 重試次數(shù)如果超過(guò) MAX_SCAN_RETRIES（單核1多核64），那么不搶了，進(jìn)入到阻塞隊(duì)列等待鎖// lock() 是阻塞方法，直到獲取鎖后返回else if (++retries > MAX_SCAN_RETRIES) {lock();break;}else if ((retries & 1) == 0 &&// 這個(gè)時(shí)候是有大問(wèn)題了，那就是有新的元素進(jìn)到了鏈表，成為了新的表頭// 所以這邊的策略是，相當(dāng)于重新走一遍這個(gè) scanAndLockForPut 方法(f = entryForHash(this, hash)) != first) {e = first = f; // re-traverse if entry changedretries = -1;}}return node; }

這個(gè)方法有兩個(gè)出口，一個(gè)是 tryLock() 成功了，循環(huán)終止，另一個(gè)就是重試次數(shù)超過(guò)了 MAX_SCAN_RETRIES，進(jìn)到 lock() 方法，此方法會(huì)阻塞等待，直到成功拿到獨(dú)占鎖。

這個(gè)方法就是看似復(fù)雜，但是其實(shí)就是做了一件事，那就是獲取該 segment 的獨(dú)占鎖，如果需要的話順便實(shí)例化了一下 node。
##擴(kuò)容: rehash

重復(fù)一下，segment 數(shù)組不能擴(kuò)容，擴(kuò)容是 segment 數(shù)組某個(gè)位置內(nèi)部的數(shù)組 HashEntry[] 進(jìn)行擴(kuò)容，擴(kuò)容后，容量為原來(lái)的 2 倍。

首先，我們要回顧一下觸發(fā)擴(kuò)容的地方，put 的時(shí)候，如果判斷該值的插入會(huì)導(dǎo)致該 segment 的元素個(gè)數(shù)超過(guò)閾值，那么先進(jìn)行擴(kuò)容，再插值，讀者這個(gè)時(shí)候可以回去 put 方法看一眼。

該方法不需要考慮并發(fā)，因?yàn)榈竭@里的時(shí)候，是持有該 segment 的獨(dú)占鎖的。

// 方法參數(shù)上的 node 是這次擴(kuò)容后，需要添加到新的數(shù)組中的數(shù)據(jù)。 private void rehash(HashEntry<K,V> node) {HashEntry<K,V>[] oldTable = table;int oldCapacity = oldTable.length;// 2 倍int newCapacity = oldCapacity << 1;threshold = (int)(newCapacity * loadFactor);// 創(chuàng)建新數(shù)組HashEntry<K,V>[] newTable =(HashEntry<K,V>[]) new HashEntry[newCapacity];// 新的掩碼，如從 16 擴(kuò)容到 32，那么 sizeMask 為 31，對(duì)應(yīng)二進(jìn)制 ‘000...00011111’int sizeMask = newCapacity - 1;// 遍歷原數(shù)組，老套路，將原數(shù)組位置 i 處的鏈表拆分到 新數(shù)組位置 i 和 i+oldCap 兩個(gè)位置for (int i = 0; i < oldCapacity ; i++) {// e 是鏈表的第一個(gè)元素HashEntry<K,V> e = oldTable[i];if (e != null) {HashEntry<K,V> next = e.next;// 計(jì)算應(yīng)該放置在新數(shù)組中的位置，// 假設(shè)原數(shù)組長(zhǎng)度為 16，e 在 oldTable[3] 處，那么 idx 只可能是 3 或者是 3 + 16 = 19int idx = e.hash & sizeMask;if (next == null) // 該位置處只有一個(gè)元素，那比較好辦newTable[idx] = e;else { // Reuse consecutive sequence at same slot// e 是鏈表表頭HashEntry<K,V> lastRun = e;// idx 是當(dāng)前鏈表的頭結(jié)點(diǎn) e 的新位置int lastIdx = idx;// 下面這個(gè) for 循環(huán)會(huì)找到一個(gè) lastRun 節(jié)點(diǎn)，這個(gè)節(jié)點(diǎn)之后的所有元素是將要放到一起的for (HashEntry<K,V> last = next;last != null;last = last.next) {int k = last.hash & sizeMask;if (k != lastIdx) {lastIdx = k;lastRun = last;}}// 將 lastRun 及其之后的所有節(jié)點(diǎn)組成的這個(gè)鏈表放到 lastIdx 這個(gè)位置newTable[lastIdx] = lastRun;// 下面的操作是處理 lastRun 之前的節(jié)點(diǎn)，// 這些節(jié)點(diǎn)可能分配在另一個(gè)鏈表中，也可能分配到上面的那個(gè)鏈表中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);}}}}// 將新來(lái)的 node 放到新數(shù)組中剛剛的 兩個(gè)鏈表之一 的 頭部int nodeIndex = node.hash & sizeMask; // add the new nodenode.setNext(newTable[nodeIndex]);newTable[nodeIndex] = node;table = newTable; }

這里的擴(kuò)容比之前的 HashMap 要復(fù)雜一些，代碼難懂一點(diǎn)。上面有兩個(gè)挨著的 for 循環(huán)，第一個(gè) for 有什么用呢？

仔細(xì)一看發(fā)現(xiàn)，如果沒有第一個(gè) for 循環(huán)，也是可以工作的，但是，這個(gè) for 循環(huán)下來(lái)，如果 lastRun 的后面還有比較多的節(jié)點(diǎn)，那么這次就是值得的。因?yàn)槲覀冎恍枰寺?lastRun 前面的節(jié)點(diǎn)，后面的一串節(jié)點(diǎn)跟著 lastRun 走就是了，不需要做任何操作。

我覺得 Doug Lea 的這個(gè)想法也是挺有意思的，不過(guò)比較壞的情況就是每次 lastRun 都是鏈表的最后一個(gè)元素或者很靠后的元素，那么這次遍歷就有點(diǎn)浪費(fèi)了。不過(guò) Doug Lea 也說(shuō)了，根據(jù)統(tǒng)計(jì)，如果使用默認(rèn)的閾值，大約只有 1/6 的節(jié)點(diǎn)需要克隆。
##get 過(guò)程分析

相對(duì)于 put 來(lái)說(shuō)，get 真的不要太簡(jiǎn)單。

• 計(jì)算 hash 值，找到 segment 數(shù)組中的具體位置，或我們前面用的“槽”
• 槽中也是一個(gè)數(shù)組，根據(jù) hash 找到數(shù)組中具體的位置
• 到這里是鏈表了，順著鏈表進(jìn)行查找即可

public V get(Object key) {Segment<K,V> s; // manually integrate access methods to reduce overheadHashEntry<K,V>[] tab;// 1. hash 值int h = hash(key);long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;// 2. 根據(jù) hash 找到對(duì)應(yīng)的 segmentif ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&(tab = s.table) != null) {// 3. 找到segment 內(nèi)部數(shù)組相應(yīng)位置的鏈表，遍歷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; }

##并發(fā)問(wèn)題分析

現(xiàn)在我們已經(jīng)說(shuō)完了 put 過(guò)程和 get 過(guò)程，我們可以看到 get 過(guò)程中是沒有加鎖的，那自然我們就需要去考慮并發(fā)問(wèn)題。

添加節(jié)點(diǎn)的操作 put 和刪除節(jié)點(diǎn)的操作 remove 都是要加 segment 上的獨(dú)占鎖的，所以它們之間自然不會(huì)有問(wèn)題，我們需要考慮的問(wèn)題就是 get 的時(shí)候在同一個(gè) segment 中發(fā)生了 put 或 remove 操作。

1、put 操作的線程安全性。

• 初始化槽，這個(gè)我們之前就說(shuō)過(guò)了，使用了 CAS 來(lái)初始化 Segment 中的數(shù)組。
• 添加節(jié)點(diǎn)到鏈表的操作是插入到表頭的，所以，如果這個(gè)時(shí)候 get 操作在鏈表遍歷的過(guò)程已經(jīng)到了中間，是不會(huì)影響的。當(dāng)然，另一個(gè)并發(fā)問(wèn)題就是 get 操作在 put 之后，需要保證剛剛插入表頭的節(jié)點(diǎn)被讀取，這個(gè)依賴于 setEntryAt 方法中使用的 UNSAFE.putOrderedObject。
• 擴(kuò)容。擴(kuò)容是新創(chuàng)建了數(shù)組，然后進(jìn)行遷移數(shù)據(jù)，最后面將 newTable 設(shè)置給屬性 table。所以，如果 get 操作此時(shí)也在進(jìn)行，那么也沒關(guān)系，如果 get 先行，那么就是在舊的 table 上做查詢操作；而 put 先行，那么 put 操作的可見性保證就是 table 使用了 volatile 關(guān)鍵字。

2、remove 操作的線程安全性。

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

• get 操作需要遍歷鏈表，但是 remove 操作會(huì)"破壞"鏈表。

• 如果 remove 破壞的節(jié)點(diǎn) get 操作已經(jīng)過(guò)去了，那么這里不存在任何問(wèn)題。

• 如果 remove 先破壞了一個(gè)節(jié)點(diǎn)，分兩種情況考慮。 1、如果此節(jié)點(diǎn)是頭結(jié)點(diǎn)，那么需要將頭結(jié)點(diǎn)的 next 設(shè)置為數(shù)組該位置的元素，table 雖然使用了 volatile 修飾，但是 volatile 并不能提供數(shù)組內(nèi)部操作的可見性保證，所以源碼中使用了 UNSAFE 來(lái)操作數(shù)組，請(qǐng)看方法 setEntryAt。2、如果要?jiǎng)h除的節(jié)點(diǎn)不是頭結(jié)點(diǎn)，它會(huì)將要?jiǎng)h除節(jié)點(diǎn)的后繼節(jié)點(diǎn)接到前驅(qū)節(jié)點(diǎn)中，這里的并發(fā)保證就是 next 屬性是 volatile 的。

出處：https://www.javadoop.com/post/hashmap#Java7%20ConcurrentHashMap

總結(jié)

以上是生活随笔為你收集整理的Java7 ConcurrentHashMap详解的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。