欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

Java7和Java8中的ConcurrentHashMap原理解析

 更新時(shí)間:2021年04月23日 14:41:17   作者:JaJian  
這篇文章主要介紹了Java7和Java8中的ConcurrentHashMap原理解析,對(duì)ConcurrentHashMap感興趣的讀者,一定要好好看一下

Java7 中 ConcurrentHashMap

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

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

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

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

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

初始化

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

loadFactor:負(fù)載因子,之前我們說了,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 arguments
    int sshift = 0;
    int ssize = 1;
    // 計(jì)算并行級(jí)別 ssize,因?yàn)橐3植⑿屑?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() 無參構(gòu)造函數(shù)進(jìn)行初始化的,那么初始化完成后:

  • Segment 數(shù)組長度為 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,姑且把它們簡單翻譯為移位數(shù)和掩碼,這兩個(gè)值馬上就會(huì)用到

put 過程分析

我們先看 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 位,無符號(hào)右移 segmentShift(28) 位,剩下高 4 位,
    //    然后和 segmentMask(15) 做一次與操作,也就是說 j 是 hash 值的高 4 位,也就是槽的數(shù)組下標(biāo)
    int j = (hash >>> segmentShift) & segmentMask;
    // 剛剛說了,初始化的時(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 ensureSegment
        s = ensureSegment(j);
    // 3. 插入新值到 槽 s 中
    return s.put(key, hash, value, false);
}

第一層皮很簡單,根據(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)雖然很長,不過也很好理解,想想該位置沒有任何元素和已經(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ān)系。
                // 如果不為 null,那就直接將它設(shè)置為鏈表表頭;如果是null,初始化并設(shè)置為鏈表表頭。
                if (node != null)
                    node.setNext(first);
                else
                    node = new HashEntry<K,V>(hash, key, value, first);

                int c = count + 1;
                // 如果超過了該 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;
}

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

到這里 put 操作就結(jié)束了,接下來,我們說一說其中幾步關(guān)鍵的操作。

初始化槽: ensureSegment

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

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

private Segment<K,V> ensureSegment(int k) {
    final Segment<K,V>[] ss = this.segments;
    long u = (k << SSHIFT) + SBASE; // raw offset
    Segment<K,V> seg;
    if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {
        // 這里看到為什么之前要初始化 segment[0] 了,
        // 使用當(dāng)前 segment[0] 處的數(shù)組長度和負(fù)載因子來初始化 segment[k]
        // 為什么要用“當(dāng)前”,因?yàn)?segment[0] 可能早就擴(kuò)容過了
        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;
}

總的來說,ensureSegment(int k) 比較簡單,對(duì)于并發(fā)操作使用 CAS 進(jìn)行控制。

我沒搞懂這里為什么要搞一個(gè) while 循環(huán),CAS 失敗不就代表有其他線程成功了嗎,為什么要再進(jìn)行判斷?
感謝評(píng)論區(qū)的李子木,如果當(dāng)前線程 CAS 失敗,這里的 while 循環(huán)是為了將 seg 賦值返回。

獲取寫入鎖: scanAndLockForPut

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

下面我們來具體分析這個(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 below
        if (retries < 0) {
            if (e == null) {
                if (node == null) // speculatively create node
                    // 進(jìn)到這里說明數(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ù)如果超過 MAX_SCAN_RETRIES(單核1多核64),那么不搶了,進(jìn)入到阻塞隊(duì)列等待鎖
        //    lock() 是阻塞方法,直到獲取鎖后返回
        else if (++retries > MAX_SCAN_RETRIES) {
            lock();
            break;
        }
        else if ((retries & 1) == 0 &&
                 // 這個(gè)時(shí)候是有大問題了,那就是有新的元素進(jìn)到了鏈表,成為了新的表頭
                 //     所以這邊的策略是,相當(dāng)于重新走一遍這個(gè) scanAndLockForPut 方法
                 (f = entryForHash(this, hash)) != first) {
            e = first = f; // re-traverse if entry changed
            retries = -1;
        }
    }
    return node;
}

這個(gè)方法有兩個(gè)出口,一個(gè)是 tryLock() 成功了,循環(huán)終止,另一個(gè)就是重試次數(shù)超過了 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ò)容后,容量為原來的 2 倍。

首先,我們要回顧一下觸發(fā)擴(kuò)容的地方,put 的時(shí)候,如果判斷該值的插入會(huì)導(dǎo)致該 segment 的元素個(gè)數(shù)超過閾值,那么先進(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ù)組長度為 16,e 在 oldTable[3] 處,那么 idx 只可能是 3 或者是 3 + 16 = 19
            int 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);
                }
            }
        }
    }
    // 將新來的 node 放到新數(shù)組中剛剛的 兩個(gè)鏈表之一 的 頭部
    int nodeIndex = node.hash & sizeMask; // add the new node
    node.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)下來,如果 lastRun 的后面還有比較多的節(jié)點(diǎn),那么這次就是值得的。因?yàn)槲覀冎恍枰寺?lastRun 前面的節(jié)點(diǎn),后面的一串節(jié)點(diǎn)跟著 lastRun 走就是了,不需要做任何操作。

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

