App下載

java核心基礎(chǔ) ConcurrentHashMap源碼分析

猿友 2021-07-19 15:47:02 瀏覽數(shù) (1738)
反饋

HashMap 作為我們平日里 java 開發(fā)使用最多的集合,本篇文章要介紹的 ConcurrentHashMap 是前者的升級,也許有些小伙伴有聽說過。本篇文章將帶大家簡單地了解一下 java 的核心基礎(chǔ) ConcurrentHashMap 的知識內(nèi)容。

1 前言

ConcurrentHashMap是基于Hash表的Map接口實現(xiàn),鍵與值均不允許為NULL,他是一個線程安全的Map。同時他也是一個無序的Map,不同時間進行遍歷可能會得到不同的順序。在JDK1.8之前,ConcurrentHashMap使用分段鎖以在保證線程安全的同時獲得更大的效率。JDK1.8開始舍棄了分段鎖,使用自旋+CAS+sync關(guān)鍵字來實現(xiàn)同步。本文所述便是基于JDK1.8。
ConcurrentHashMap與HashMap有共同之處,一些HashMap的基本概念與實現(xiàn),本文不再贅述。

2 繼承關(guān)系

public class ConcurrentHashMap<K,V> extends AbstractMap<K,V>
    implements ConcurrentMap<K,V>, Serializable

可以看到ConcurrentHashMap繼承了AbstractMap及ConcurrentMap抽象類,并實現(xiàn)了Serializable接口,這說明ConcurrentHashMap是一個線程安全的標準Map,且允許序列化。與HashMap不同是ConcurrentHashMap不允許Clone。

3 構(gòu)造方法

ConcurrentHashMap同樣是采用懶初始化的方式,有實際元素時才進行容器的初始化。因此其構(gòu)造方法與HashMap相差無幾。

// 無參構(gòu)造
public ConcurrentHashMap() {
}
// 帶有初始容量的構(gòu)造
public ConcurrentHashMap(int initialCapacity) {
    if (initialCapacity < 0)
        throw new IllegalArgumentException();
    int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
               MAXIMUM_CAPACITY :
               tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1)); //求最小二次冪
    this.sizeCtl = cap; // sizeCtl 暫存容量
}
// 帶有初始集合的構(gòu)造
public ConcurrentHashMap(Map<? extends K, ? extends V> m) {
    this.sizeCtl = DEFAULT_CAPACITY;
    putAll(m);
}
// 指定初始容量和裝載因子的構(gòu)造
public ConcurrentHashMap(int initialCapacity, float loadFactor) {
    this(initialCapacity, loadFactor, 1);
}

以上構(gòu)造并沒有什么特別的,邏輯也相對簡單,不再詳細解析,感興趣的話可以到前言提到的HashMap篇了解。
除此之外,ConcurrentHashMap擁有另外一個值得注意的構(gòu)造方法:
指定初始容量,裝載因子以及并發(fā)級別的構(gòu)造方法:
源碼:

public ConcurrentHashMap(int initialCapacity,
                         float loadFactor, int concurrencyLevel) {
    if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0) // 基礎(chǔ)校驗
        throw new IllegalArgumentException();
    if (initialCapacity < concurrencyLevel)   // 取初始容量和并發(fā)級別的最大值
        initialCapacity = concurrencyLevel;  
    long size = (long)(1.0 + (long)initialCapacity / loadFactor);  // 指定了裝載因子,因此就用初始容量和裝載因子計算出實際要的初始容量
    int cap = (size >= (long)MAXIMUM_CAPACITY) ?
        MAXIMUM_CAPACITY : tableSizeFor((int)size);	// 確定計算出初始容量,
    this.sizeCtl = cap;
}

解析: 除了初始容量與裝載因子外,此構(gòu)造方法還有一個并發(fā)級別concurrencyLevel的參數(shù)。在jdk1.7時,并發(fā)級別作為分段鎖的標準進行分段。但是jdk1.8開始舍棄了分段鎖,為了版本兼容,此構(gòu)造方法依然存在,但是concurrencyLevel也不再具有其分段依據(jù)的意義,而是作為初始容量的定義依據(jù)。

4 初始化