get 過程分析

相對(duì)于 put 來說,get 真的不要太簡單。

  • 計(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 overhead
    HashEntry<K,V>[] tab;
    // 1. hash 值
    int h = hash(key);
    long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
    // 2. 根據(jù) hash 找到對(duì)應(yīng)的 segment
    if ((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ā)問題分析

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

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

put 操作的線程安全性

  • 初始化槽,這個(gè)我們之前就說過了,使用了 CAS 來初始化 Segment 中的數(shù)組。
  • 添加節(jié)點(diǎn)到鏈表的操作是插入到表頭的,所以,如果這個(gè)時(shí)候 get 操作在鏈表遍歷的過程已經(jīng)到了中間,是不會(huì)影響的。當(dāng)然,另一個(gè)并發(fā)問題就是 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)鍵字。

remove 操作的線程安全性

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

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

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

如果 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 來操作數(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 的。

Java8 ConcurrentHashMap

Java7 中實(shí)現(xiàn)的 ConcurrentHashMap 說實(shí)話還是比較復(fù)雜的,Java8 對(duì) ConcurrentHashMap 進(jìn)行了比較大的改動(dòng)。建議讀者可以參考 Java8 中 HashMap 相對(duì)于 Java7 HashMap 的改動(dòng),對(duì)于 ConcurrentHashMap,Java8 也引入了紅黑樹。

說實(shí)話,Java8 ConcurrentHashMap 源碼真心不簡單,最難的在于擴(kuò)容,數(shù)據(jù)遷移操作不容易看懂。

我們先用一個(gè)示意圖來描述下其結(jié)構(gòu):

結(jié)構(gòu)上和 Java8 的 HashMap 基本上一樣,不過它要保證線程安全性,所以在源碼上確實(shí)要復(fù)雜一些。

初始化

// 這構(gòu)造函數(shù)里,什么都不干
public ConcurrentHashMap() {
}
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;
}

這個(gè)初始化方法有點(diǎn)意思,通過提供初始容量,計(jì)算了 sizeCtl,sizeCtl = 【 (1.5 * initialCapacity + 1),然后向上取最近的 2 的 n 次方】。如 initialCapacity 為 10,那么得到 sizeCtl 為 16,如果 initialCapacity 為 11,得到 sizeCtl 為 32。

sizeCtl 這個(gè)屬性使用的場景很多,不過只要跟著文章的思路來,就不會(huì)被它搞暈了。

如果你愛折騰,也可以看下另一個(gè)有三個(gè)參數(shù)的構(gòu)造方法,這里我就不說了,大部分時(shí)候,我們會(huì)使用無參構(gòu)造函數(shù)進(jìn)行實(shí)例化,我們也按照這個(gè)思路來進(jìn)行源碼分析吧。

put 過程分析

仔細(xì)地一行一行代碼看下去:

public V put(K key, V value) {
    return putVal(key, value, false);
}
final V putVal(K key, V value, boolean onlyIfAbsent) {
    if (key == null || value == null) throw new NullPointerException();
    // 得到 hash 值
    int hash = spread(key.hashCode());
    // 用于記錄相應(yīng)鏈表的長度
    int binCount = 0;
    for (Node<K,V>[] tab = table;;) {
        Node<K,V> f; int n, i, fh;
        // 如果數(shù)組"空",進(jìn)行數(shù)組初始化
        if (tab == null || (n = tab.length) == 0)
            // 初始化數(shù)組,后面會(huì)詳細(xì)介紹
            tab = initTable();

        // 找該 hash 值對(duì)應(yīng)的數(shù)組下標(biāo),得到第一個(gè)節(jié)點(diǎn) f
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
            // 如果數(shù)組該位置為空,
            //    用一次 CAS 操作將這個(gè)新值放入其中即可,這個(gè) put 操作差不多就結(jié)束了,可以拉到最后面了
            //          如果 CAS 失敗,那就是有并發(fā)操作,進(jìn)到下一個(gè)循環(huán)就好了
            if (casTabAt(tab, i, null,
                         new Node<K,V>(hash, key, value, null)))
                break;                   // no lock when adding to empty bin
        }
        // hash 居然可以等于 MOVED,這個(gè)需要到后面才能看明白,不過從名字上也能猜到,肯定是因?yàn)樵跀U(kuò)容
        else if ((fh = f.hash) == MOVED)
            // 幫助數(shù)據(jù)遷移,這個(gè)等到看完數(shù)據(jù)遷移部分的介紹后,再理解這個(gè)就很簡單了
            tab = helpTransfer(tab, f);

        else { // 到這里就是說,f 是該位置的頭結(jié)點(diǎn),而且不為空

            V oldVal = null;
            // 獲取數(shù)組該位置的頭結(jié)點(diǎn)的監(jiān)視器鎖
            synchronized (f) {
                if (tabAt(tab, i) == f) {
                    if (fh >= 0) { // 頭結(jié)點(diǎn)的 hash 值大于 0,說明是鏈表
                        // 用于累加,記錄鏈表的長度
                        binCount = 1;
                        // 遍歷鏈表
                        for (Node<K,V> e = f;; ++binCount) {
                            K ek;
                            // 如果發(fā)現(xiàn)了"相等"的 key,判斷是否要進(jìn)行值覆蓋,然后也就可以 break 了
                            if (e.hash == hash &&
                                ((ek = e.key) == key ||
                                 (ek != null && key.equals(ek)))) {
                                oldVal = e.val;
                                if (!onlyIfAbsent)
                                    e.val = value;
                                break;
                            }
                            // 到了鏈表的最末端,將這個(gè)新值放到鏈表的最后面
                            Node<K,V> pred = e;
                            if ((e = e.next) == null) {
                                pred.next = new Node<K,V>(hash, key,
                                                          value, null);
                                break;
                            }
                        }
                    }
                    else if (f instanceof TreeBin) { // 紅黑樹
                        Node<K,V> p;
                        binCount = 2;
                        // 調(diào)用紅黑樹的插值方法插入新節(jié)點(diǎn)
                        if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
                                                       value)) != null) {
                            oldVal = p.val;
                            if (!onlyIfAbsent)
                                p.val = value;
                        }
                    }
                }
            }

            if (binCount != 0) {
                // 判斷是否要將鏈表轉(zhuǎn)換為紅黑樹,臨界值和 HashMap 一樣,也是 8
                if (binCount >= TREEIFY_THRESHOLD)
                    // 這個(gè)方法和 HashMap 中稍微有一點(diǎn)點(diǎn)不同,那就是它不是一定會(huì)進(jìn)行紅黑樹轉(zhuǎn)換,
                    // 如果當(dāng)前數(shù)組的長度小于 64,那么會(huì)選擇進(jìn)行數(shù)組擴(kuò)容,而不是轉(zhuǎn)換為紅黑樹
                    //    具體源碼我們就不看了,擴(kuò)容部分后面說
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    // 
    addCount(1L, binCount);
    return null;
}

put 的主流程看完了,但是至少留下了幾個(gè)問題,第一個(gè)是初始化,第二個(gè)是擴(kuò)容,第三個(gè)是幫助數(shù)據(jù)遷移,這些我們都會(huì)在后面進(jìn)行一一介紹。

初始化數(shù)組:initTable

這個(gè)比較簡單,主要就是初始化一個(gè)合適大小的數(shù)組,然后會(huì)設(shè)置 sizeCtl。

初始化方法中的并發(fā)問題是通過對(duì) sizeCtl 進(jìn)行一個(gè) CAS 操作來控制的。

private final Node<K,V>[] initTable() {
    Node<K,V>[] tab; int sc;
    while ((tab = table) == null || tab.length == 0) {
        // 初始化的"功勞"被其他線程"搶去"了
        if ((sc = sizeCtl) < 0)
            Thread.yield(); // lost initialization race; just spin
        // CAS 一下,將 sizeCtl 設(shè)置為 -1,代表搶到了鎖
        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
            try {
                if ((tab = table) == null || tab.length == 0) {
                    // DEFAULT_CAPACITY 默認(rèn)初始容量是 16
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                    // 初始化數(shù)組,長度為 16 或初始化時(shí)提供的長度
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                    // 將這個(gè)數(shù)組賦值給 table,table 是 volatile 的
                    table = tab = nt;
                    // 如果 n 為 16 的話,那么這里 sc = 12
                    // 其實(shí)就是 0.75 * n
                    sc = n - (n >>> 2);
                }
            } finally {
                // 設(shè)置 sizeCtl 為 sc,我們就當(dāng)是 12 吧
                sizeCtl = sc;
            }
            break;
        }
    }
    return tab;
}