ConcurrentHashMap采用懶初始化的方式,在第一次putAvl時如果容器為空,則會調(diào)用initTable()進行容器的初始化:

private final Node<K,V>[] initTable() {
    Node<K,V>[] tab; int sc;
    while ((tab = table) == null || tab.length == 0) {    // tab為空時就一直循環(huán)
        if ((sc = sizeCtl) < 0)       // sc小于0說明正在初始化或者正在復制,放棄CPU等下一次
            Thread.yield(); // lost initialization race; just spin
        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {    // 否則的話就用CAS把sizectl設(shè)置成-1,標識正在初始化,且成功的話
            try {
			// CAS之后再判斷一下,這是為了避免并發(fā),比如上面判斷了tab為空,然后另一個線程做了初始化操作,結(jié)束時sc被設(shè)置為擴容閾值,然后繼續(xù)(sc = sizeCtl) < 0,這時cs還是大于0,所以還是能走進來的,所以這里再判斷一下
                if ((tab = table) == null || tab.length == 0) {   
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;     // sc最初始是存在的初始容量的
                    @SuppressWarnings("unchecked")
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];   // 建立一個容器
                    table = tab = nt; // 然后復制到table屬性上
                    sc = n - (n >>> 2);   // sc等于 n>>>2就是n/4就是0.25n   sc = 0.75n,相當于設(shè)置擴容閾值
                }
            } finally {
                sizeCtl = sc; // sizeCtl賦值成擴容閾值
            }
            break;    // 只要走到這里面,那么說明一定已經(jīng)初始化結(jié)束了,無論初始化是自己做的還是其他線程做的,然后就跳出
        }
    }
    return tab;   // 返回新的tab
}

解析: 容器初始化完成之前不斷的進行循環(huán)。這是因為ConcurrentHashMap是一個支持并發(fā)的Map,可能同時會有多個線程進入initTable方法,但是只有一個線程執(zhí)行初始化操作,那么剩下的線程就需要等待初始化完成再跳出initTable方法,以滿足接下來的putVal操作。
為了保證只有一個線程執(zhí)行初始化操作,使用sizeCtl來作為標識,sizeCtl為-1時即說明當前已有線程正在初始化,則放棄CPU繼續(xù)循環(huán)。sizeCtl不為負數(shù)時,則使用CAS(通過Unsafe+偏移量的方式實現(xiàn))將sizeCtl置為-1,如果當前線程成功的話則進入實際的擴容操作。
通過CAS鎖定成功后再次判斷容器是否為空,這是為了避免并發(fā),比如上面判斷了tab為空,然后另一個線程做了初始化操作,結(jié)束時sc被設(shè)置為擴容閾值,然后繼續(xù)(sc = sizeCtl) < 0,這時cs還是大于0,所以還是能走進來的,所以這里再判斷一下。
如果sc(sizeCtl原值)大于0,則以sc做為初始容量。這個在構(gòu)造方法篇提到過,對于有初始容量要求的構(gòu)造,會以sizeCtl暫存初始容量。否則的話就取默認的初始容量DEFAULT_CAPACITY(16)
接下來就建立目標容量的容器并賦值給容器屬性table,計算該容量下的擴容閾值賦值給sizeCtl 。sizeCtl的賦值放到finally里面的原因是因為無論容器創(chuàng)建成功還是失敗,都需要放開以sizeCtl為負值作為判斷條件的鎖,以保證在本線程創(chuàng)建失敗的情況其它線程能繼續(xù)競爭鎖繼續(xù)進行容器創(chuàng)建的工作。
至此,容器的初始化便完成了。

5 新數(shù)據(jù)載入-putVal

ConcurrentHashMap新數(shù)據(jù)載入主要通過putVal實現(xiàn):