鏈表轉(zhuǎn)紅黑樹: treeifyBin

前面我們?cè)?put 源碼分析也說過,treeifyBin 不一定就會(huì)進(jìn)行紅黑樹轉(zhuǎn)換,也可能是僅僅做數(shù)組擴(kuò)容。我們還是進(jìn)行源碼分析吧。

private final void treeifyBin(Node<K,V>[] tab, int index) {
    Node<K,V> b; int n, sc;
    if (tab != null) {
        // MIN_TREEIFY_CAPACITY 為 64
        // 所以,如果數(shù)組長度小于 64 的時(shí)候,其實(shí)也就是 32 或者 16 或者更小的時(shí)候,會(huì)進(jìn)行數(shù)組擴(kuò)容
        if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
            // 后面我們?cè)僭敿?xì)分析這個(gè)方法
            tryPresize(n << 1);
        // b 是頭結(jié)點(diǎn)
        else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
            // 加鎖
            synchronized (b) {

                if (tabAt(tab, index) == b) {
                    // 下面就是遍歷鏈表,建立一顆紅黑樹
                    TreeNode<K,V> hd = null, tl = null;
                    for (Node<K,V> e = b; e != null; e = e.next) {
                        TreeNode<K,V> p =
                            new TreeNode<K,V>(e.hash, e.key, e.val,
                                              null, null);
                        if ((p.prev = tl) == null)
                            hd = p;
                        else
                            tl.next = p;
                        tl = p;
                    }
                    // 將紅黑樹設(shè)置到數(shù)組相應(yīng)位置中
                    setTabAt(tab, index, new TreeBin<K,V>(hd));
                }
            }
        }
    }
}

擴(kuò)容:tryPresize

如果說 Java8 ConcurrentHashMap 的源碼不簡單,那么說的就是擴(kuò)容操作和遷移操作。

這個(gè)方法要完完全全看懂還需要看之后的 transfer 方法,讀者應(yīng)該提前知道這點(diǎn)。

這里的擴(kuò)容也是做翻倍擴(kuò)容的,擴(kuò)容后數(shù)組容量為原來的 2 倍。

// 首先要說明的是,方法參數(shù) size 傳進(jìn)來的時(shí)候就已經(jīng)翻了倍了
private final void tryPresize(int size) {
    // c:size 的 1.5 倍,再加 1,再往上取最近的 2 的 n 次方。
    int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :
        tableSizeFor(size + (size >>> 1) + 1);
    int sc;
    while ((sc = sizeCtl) >= 0) {
        Node<K,V>[] tab = table; int n;

        // 這個(gè) if 分支和之前說的初始化數(shù)組的代碼基本上是一樣的,在這里,我們可以不用管這塊代碼
        if (tab == null || (n = tab.length) == 0) {
            n = (sc > c) ? sc : c;
            if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
                try {
                    if (table == tab) {
                        @SuppressWarnings("unchecked")
                        Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                        table = nt;
                        sc = n - (n >>> 2); // 0.75 * n
                    }
                } finally {
                    sizeCtl = sc;
                }
            }
        }
        else if (c <= sc || n >= MAXIMUM_CAPACITY)
            break;
        else if (tab == table) {
            // 我沒看懂 rs 的真正含義是什么,不過也關(guān)系不大
            int rs = resizeStamp(n);

            if (sc < 0) {
                Node<K,V>[] nt;
                if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                    sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                    transferIndex <= 0)
                    break;
                // 2. 用 CAS 將 sizeCtl 加 1,然后執(zhí)行 transfer 方法
                //    此時(shí) nextTab 不為 null
                if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                    transfer(tab, nt);
            }
            // 1. 將 sizeCtl 設(shè)置為 (rs << RESIZE_STAMP_SHIFT) + 2)
            //     我是沒看懂這個(gè)值真正的意義是什么?不過可以計(jì)算出來的是,結(jié)果是一個(gè)比較大的負(fù)數(shù)
            //  調(diào)用 transfer 方法,此時(shí) nextTab 參數(shù)為 null
            else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                         (rs << RESIZE_STAMP_SHIFT) + 2))
                transfer(tab, null);
        }
    }
}

這個(gè)方法的核心在于 sizeCtl 值的操作,首先將其設(shè)置為一個(gè)負(fù)數(shù),然后執(zhí)行 transfer(tab, null),再下一個(gè)循環(huán)將 sizeCtl 加 1,并執(zhí)行 transfer(tab, nt),之后可能是繼續(xù) sizeCtl 加 1,并執(zhí)行 transfer(tab, nt)。