final V putVal(K key, V value, boolean onlyIfAbsent) {
    if (key == null || value == null) throw new NullPointerException();
    int hash = spread(key.hashCode());
    int binCount = 0;
    for (Node<K,V>[] tab = table;;) { // 會一直循環(huán),直到break
        Node<K,V> f; int n, i, fh;
        if (tab == null || (n = tab.length) == 0)
            tab = initTable();  // 初始化table
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {    // unsafe獲取i處的數(shù)據(jù),如果為null
            if (casTabAt(tab, i, null,
                         new Node<K,V>(hash, key, value, null)))    // cas操作,如果i處為null就構(gòu)造新元素,成功就break  失敗就繼續(xù)循環(huán)
                break;                   // no lock when adding to empty bin
        }
        else if ((fh = f.hash) == MOVED)        // f處的元素hash是-1   -1意味著這是一個遷移節(jié)點
            tab = helpTransfer(tab, f);   // 輔助遷移開始
        else {
            V oldVal = null;
            synchronized (f) {    // 用f進行同步
                if (tabAt(tab, i) == f) { // 這里要保證i處還是f,而不是已經(jīng)被其它元素并發(fā)占據(jù)了
                    if (fh >= 0) {    // f處的hash要大于等于0,這是為什么呢?除了MOVED還會有其它負值嗎
                        binCount = 1;
                        for (Node<K,V> e = f;; ++binCount) {  // 從e往后找,并且記錄元素數(shù)量
                            K ek;
                            if (e.hash == hash &&
                                ((ek = e.key) == key ||
                                 (ek != null && key.equals(ek)))) { // 一系列判斷Key是否相等,如果相等
                                oldVal = e.val;   // 記錄舊值
                                if (!onlyIfAbsent)    // 不是僅允許新增的話
                                    e.val = value;    // 記錄新值
                                break;    // 成功的話就跳出大循環(huán)
                            }
                            Node<K,V> pred = e;
                            if ((e = e.next) == null) {   // e向后移動,  如果移動之后是nul,就是在末尾的話
                                pred.next = new Node<K,V>(hash, key,
                                                          value, null);   // 建立一個新節(jié)點
                                break;     // 成功的話就跳出大循環(huán)
                            }
                        }
                    }
                    else if (f instanceof TreeBin) {  // 如果是一個樹化節(jié)點
                        Node<K,V> p;
                        binCount = 2; // 樹化的節(jié)點的話固定元素數(shù)量為2,這是一個相對特殊的值,即會擴容又不會重復樹化
                        if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
                                                       value)) != null) { // 往樹里面新增元素,如果存在就返回那個節(jié)點,不存在就是新增節(jié)點
                            oldVal = p.val;   // 記錄舊值
                            if (!onlyIfAbsent)    // 不是僅允許新增的話
                                p.val = value; // 記錄新值
                        }
                    }
                }
            }
            if (binCount != 0) {  // 如果元素數(shù)不為0的話
                if (binCount >= TREEIFY_THRESHOLD)    // 判斷是否到了樹化閾值
                    treeifyBin(tab, i);   // 到了的話進行樹化
                if (oldVal != null)   // 覆蓋操作直接返回舊值不會執(zhí)行addCount
                    return oldVal;
                break;     // binCount != 0才跳出大循環(huán),為0是tabAt(tab, i) != f導致的,即i處已經(jīng)被其他線程覆蓋了
            }
        }
    }
    addCount(1L, binCount);   // 
    return null;
}