所以,可能的操作就是執(zhí)行 1 次 transfer(tab, null) + 多次 transfer(tab, nt),這里怎么結(jié)束循環(huán)的需要看完 transfer 源碼才清楚。

數(shù)據(jù)遷移:transfer

下面這個(gè)方法很點(diǎn)長,將原來的 tab 數(shù)組的元素遷移到新的 nextTab 數(shù)組中。

雖然我們之前說的 tryPresize 方法中多次調(diào)用 transfer 不涉及多線程,但是這個(gè) transfer 方法可以在其他地方被調(diào)用,典型地,我們之前在說 put 方法的時(shí)候就說過了,請(qǐng)往上看 put 方法,是不是有個(gè)地方調(diào)用了 helpTransfer 方法,helpTransfer 方法會(huì)調(diào)用 transfer 方法的。

此方法支持多線程執(zhí)行,外圍調(diào)用此方法的時(shí)候,會(huì)保證第一個(gè)發(fā)起數(shù)據(jù)遷移的線程,nextTab 參數(shù)為 null,之后再調(diào)用此方法的時(shí)候,nextTab 不會(huì)為 null。

閱讀源碼之前,先要理解并發(fā)操作的機(jī)制。原數(shù)組長度為 n,所以我們有 n 個(gè)遷移任務(wù),讓每個(gè)線程每次負(fù)責(zé)一個(gè)小任務(wù)是最簡單的,每做完一個(gè)任務(wù)再檢測是否有其他沒做完的任務(wù),幫助遷移就可以了,而 Doug Lea 使用了一個(gè) stride,簡單理解就是步長,每個(gè)線程每次負(fù)責(zé)遷移其中的一部分,如每次遷移 16 個(gè)小任務(wù)。所以,我們就需要一個(gè)全局的調(diào)度者來安排哪個(gè)線程執(zhí)行哪幾個(gè)任務(wù),這個(gè)就是屬性 transferIndex 的作用。

第一個(gè)發(fā)起數(shù)據(jù)遷移的線程會(huì)將 transferIndex 指向原數(shù)組最后的位置,然后從后往前的 stride 個(gè)任務(wù)屬于第一個(gè)線程,然后將 transferIndex 指向新的位置,再往前的 stride 個(gè)任務(wù)屬于第二個(gè)線程,依此類推。當(dāng)然,這里說的第二個(gè)線程不是真的一定指代了第二個(gè)線程,也可以是同一個(gè)線程,這個(gè)讀者應(yīng)該能理解吧。其實(shí)就是將一個(gè)大的遷移任務(wù)分為了一個(gè)個(gè)任務(wù)包。

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
    int n = tab.length, stride;

    // stride 在單核下直接等于 n,多核模式下為 (n>>>3)/NCPU,最小值是 16
    // stride 可以理解為”步長“,有 n 個(gè)位置是需要進(jìn)行遷移的,
    //   將這 n 個(gè)任務(wù)分為多個(gè)任務(wù)包,每個(gè)任務(wù)包有 stride 個(gè)任務(wù)
    if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
        stride = MIN_TRANSFER_STRIDE; // subdivide range

    // 如果 nextTab 為 null,先進(jìn)行一次初始化
    //    前面我們說了,外圍會(huì)保證第一個(gè)發(fā)起遷移的線程調(diào)用此方法時(shí),參數(shù) nextTab 為 null
    //       之后參與遷移的線程調(diào)用此方法時(shí),nextTab 不會(huì)為 null
    if (nextTab == null) {
        try {
            // 容量翻倍
            Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
            nextTab = nt;
        } catch (Throwable ex) {      // try to cope with OOME
            sizeCtl = Integer.MAX_VALUE;
            return;
        }
        // nextTable 是 ConcurrentHashMap 中的屬性
        nextTable = nextTab;
        // transferIndex 也是 ConcurrentHashMap 的屬性,用于控制遷移的位置
        transferIndex = n;
    }

    int nextn = nextTab.length;

    // ForwardingNode 翻譯過來就是正在被遷移的 Node
    // 這個(gè)構(gòu)造方法會(huì)生成一個(gè)Node,key、value 和 next 都為 null,關(guān)鍵是 hash 為 MOVED
    // 后面我們會(huì)看到,原數(shù)組中位置 i 處的節(jié)點(diǎn)完成遷移工作后,
    //    就會(huì)將位置 i 處設(shè)置為這個(gè) ForwardingNode,用來告訴其他線程該位置已經(jīng)處理過了
    //    所以它其實(shí)相當(dāng)于是一個(gè)標(biāo)志。
    ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);


    // advance 指的是做完了一個(gè)位置的遷移工作,可以準(zhǔn)備做下一個(gè)位置的了
    boolean advance = true;
    boolean finishing = false; // to ensure sweep before committing nextTab

    /*
     * 下面這個(gè) for 循環(huán),最難理解的在前面,而要看懂它們,應(yīng)該先看懂后面的,然后再倒回來看
     * 
     */

    // i 是位置索引,bound 是邊界,注意是從后往前
    for (int i = 0, bound = 0;;) {
        Node<K,V> f; int fh;

        // 下面這個(gè) while 真的是不好理解
        // advance 為 true 表示可以進(jìn)行下一個(gè)位置的遷移了
        //   簡單理解結(jié)局:i 指向了 transferIndex,bound 指向了 transferIndex-stride
        while (advance) {
            int nextIndex, nextBound;
            if (--i >= bound || finishing)
                advance = false;

            // 將 transferIndex 值賦給 nextIndex
            // 這里 transferIndex 一旦小于等于 0,說明原數(shù)組的所有位置都有相應(yīng)的線程去處理了
            else if ((nextIndex = transferIndex) <= 0) {
                i = -1;
                advance = false;
            }
            else if (U.compareAndSwapInt
                     (this, TRANSFERINDEX, nextIndex,
                      nextBound = (nextIndex > stride ?
                                   nextIndex - stride : 0))) {
                // 看括號(hào)中的代碼,nextBound 是這次遷移任務(wù)的邊界,注意,是從后往前
                bound = nextBound;
                i = nextIndex - 1;
                advance = false;
            }
        }
        if (i < 0 || i >= n || i + n >= nextn) {
            int sc;
            if (finishing) {
                // 所有的遷移操作已經(jīng)完成
                nextTable = null;
                // 將新的 nextTab 賦值給 table 屬性,完成遷移
                table = nextTab;
                // 重新計(jì)算 sizeCtl:n 是原數(shù)組長度,所以 sizeCtl 得出的值將是新數(shù)組長度的 0.75 倍
                sizeCtl = (n << 1) - (n >>> 1);
                return;
            }

            // 之前我們說過,sizeCtl 在遷移前會(huì)設(shè)置為 (rs << RESIZE_STAMP_SHIFT) + 2
            // 然后,每有一個(gè)線程參與遷移就會(huì)將 sizeCtl 加 1,
            // 這里使用 CAS 操作對(duì) sizeCtl 進(jìn)行減 1,代表做完了屬于自己的任務(wù)
            if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
                // 任務(wù)結(jié)束,方法退出
                if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
                    return;

                // 到這里,說明 (sc - 2) == resizeStamp(n) << RESIZE_STAMP_SHIFT,
                // 也就是說,所有的遷移任務(wù)都做完了,也就會(huì)進(jìn)入到上面的 if(finishing){} 分支了
                finishing = advance = true;
                i = n; // recheck before commit
            }
        }
        // 如果位置 i 處是空的,沒有任何節(jié)點(diǎn),那么放入剛剛初始化的 ForwardingNode ”空節(jié)點(diǎn)“
        else if ((f = tabAt(tab, i)) == null)
            advance = casTabAt(tab, i, null, fwd);
        // 該位置處是一個(gè) ForwardingNode,代表該位置已經(jīng)遷移過了
        else if ((fh = f.hash) == MOVED)
            advance = true; // already processed
        else {
            // 對(duì)數(shù)組該位置處的結(jié)點(diǎn)加鎖,開始處理數(shù)組該位置處的遷移工作
            synchronized (f) {
                if (tabAt(tab, i) == f) {
                    Node<K,V> ln, hn;
                    // 頭結(jié)點(diǎn)的 hash 大于 0,說明是鏈表的 Node 節(jié)點(diǎn)
                    if (fh >= 0) {
                        // 下面這一塊和 Java7 中的 ConcurrentHashMap 遷移是差不多的,
                        // 需要將鏈表一分為二,
                        //   找到原鏈表中的 lastRun,然后 lastRun 及其之后的節(jié)點(diǎn)是一起進(jìn)行遷移的
                        //   lastRun 之前的節(jié)點(diǎn)需要進(jìn)行克隆,然后分到兩個(gè)鏈表中
                        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) {
                                runBit = b;
                                lastRun = p;
                            }
                        }
                        if (runBit == 0) {
                            ln = lastRun;
                            hn = null;
                        }
                        else {
                            hn = lastRun;
                            ln = null;
                        }
                        for (Node<K,V> p = f; p != lastRun; p = p.next) {
                            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);
                        }
                        // 其中的一個(gè)鏈表放在新數(shù)組的位置 i
                        setTabAt(nextTab, i, ln);
                        // 另一個(gè)鏈表放在新數(shù)組的位置 i+n
                        setTabAt(nextTab, i + n, hn);
                        // 將原數(shù)組該位置處設(shè)置為 fwd,代表該位置已經(jīng)處理完畢,
                        //    其他線程一旦看到該位置的 hash 值為 MOVED,就不會(huì)進(jìn)行遷移了
                        setTabAt(tab, i, fwd);
                        // advance 設(shè)置為 true,代表該位置已經(jīng)遷移完畢
                        advance = true;
                    }
                    else if (f instanceof TreeBin) {
                        // 紅黑樹的遷移
                        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;
                            }
                        }
                        // 如果一分為二后,節(jié)點(diǎn)數(shù)少于 8,那么將紅黑樹轉(zhuǎn)換回鏈表
                        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;

                        // 將 ln 放置在新數(shù)組的位置 i
                        setTabAt(nextTab, i, ln);
                        // 將 hn 放置在新數(shù)組的位置 i+n
                        setTabAt(nextTab, i + n, hn);
                        // 將原數(shù)組該位置處設(shè)置為 fwd,代表該位置已經(jīng)處理完畢,
                        //    其他線程一旦看到該位置的 hash 值為 MOVED,就不會(huì)進(jìn)行遷移了
                        setTabAt(tab, i, fwd);
                        // advance 設(shè)置為 true,代表該位置已經(jīng)遷移完畢
                        advance = true;
                    }
                }
            }
        }
    }
}