解析: 完成為空校驗后,通過spread方法來計算出hash以確定下標位置,spread的計算方式為(h ^ (h >>> 16)) & HASH_BITS,HASH_BITS的值為0x7fffffff,描述出來就是將hashCode的高16位和低16位做異或操作,并保證最高位符號位為0(結(jié)果是一個正數(shù)),注解表示這樣做的原因是為了使數(shù)據(jù)更加分散,盡可能的避免hash沖突。
如果容器tab為空或者長度為0說明容器未初始化,那么就調(diào)用initTable進行容器初始化。
否則的話對hash與現(xiàn)容量進行與運算得出數(shù)據(jù)應(yīng)處的下標,判斷此下標所在處節(jié)點是否為空,為空說明沒有元素,直接創(chuàng)建一個新節(jié)點作為頭元素放進去,這一步通過CAS操作實現(xiàn),避免并發(fā)情況下另外的線程先一步完成頭節(jié)點創(chuàng)建操作。
如果下標所在處節(jié)點不為空,說明該處已經(jīng)有元素了,此時判斷頭節(jié)點的hash是否為MOVED,MOVED的值為-1。上面提到過,正常元素通過spread方法計算出來的hash值都會使正數(shù)。此處-1為一個特殊值,意味著此節(jié)點正在進行擴容遷移工作,那么此時就調(diào)用8 輔助擴容-helpTransfer方法進行輔助遷移工作。
如果節(jié)點hash不為MOVED,意味著這是一個正常節(jié)點,就執(zhí)行元素載入工作,使用synchronized關(guān)鍵字實現(xiàn)同步, 同步塊內(nèi)開始還要使用tabAt(tab, i) == f判斷進入同步后i處的節(jié)點還是原節(jié)點,接下來就是元素的載入工作了,整體和HashMap的流程是相差無幾的,感興趣的話可以去文首提到的解析HashMap的文章了解一下。synchronized塊將元素載入節(jié)點后,接著會對binCount進行判斷,binCount在節(jié)點還處于鏈表模式下的情況下記錄節(jié)點在新元素載入后的元素總數(shù)量,此處判斷binCount達到TREEIFY_THRESHOLD(8) 樹化閾值后,就會對該節(jié)點進行樹化操作。樹化操作依然是通過synchronized來完成的,這里不做過多延伸。
以上整個新元素載入操作都是一個自旋的過程,一直到新元素載入成功后通過break跳出自旋過程。
新元素載入節(jié)點后,需要對count實時維護,count通過 6 維護與啟動擴容-addCount 方法完成同步維護工作。

6 維護與啟動擴容-addCount

ConcurrentHashMap通過addCount方法實時維護內(nèi)部元素數(shù)量,并在達到擴容閾值的情況下啟動擴容操作:

private final void addCount(long x, int check) {
    CounterCell[] as; long b, s;  
    if ((as = counterCells) != null ||    // counterCells不為空,如果沒經(jīng)歷過并發(fā)肯定是空的,這個判斷意思是維護count的過程中已經(jīng)經(jīng)歷過并發(fā),已經(jīng)處于不使用baseCount維護的階段了
        !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {       //這個比較唯一會出現(xiàn)false的情況就是b = baseCount執(zhí)行之后,還沒進行cas時,baseCount又被改了。意思是如果沒有并發(fā)就用baseCount維護,并發(fā)導致baseCount維護失敗就進入counterCells維護
        CounterCell a; long v; int m;
        boolean uncontended = true;   // 記錄是否遭遇了并發(fā)
        if (as == null || (m = as.length - 1) < 0 ||      // as等于null意味著baseCount維護遭遇并發(fā)了,且是初次(非初次counterCells不會為null)  (m = as.length - 1) < 0說明counterCells長度為0,即沒有實際內(nèi)容
            (a = as[ThreadLocalRandom.getProbe() & m]) == null ||     // counterCells是一個分段計算的概念,這里就是隨機找一個槽,如果這個槽位空的話
            !(uncontended =
              U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {      // 槽不為空的話就用cas把槽里面的值賦值給v+a.val    如果失敗的話
            fullAddCount(x, uncontended); //  用fullAddCount  就是LongAdder的模式將x放進去,初始化counterCells
            return;
        }
        if (check <= 1)       // check小于等于1(1表示是當前節(jié)點的第一個元素,-1標識是replace方法或者clear過來的,這些都不需要擴容)  
            return;
        s = sumCount();   // 計算一下現(xiàn)在的總量
    }
    if (check >= 0) {  //baseCount成功的情況且 check大于0,即不是replace或者clear方法進來的
        Node<K,V>[] tab, nt; int n, sc;
        /**
        1.s >= (long)(sc = sizeCtl)  s是總數(shù)量  s大于等于擴容閾值sizeCtl時,或者sizeCtl為負數(shù),正在擴容時
        2.(tab = table) != null 并且容器tab不為空,意思就是表已經(jīng)初始化完畢了,是執(zhí)行擴容而不是執(zhí)行初始化
        3. (n = tab.length) < MAXIMUM_CAPACITY)  并且容量還沒有達到最大容量時
         */
        while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
               (n = tab.length) < MAXIMUM_CAPACITY) {
            int rs = resizeStamp(n);  // 獲得容量特征值:容量n的左0數(shù)量并將高位置1
            if (sc < 0) { // 如果sc小于0  意味這個正在進行并發(fā)擴容
                /**
                sc >>> RESIZE_STAMP_SHIFT != rs是判斷現(xiàn)在的容量與sc中記錄的長度是不是不一樣長,不一樣長的話說明不是同一長度的擴容,就要break-保證是同等長度的擴容
                sc == rs + 1 這個地方在看來是一個bug,如果已經(jīng)處于擴容中,sc目前處于高16位是容量特征值,低16位標識擴容線程量特征值的狀態(tài),而此時的rs是容量特征值的狀態(tài),也就是目前sc的高16位,這里原意應(yīng)該是判斷目前是不是只剩下一個非初始擴容線程
                             ,為什么這么說呢,因為初始擴容線程會把sc設(shè)置稱高16位是容量特征值即rs,低16位標識擴容線程量特征值(初始擴容線程+2,非初始擴容線程+1),這一步判斷的就是如過只剩下一個非初始擴容線程了,說明初始擴容線程已經(jīng)結(jié)束,現(xiàn)在已經(jīng)進入擴容的最后階段了,就不需要再繼續(xù)進行輔助擴容了
                            ,正確的方式應(yīng)該是sc == (rs << RESIZE_STAMP_SHIFT) + 1
                sc == rs + MAX_RESIZERS  和上面+1一樣,不過這個是判斷參與resize的線程是否已經(jīng)達到所允許的最大值 ,看起來也是一個bug,正確的應(yīng)是sc == (rs << RESIZE_STAMP_SHIFT) + MAX_RESIZERS
                (nt = nextTable) == null  nextTable為空說明已經(jīng)擴容完畢了,不再需要輔助擴容了
                transferIndex <= 0   // 判斷下邊界是不是已經(jīng)到0了,擴容是根據(jù)CPU算出步長,一個擴容線程擴容一段,從高到低分配,transferIndex記錄的就行目前分配到哪個下標了,如果這個下標小于等于0說明整個容器已經(jīng)分配完了,就不需要在輔助擴容了
                 */
                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))       // 如果是輔助擴容的sc已經(jīng)處于高16位是容量特征值,低16位標識擴容線程量特征值的狀態(tài)了+1讓地位的擴容線程量特征值再+1
                    transfer(tab, nt);    // 進行擴容
            }
            else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                         (rs << RESIZE_STAMP_SHIFT) + 2)) // 初始擴容現(xiàn)在是+2 執(zhí)行完這一步sc的值高16位是容量特征值,低16位標識擴容線程量特征值,因為resizeStamp時將高位置為1是的其又是一個負數(shù),表示正在擴容
                transfer(tab, null);
            s = sumCount(); // 計算總數(shù)
        } // 一直循環(huán)擴容
    }
}