說到底,transfer 這個(gè)方法并沒有實(shí)現(xiàn)所有的遷移任務(wù),每次調(diào)用這個(gè)方法只實(shí)現(xiàn)了 transferIndex 往前 stride 個(gè)位置的遷移工作,其他的需要由外圍來控制。

這個(gè)時(shí)候,再回去仔細(xì)看 tryPresize 方法可能就會(huì)更加清晰一些了。

get 過程分析

get 方法從來都是最簡單的,這里也不例外:

1.計(jì)算 hash 值

2.根據(jù) hash 值找到數(shù)組對(duì)應(yīng)位置: (n - 1) & h

3.根據(jù)該位置處結(jié)點(diǎn)性質(zhì)進(jìn)行相應(yīng)查找

  • 如果該位置為 null,那么直接返回 null 就可以了
  • 如果該位置處的節(jié)點(diǎn)剛好就是我們需要的,返回該節(jié)點(diǎn)的值即可
  • 如果該位置節(jié)點(diǎn)的 hash 值小于 0,說明正在擴(kuò)容,或者是紅黑樹,后面我們?cè)俳榻B find 方法
  • 如果以上 3 條都不滿足,那就是鏈表,進(jìn)行遍歷比對(duì)即可
public V get(Object key) {
    Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
    int h = spread(key.hashCode());
    if ((tab = table) != null && (n = tab.length) > 0 &&
        (e = tabAt(tab, (n - 1) & h)) != null) {
        // 判斷頭結(jié)點(diǎn)是否就是我們需要的節(jié)點(diǎn)
        if ((eh = e.hash) == h) {
            if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                return e.val;
        }
        // 如果頭結(jié)點(diǎn)的 hash 小于 0,說明 正在擴(kuò)容,或者該位置是紅黑樹
        else if (eh < 0)
            // 參考 ForwardingNode.find(int h, Object k) 和 TreeBin.find(int h, Object k)
            return (p = e.find(h, key)) != null ? p.val : null;

        // 遍歷鏈表
        while ((e = e.next) != null) {
            if (e.hash == h &&
                ((ek = e.key) == key || (ek != null && key.equals(ek))))
                return e.val;
        }
    }
    return null;
}

簡單說一句,此方法的大部分內(nèi)容都很簡單,只有正好碰到擴(kuò)容的情況,F(xiàn)orwardingNode.find(int h, Object k) 稍微復(fù)雜一些,不過在了解了數(shù)據(jù)遷移的過程后,這個(gè)也就不難了,所以限于篇幅這里也不展開說了。

以上就是Java7和Java8中的ConcurrentHashMap原理解析的詳細(xì)內(nèi)容,更多關(guān)于ConcurrentHashMap原理的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章!