解析: count的維護有兩種方式:未產(chǎn)生并發(fā)場景時通過baseCount維護,經(jīng)歷過并發(fā)場景后轉(zhuǎn)變?yōu)橥ㄟ^counterCells維護。baseCount模式為直接使用int進行加減運算,cas保證同步,counterCells模式類似于1.8之前的ConcurrentHashMap,采用一個分段的概念,運算時隨機找一個槽,通過cas保證一個槽的值加算同步,count即為所以槽的加和(使用LongAdder實現(xiàn))。
addCount方法初始即通過counterCells是否為空(為空說明未經(jīng)歷初始化,使用的baseCount模式)來判斷當前是否為counterCells模式,如果處于counterCells模式則進入counterCells模式計算邏輯,否則的話使用baseCount模式來維護count并記錄為s,如果遭遇并發(fā)導致維護失敗,則轉(zhuǎn)換為counterCells模式進入counterCells模式計算邏輯。
進入counterCells模式,首先看counterCells是否已完成初始化,若為初始化進入fullAddCount,已初始化的話則隨機找到counterCells中的一個槽位,如果這個槽位未初始化則進入fullAddCount,如果已初始化則對這個操作進行CAS的加操作來維護count,CAS成功則維護完成,并發(fā)導致CAS維護失敗則進入fullAddCountfullAddCount有點類似于CMS的full GC退化機制,會完成counterCells的初始化以及并發(fā)沖突場景下同步完成count維護。counterCells最后則判斷check是否小于等于1(1表示是當前節(jié)點的第一個元素,-1標識是replace方法或者clear過來的,這些都不需要擴容)則直接結(jié)束,不進行擴容判定,否則的話通過sumCount獲得當前count并記錄為s以準備后續(xù)的擴容判定。
check大于等于0(即不是replace或者clear方法引起的count改變)時進行擴容判定,執(zhí)行擴容需要滿足三個條件:1.count大于擴容閾值或者當前正在執(zhí)行擴容操作2.容器已初始化完畢3.容量未達到最大容量。滿足以上條件則需判定當前是否正在擴容,因為擴容時會將SIZECTL設(shè)置為一個特征值,這個特征值為負值,因此通過sc是否小于0來判定當前是否處于擴容狀態(tài)。接下來使用resizeStamp獲得容量特征值(一個表示容量n的左0數(shù)量并將高位置1的值),如果當前未處于擴容過程,則通過容量特征值獲得擴容特征值(高16位為容量特征值,低16位為擴容線程量特征值)并通過CAS修改SIZECTL為擴容特征值,成功則當前線程作為第一個擴容線程啟動擴容,失敗則重新計算count并進行擴容判定。
如果已處于擴容過程,則判斷是否還要寫輔助擴容線程,滿足以下幾個條件則:1.此次擴容的容量與當前正在進行的容量一致2.擴容線程未達到最大額定值3.不是處于擴容收尾階段(只剩下一個輔助擴容線程,nextTable已經(jīng)為空,擴容下標已為0無法繼續(xù)分配)則使用CAS修改擴容特征值成功后當前線程作為輔助擴容線程加入擴容任務(wù),不滿足條件則重新計算count并自旋進行擴容判定。

7 進行擴容-transfer