相關(guān)文章

  • Java多線程通信:交替打印ABAB實(shí)例

    Java多線程通信:交替打印ABAB實(shí)例

    這篇文章主要介紹了Java多線程通信:交替打印ABAB實(shí)例,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過來看看吧
    2020-08-08
  • 如何利用JAVA實(shí)現(xiàn)走迷宮程序

    如何利用JAVA實(shí)現(xiàn)走迷宮程序

    最近經(jīng)常在機(jī)房看同學(xué)在玩一個(gè)走迷宮的游戲,比較有趣,自己也用java實(shí)現(xiàn)了一個(gè),這篇文章主要給大家介紹了關(guān)于如何利用JAVA實(shí)現(xiàn)走迷宮程序的相關(guān)資料,需要的朋友可以參考下
    2021-06-06
  • IDEA集成Docker實(shí)現(xiàn)快捷部署的操作步驟

    IDEA集成Docker實(shí)現(xiàn)快捷部署的操作步驟

    這篇文章主要介紹了IDEA集成Docker實(shí)現(xiàn)快捷部署的操作步驟,通過靈活利用這一功能,開發(fā)人員可以更快速地開發(fā)、調(diào)試和部署應(yīng)用程序,從而提高開發(fā)工作的效率和質(zhì)量,需要的朋友可以參考下
    2024-06-06
  • Java 如何使用Feign發(fā)送HTTP請(qǐng)求

    Java 如何使用Feign發(fā)送HTTP請(qǐng)求

    這篇文章主要介紹了Java 如何使用Feign發(fā)送HTTP請(qǐng)求,幫助大家更好的理解和學(xué)習(xí)Java,感興趣的朋友可以了解下
    2020-11-11
  • 淺談Java設(shè)計(jì)模式之原型模式知識(shí)總結(jié)

    淺談Java設(shè)計(jì)模式之原型模式知識(shí)總結(jié)

    Java原型模式主要用于創(chuàng)建重復(fù)的對(duì)象,同時(shí)又能保證性能,這篇文章就帶大家仔細(xì)了解一下原型模式的知識(shí),對(duì)正在學(xué)習(xí)java的小伙伴們很有幫助,需要的朋友可以參考下
    2021-05-05
  • 探索Java I/O 模型的演進(jìn)

    探索Java I/O 模型的演進(jìn)

    什么是同步?什么是異步?阻塞和非阻塞又有什么區(qū)別?本文先從 Unix 的 I/O 模型講起,介紹了5種常見的 I/O 模型。而后再引出 Java 的 I/O 模型的演進(jìn)過程,并用實(shí)例說明如何選擇合適的 Java I/O 模型來提高系統(tǒng)的并發(fā)量和可用性。,需要的朋友可以參考下
    2019-06-06
  • java多線程中執(zhí)行多個(gè)程序的實(shí)例分析

    java多線程中執(zhí)行多個(gè)程序的實(shí)例分析

    在本篇文章里小編給大家整理的是一篇關(guān)于java多線程中執(zhí)行多個(gè)程序的實(shí)例分析內(nèi)容,有需要的朋友們可以學(xué)習(xí)參考下。
    2021-02-02
  • SpringBoot使用Shiro實(shí)現(xiàn)動(dòng)態(tài)加載權(quán)限詳解流程

    SpringBoot使用Shiro實(shí)現(xiàn)動(dòng)態(tài)加載權(quán)限詳解流程

    本文小編將基于?SpringBoot?集成?Shiro?實(shí)現(xiàn)動(dòng)態(tài)uri權(quán)限,由前端vue在頁面配置uri,Java后端動(dòng)態(tài)刷新權(quán)限,不用重啟項(xiàng)目,以及在頁面分配給用戶?角色?、?按鈕?、uri?權(quán)限后,后端動(dòng)態(tài)分配權(quán)限,用戶無需在頁面重新登錄才能獲取最新權(quán)限,一切權(quán)限動(dòng)態(tài)加載,靈活配置
    2022-07-07
  • Elasticsearch算分優(yōu)化方案之rescore_query示例詳解

    Elasticsearch算分優(yōu)化方案之rescore_query示例詳解

    這篇文章主要為大家介紹了Elasticsearch算分優(yōu)化方案之rescore_query示例詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪
    2023-08-08
  • 詳解Spring中bean實(shí)例化的三種方式

    詳解Spring中bean實(shí)例化的三種方式

    本篇文章主要介紹了詳解Spring中bean實(shí)例化的三種方式,小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,也給大家做個(gè)參考。一起跟隨小編過來看看吧
    2017-04-04

最新評(píng)論