ConcurrentHashMap的實際擴容操作通過transfer來完成:

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
    int n = tab.length, stride;
   // CPU大于1的話  步長就是0.25n/CPU數(shù)    否則就是n     算出來的步長小于最小步長的話  就按最小步長來
    if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
        stride = MIN_TRANSFER_STRIDE; // subdivide range
    if (nextTab == null) {            // initiating   // nextTab == null 說明是第一個來擴容的線程
        try {
            @SuppressWarnings("unchecked")
            Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];  // 建立一個二倍長度的數(shù)組,然后賦值給nextTab
            nextTab = nt;
        } catch (Throwable ex) {      // try to cope with OOME
            sizeCtl = Integer.MAX_VALUE;      // 異常的話就是OOM?  唯一的異常也就是n過大?
            return;
        }
        nextTable = nextTab;  // 這個賦值為什么不做同步呢?不怕并發(fā)問題嗎,哦  不會  因為調(diào)用這個方法的地方已經(jīng)確定了,只會有一個初始線程過來
        transferIndex = n;    //   n就是新的長度   
    }
    int nextn = nextTab.length;   // 新容量
    ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);   // 創(chuàng)建一個中繼節(jié)點
    boolean advance = true;       //一個終止的標記
    boolean finishing = false; // to ensure sweep before committing nextTab  // 是否已經(jīng)處于收尾掃描提交階段
    for (int i = 0, bound = 0;;) {    // 一直循環(huán)著按步長擴容
        Node<K,V> f; int fh;
        while (advance) {     //  計算出bound和i   賦值TRANSFERINDEX
            int nextIndex, nextBound;
            if (--i >= bound || finishing)        // i大于bound,說明什么呢,說明i已經(jīng)在區(qū)間內(nèi)了
                advance = false;
            else if ((nextIndex = transferIndex) <= 0) {      如果下一次的上邊界已經(jīng)到0了 也停止
                i = -1;
                advance = false;
            }
            else if (U.compareAndSwapInt
                     (this, TRANSFERINDEX, nextIndex,
                      nextBound = (nextIndex > stride ?           //nextBound作為新的上邊界
                                   nextIndex - stride : 0))) {
                bound = nextBound;        //nextBound作為這次的下邊界
                i = nextIndex - 1;    // 從nextIndex(上邊界) - 1開始依次遞減到bound區(qū)域
                advance = false;
            }
        }
        if (i < 0 || i >= n || i + n >= nextn) {  // 如果i不在0-n范圍內(nèi),已經(jīng)越界
            int sc;
            if (finishing) {  // 如果已經(jīng)在收尾階段
                nextTable = null; // 重置nextTable
                table = nextTab;  // 容器確認
                sizeCtl = (n << 1) - (n >>> 1); // 2n-0.5n = 0.75*2n   就是新的擴容閾值
                return;
            }
            if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {   // sc減1  意味當前擴容線程結(jié)束
                if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)     // 如果sc-2和容量特征值一樣時,就說明現(xiàn)在是最后的一個擴容線程了
                    return;   // 直接返回
                finishing = advance = true;   // 不是最后的話   設(shè)置為收尾階段?,確實,進入這里面已經(jīng)越界了...
                i = n; // recheck before commit           // 提交前的重復檢查?
            }
        }
        else if ((f = tabAt(tab, i)) == null)     // 如果i處還沒有使用
            advance = casTabAt(tab, i, null, fwd);    // cas設(shè)置i處的新fwdnode,成功還是失敗會記錄在advance上面
        else if ((fh = f.hash) == MOVED)  // 如果這節(jié)點已經(jīng)是一個中繼節(jié)點   說明什么呢?說明已經(jīng)有人處理過了上一步
            advance = true; // already processed
        else {    // 如果有值,并且不是中繼節(jié)點的話
            synchronized (f) {    // 同步處理f,所以只會有一個線程作為參與者
                if (tabAt(tab, i) == f) { // 同步之后還要再判斷一下  萬一f被并發(fā)處理掉了呢
                    Node<K,V> ln, hn;
                    if (fh >= 0) {    // 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) {    // 這是找最后一段的開頭節(jié)點
                                runBit = b;
                                lastRun = p;
                            }
                        }
                        if (runBit == 0) {    // 最后一段的開頭節(jié)點是低位的話
                            ln = lastRun; // 開頭節(jié)點賦值到ln   也就是說從lastRun開始,之后的都沒有變化了,都是在同一位
                            hn = null;
                        }
                        else {
                            hn = lastRun;
                            ln = null;
                        }
                        for (Node<K,V> p = f; p != lastRun; p = p.next) { // 只需遍歷到lastRun即可
                            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);   // 往前補
                        }
                        setTabAt(nextTab, i, ln); // 賦值新容器
                        setTabAt(nextTab, i + n, hn);
                        setTabAt(tab, i, fwd);    // 舊容器節(jié)點變?yōu)橹欣^節(jié)點,但是如果在這一步之前就有新元素來了怎么辦,不會來,因為新增也是sync(f)的
                        advance = true;
                    }
                    else if (f instanceof TreeBin) {  // 如果是一個樹化節(jié)點
                        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;
                            }
                        }
                        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;
                    }
                }
            }
        }
    }
}

解析: transfer采用分段擴容的方式,從n-1->0,每個線程每次會占用一個步長的區(qū)間,然后針對這個區(qū)間進行擴容,擴容完畢再去占用下一個區(qū)間,直到無法占用新的區(qū)間則結(jié)束擴容流程。
步長的計算與應(yīng)用的CPU數(shù)有關(guān),如果當前應(yīng)用CPU數(shù)為1則步長為n,即單線程庫擴容。如果當前應(yīng)用CPU數(shù)不為1,則為0.25n/CPU,但是如果算出的步長小于最小MIN_TRANSFER_STRIDE(16),則以MIN_TRANSFER_STRIDE為步長。
接下來通過nextTable 是否為空判斷是否為初始擴容線程,如果是的話就創(chuàng)建一個長度為2n的新容器并記錄到nextTable屬性,所有擴容線程共同使用nextTable作為新容器,同時記錄當前擴容進度下標transferIndex為n(擴容占用是從n到0的,因此初始未占用時就是n)。這里雖然沒進行加鎖但是依然不會有并發(fā)問題,因為nextTable為空的只有初始線程,而初始線程的創(chuàng)建則是線程安全的。
步長確認完畢,新容器創(chuàng)建完畢,擴容進度下標初始化完成,接下來就開始進行區(qū)間占用了:使用自旋+CAS來進行線程安全的區(qū)間占用,advance作為自旋結(jié)束的條件:如果i大于bound意味著本次占用的區(qū)間還沒處理完則結(jié)束自旋。finishing意味著擴容已經(jīng)進入尾聲了,也無需再加入擴容,結(jié)束自旋。擴容進度下標transferIndex小于等于0說明所有的區(qū)間都已經(jīng)被占用了也無需再加入擴容,結(jié)束自旋。如果以上條件不成立,則使用CAS修改擴容進度下標transferIndex嘗試占用一個步長的區(qū)間,如果失敗則自旋繼續(xù)判定區(qū)間占用,修改成功則意味著區(qū)間占用成功,賦值本次區(qū)間的擴容游標i以及下標邊界bound。
區(qū)間占用成功后開始從i到bound的擴容階段,首先需要判斷i的合法性(需要在i和n之間,即判斷是不是已擴容完畢,如–i的循環(huán)會使得i為負數(shù)表示擴容完畢),如果沒占據(jù)到區(qū)間則判斷是不是處于于finishing階段,處于的話說明自己以及是最后一個擴容線程了,將新容器nextTable賦值為table,并設(shè)置新的擴容閾值。如果不是處于finishing階段則因為當前線程沒占據(jù)到區(qū)間需要退出擴容,那么響應(yīng)的擴容特征值也要相應(yīng)的-1來完成擴容線程量特征值的維護,如果擴容特征值維護失敗則需要繼續(xù)自旋嘗試退出,如果維護成功則判斷自己退出之后是不是只能先一個擴容線程,因為起始擴容線程的線程擴容量特征值為+2,所以通過-2來判斷,如果自己退出之后僅剩一個線程在擴容,那么就把finishing設(shè)置為true,以使其在擴容工作結(jié)束后進行新容器的賦值以及新擴容閾值的維護。
如果i處于合法范圍內(nèi),說明處于正常擴容中,那么獲取i處的節(jié)點,節(jié)點為空則利用一個CAS為其賦值一個ForwardingNode節(jié)點,這個節(jié)點是一個中繼節(jié)點,意味著這個節(jié)點正處于擴容狀態(tài),其hash值為特殊值MOVED(-1),在5 新數(shù)據(jù)載入-putVal 的過程中如果發(fā)現(xiàn)目標節(jié)點hash值為MOVED,那么putVal線程就會暫停put操作,作為輔助擴容線程先行擴容容器,擴容完畢后再進行put操作。
如果i處節(jié)點不會空但是hash值已經(jīng)是MOVED,那么說明節(jié)點已經(jīng)處于遷移狀態(tài)則跳過。
如果以上均不符合,說明i合法,且i處擁有實際的數(shù)據(jù)節(jié)點,那么使用i處的節(jié)點f作為鎖通過synchronized來達成線程安全的擴容,因為putVal過程中也是用i處的節(jié)點f作為鎖進行synchronized的,意味著對i處的操作擴容和賦值只有一個過程能操作,以此來保證putVal和transfer的并發(fā)安全性。synchronized內(nèi)部的擴容過程因為已經(jīng)保證了線程同步,因此和HashMap的擴容過程區(qū)別并不大,這里不再描述,感興趣的話可以查看9: 從源碼看HashMap:一文看懂HashMap了解。

8 輔助擴容-helpTransfer

5 新數(shù)據(jù)載入-putVal 中以及7 進行擴容-transfer中都提到過如果putVal過程中發(fā)現(xiàn)目標節(jié)點是一個中繼節(jié)點ForwardingNode(hash值為特殊值MOVED)那么putVal會暫停put操作,通過helpTransfer方法進行輔助擴容:

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) {
		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;
}

解析: 這個方法是不是看起來很熟悉,是的,除了ForwardingNode判斷,其主要邏輯和6 維護與啟動擴容-addCount判定擴容時的邏輯大致相當。因此也不在額外表述。這里貼出來的目的僅僅是為了完整的體現(xiàn)putVal遭遇擴容并發(fā)的處理過程。

以上就是java核心基礎(chǔ)之從源碼看ConcurrentHashMap的詳細內(nèi)容,更多關(guān)于JAVA核心知識之從源碼看ConcurrentHashMap的資料請關(guān)注W3Cschool其它相關(guān)文章!


0 人點贊