簡單總結ConcurrentHashMap

時間 2020-07-07

原文原文鏈接

1、HashTablehtml

hashTable是一個線程安全的容器，是線程安全版本的HashMap。但它的底層是和HashMap同樣的，只是在方法上都加上了synchronized關鍵字。node

這樣子有什麼後果呢：算法

效率及低，意味着每一個線程在執行HashTable的方法的時候，或者說操縱HashTable的時候，都要鎖住整個對象。也就是讓並行併發的訪問，變成了串行。
複合操做會有線程安全問題。由於它是每一個方法都加鎖了，這意味着在執行單個方法像put，contains方法的時候，是能夠保證原子性的，但若是是執行一個複合操做的時候，就不保證了。

if(!table.contains("key")) { map.put("key", object); }

相似於這樣的方法，當線程1在執行if裏面的判斷的時候，線程1會得到table實例的所，其餘線程沒法訪問table的其餘同步方法。但當線程1判斷完if後，鎖會放掉，這個時候若是線程2進來，得到table實例的鎖，而後put了一個」key「進來，而後再放鎖；那麼線程1再執行put方法就不對了。（它原本是覺得沒有這個key再put的）數組

2、concurrentHashMap1.7安全

併發思路多線程

concurrenthashMap是採用一個叫作分段所的機制。併發

它能夠看做是一個二重hashMap，首先concurrentHashMap是一個segment數組，每一個segment都是一個繼承了ReentrantLock的類，這樣就能夠方便地在各個segment裏面加鎖因此每次須要加鎖的操做鎖住的是一個 segment，這樣只要保證每一個 Segment 是線程安全的，也就實現了全局的線程安全。ssh

哦哦還要注意，這個最外面的Segment[]數組，是不能夠擴容的！ide

而後進到Segment內部，會發現，每一個Segment能夠看做一個HashMap。也就是在一個Segment裏面，有個HashEntry[]數組，而後這個數組是一個個桶，桶裏面是單向鏈表。函數

（圖片來自：http://www.importnew.com/28263.html）

構造函數

而後咱們經過構造函數進入，順便了解ConcurrentHashMap中重要的field吧。

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; // 計算並行級別 ssize，由於要保持並行級別是 2 的 n 次方
    while (ssize < concurrencyLevel) { ++sshift; ssize <<= 1; } // 咱們這裏先不要那麼燒腦，用默認值，concurrencyLevel 爲 16，sshift 爲 4 // 那麼計算出 segmentShift 爲 28，segmentMask 爲 15，後面會用到這兩個值
    this.segmentShift = 32 - sshift; this.segmentMask = ssize - 1; if (initialCapacity > MAXIMUM_CAPACITY) initialCapacity = MAXIMUM_CAPACITY; // initialCapacity 是設置整個 map 初始的大小， // 這裏根據 initialCapacity 計算 Segment 數組中每一個位置能夠分到的大小 // 如 initialCapacity 爲 64，那麼每一個 Segment 或稱之爲"槽"能夠分到 4 個
    int c = initialCapacity / ssize; if (c * ssize < initialCapacity) ++c; // 默認 MIN_SEGMENT_TABLE_CAPACITY 是 2，這個值也是有講究的，由於這樣的話，對於具體的槽上， // 插入一個元素不至於擴容，插入第二個的時候纔會擴容
    int cap = MIN_SEGMENT_TABLE_CAPACITY; while (cap < c) cap <<= 1; // 建立 Segment 數組， // 並建立數組的第一個元素 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]; // 往數組寫入 segment[0]
    UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
    this.segments = ss; }

initialCapacity和之前同樣，指的是這個ConcurrenthashMap的初始容量，或者說是理解成初始桶的數量。但咱們這個hashmap是有兩重表的嘛，因此在實際操做的時候會把這個值分配給各個Segment，也就至關於間接指定了每一個Segment中應該有幾個桶。

loadFactor和通常的hashTable同樣，負載因子，size/capacity。但上面說了Segment數組是不能夠擴容的，因此這個也是給Segment裏面的數組用的。

concurrencyLevel：concurrencyLevel：並行級別、併發數、Segment 數，怎麼翻譯不重要，理解它。默認是 16，也就是說 ConcurrentHashMap 有 16 個 Segments，因此理論上，這個時候，最多能夠同時支持 16 個線程併發寫，只要它們的操做分別分佈在不一樣的 Segment 上。這個值能夠在初始化的時候設置爲其餘值，可是一旦初始化之後，它是不能夠擴容的。

segmentShift：

這個值=32 - shift，shift是你>=你傳進來的concurrentLevel的一個2次冪數的左移位數。而二次冪的數字，都是10000這樣的嘛，因此shift就是10000中0的個數。

因此field segmentShift我以爲能夠理解成000000100000（32位數字），而後是前面的0加上1的位數就是segmentShift吧。

SegmetnMask：

掩碼嘛，就二次冪處理後的concurrentLevel的長度 - 1，獲得的就相似0111111這樣咯，因此等等用來作與操做用的。

而後最後那個Unsafe的putOrderObject一個不安全的直接操縱內存的方法，應該是由於這樣會快點吧。這個order應該是防止指令重排序的意思。

要了解Unsafe能夠看這篇文章：https://www.cnblogs.com/throwable/p/9139947.html

若是咱們就當是用 new ConcurrentHashMap() 無參構造函數進行初始化的，那麼初始化完成後：

Segment 數組長度爲 16，不能夠擴容
Segment[i] 的默認大小爲 2，負載因子是 0.75，得出初始閾值爲 1.5，也就是之後插入第一個元素不會觸發擴容，插入第二個會進行第一次擴容
這裏初始化了 segment[0]，其餘位置仍是 null，至於爲何要初始化 segment[0]，後面的代碼會介紹
當前 segmentShift 的值爲 32 – 4 = 28，segmentMask 爲 16 – 1 = 15，姑且把它們簡單翻譯爲移位數和掩碼，這兩個值立刻就會用到

put方法

而後來看重要的put方法。

先看put的主流程：

public V put(K key, V value) { Segment<K,V> s; if (value == null) throw new NullPointerException(); // 1. 計算 key 的 hash 值
    int hash = hash(key); // 2. 根據 hash 值找到 Segment 數組中的位置 j // hash 是 32 位，無符號右移 segmentShift(28) 位，剩下低 4 位， // 而後和 segmentMask(15) 作一次與操做，也就是說 j 是 hash 值的最後 4 位，也就是槽的數組下標
    int j = (hash >>> segmentShift) & segmentMask; // 剛剛說了，初始化的時候初始化了 segment[0]，可是其餘位置仍是 null， // ensureSegment(j) 對 segment[j] 進行初始化
    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); }

這裏的主流程是在第一層表格的操做。就根據key的hash值，找到Segment[]數組的桶序號，而後先初始化這個segment[j]（構造器中只初始化了segment[0]），而後進入這個segmetn[j]，交給這個segment[j]（局部HashMap）繼續執行put操做。

求j的時候，hash值移了segmentShift後，恰好只剩後面四位（默認狀況的話），恰好等於segmentMask15（4位）的位數，而後再相與就獲得一個序號咯。

而後就經過s.put(key, hash, value, false)；進入Segment內部的那個局部Hashmap的put方法。、

先看看這個初始化segment[j]的方法。

ensureSegment(j)：

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] 了， // 使用當前 segment[0] 處的數組長度和負載因子來初始化 segment[k] // 爲何要用「當前」，由於 segment[0] 可能早就擴容過了
        Segment<K,V> proto = ss[0]; int cap = proto.table.length; float lf = proto.loadFactor; int threshold = (int)(cap * lf); // 初始化 segment[k] 內部的數組
        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 循環，內部用 CAS，當前線程成功設值或其餘線程成功設值後，退出
            while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) { if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s)) break; } } } return seg; }

這裏須要考慮併發，由於極可能會有多個線程同時進來初始化同一個槽 segment[k]，不過只要有一個成功了就能夠。

這裏就用構造其中已經初始化好了的segment[0]（也可能已經有元素了）的數據來構造segment[j]咯，而後再用自旋的CAS操做來更新segment數組中的j桶，更新成功或者是有別的線程更新成功都會跳出循環。

再來看segment裏面的局部HashMap的put方法。

Segment裏面的hashMap的put方法：

final V put(K key, int hash, V value, boolean onlyIfAbsent) { // 在往該 segment 寫入前，須要先獲取該 segment 的獨佔鎖 // 先看主流程，後面還會具體介紹這部份內容
    HashEntry<K,V> node = tryLock() ? null : scanAndLockForPut(key, hash, value); V oldValue; try { // 這個是 segment 內部的數組
        HashEntry<K,V>[] tab = table; // 再利用 hash 值，求應該放置的數組下標
        int index = (tab.length - 1) & hash; // first 是數組該位置處的鏈表的表頭
        HashEntry<K,V> first = entryAt(tab, index); // 下面這串 for 循環雖然很長，不過也很好理解，想一想該位置沒有任何元素和已經存在一個鏈表這兩種狀況
        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 { // node 究竟是不是 null，這個要看獲取鎖的過程，不過和這裏都沒有關係。 // 若是不爲 null，那就直接將它設置爲鏈表表頭；若是是null，初始化並設置爲鏈表表頭。
                if (node != null) node.setNext(first); else node = new HashEntry<K,V>(hash, key, value, first); int c = count + 1; // 若是超過了該 segment 的閾值，這個 segment 須要擴容
                if (c > threshold && tab.length < MAXIMUM_CAPACITY) rehash(node); // 擴容後面也會具體分析
                else
                    // 沒有達到閾值，將 node 放到數組 tab 的 index 位置， // 其實就是將新的節點設置成原鏈表的表頭
 setEntryAt(tab, index, node); ++modCount; count = c; oldValue = null; break; } } } finally { // 解鎖
 unlock(); } return oldValue; }

View Code

咱們能夠看到，代碼一開始就先去得到所在Segment的鎖：

HashEntry<K,V> node = tryLock() ? null : scanAndLockForPut(key, hash, value);

首先，這個tryLock()是個快速得到鎖的方法，得到的話就返回ture，那麼node就賦值爲null。

若是沒得到鎖的話，說明存在競爭，那麼就進入scanAndLockForPut方法。這個方法的話其實也就是不斷去嘗試得到這個Segment的鎖，裏面還有可能順便初始化下這個node元素。（就可能順便構造下你要插入的那個鍵值對的node）

這個scanAndLockForPut的方法等等下面纔去分析，這兩行代碼的結果就是——得到了segment的鎖，而後可能初始化了node也可能沒有。（看下面代碼會知道node有沒有初始化沒所謂的）

而後就正常的put操做了。這裏是帶鎖了的，因此不用怕其餘的寫操做會影響。

能夠看到，node爲空它就new一個相關Entry，不爲空就直接頭插入，因此是否是null不影響代碼邏輯。

關於這個setEntryAt方法，能夠簡單看看它的代碼：

static final <K,V> void setEntryAt(HashEntry<K,V>[] tab, int i, HashEntry<K,V> e) { UNSAFE.putOrderedObject(tab, ((long)i << TSHIFT) + TBASE, e); }

也是用了個防止重排序的方法，再加上原本Segment裏面的table還有Entry裏面重要的相關指針都是volatile的，因此可讓讀操做也安全。

而後就來看這個scanAndLockForPut了：

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 // 循環獲取鎖
    while (!tryLock()) { HashEntry<K,V> f; // to recheck first below
        if (retries < 0) { if (e == null) { if (node == null) // speculatively create node // 進到這裏說明數組該位置的鏈表是空的，沒有任何元素 // 固然，進到這裏的另外一個緣由是 tryLock() 失敗，因此該槽存在併發，不必定是該位置
                    node = new HashEntry<K,V>(hash, key, value, null); retries = 0; } else if (key.equals(e.key)) retries = 0; else
                // 順着鏈表往下走
                e = e.next; } // 重試次數若是超過 MAX_SCAN_RETRIES（單核1多核64），那麼不搶了，進入到阻塞隊列等待鎖 // lock() 是阻塞方法，直到獲取鎖後返回
        else if (++retries > MAX_SCAN_RETRIES) { lock(); break; } else if ((retries & 1) == 0 &&
                 // 這個時候是有大問題了，那就是有新的元素進到了鏈表，成爲了新的表頭 // 因此這邊的策略是，至關於從新走一遍這個 scanAndLockForPut 方法
                 (f = entryForHash(this, hash)) != first) { e = first = f; // re-traverse if entry changed
            retries = -1; } } return node;

View Code

代碼全在一個不斷嘗試拿鎖的while循環裏進行，代碼邏輯大概是用這個retries來控制流程。

當這個retries<0的時候，也就是初始狀況，這裏作的是遍歷這個桶的鏈表，看看有沒這個要put的key的entry，若是有的話就停下來，retries置爲0，沒有的話順便new一個node，而後retries置爲0。

若是retries嘗試的次數太大了，就會lock()，這個方法是堵塞鎖，相似synchronized（解鎖在put方法中），直到拿到鎖才break。

最後一個狀況大概是發生了衝突了，就從新走一次這個方法。

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

這個方法就是看似複雜，可是其實就是作了一件事，那就是獲取該 segment 的獨佔鎖，若是須要的話順便實例化了一下 node。

擴容方法rehash

重複一下，segment 數組不能擴容，擴容是 segment 數組某個位置內部的數組 HashEntry\[] 進行擴容，擴容後，容量爲原來的 2 倍。

首先，咱們要回顧一下觸發擴容的地方，put 的時候，若是判斷該值的插入會致使該 segment 的元素個數超過閾值，那麼先進行擴容，再插值，讀者這個時候能夠回去 put 方法看一眼。

該方法不須要考慮併發，由於到這裏的時候，是持有該 segment 的獨佔鎖的。

看代碼：

// 方法參數上的 node 是此次擴容後，須要添加到新的數組中的數據。
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); // 建立新數組
    HashEntry<K,V>[] newTable = (HashEntry<K,V>[]) new HashEntry[newCapacity]; // 新的掩碼，如從 16 擴容到 32，那麼 sizeMask 爲 31，對應二進制 ‘000...00011111’
    int sizeMask = newCapacity - 1; // 遍歷原數組，老套路，將原數組位置 i 處的鏈表拆分到 新數組位置 i 和 i+oldCap 兩個位置
    for (int i = 0; i < oldCapacity ; i++) { // e 是鏈表的第一個元素
        HashEntry<K,V> e = oldTable[i]; if (e != null) { HashEntry<K,V> next = e.next; // 計算應該放置在新數組中的位置， // 假設原數組長度爲 16，e 在 oldTable[3] 處，那麼 idx 只多是 3 或者是 3 + 16 = 19
            int idx = e.hash & sizeMask; if (next == null)   // 該位置處只有一個元素，那比較好辦
                newTable[idx] = e; else { // Reuse consecutive sequence at same slot // e 是鏈表表頭
                HashEntry<K,V> lastRun = e; // idx 是當前鏈表的頭結點 e 的新位置
                int lastIdx = idx; // 下面這個 for 循環會找到一個 lastRun 節點，這個節點以後的全部元素是將要放到一塊兒的
                for (HashEntry<K,V> last = next; last != null; last = last.next) { int k = last.hash & sizeMask; if (k != lastIdx) { lastIdx = k; lastRun = last; } } // 將 lastRun 及其以後的全部節點組成的這個鏈表放到 lastIdx 這個位置
                newTable[lastIdx] = lastRun; // 下面的操做是處理 lastRun 以前的節點， // 這些節點可能分配在另外一個鏈表中，也可能分配到上面的那個鏈表中
                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 放到新數組中剛剛的 兩個鏈表之一 的 頭部
    int nodeIndex = node.hash & sizeMask; // add the new node
 node.setNext(newTable[nodeIndex]); newTable[nodeIndex] = node; table = newTable; }

View Code

找桶用的是直接從新和新的capacity - 1的值與的方法。

轉移操做用了個改良的算法吧，沒怎麼認真看，直接上參考文章的分析：

這裏的擴容比以前的 HashMap 要複雜一些，代碼難懂一點。上面有兩個挨着的 for 循環，第一個 for 有什麼用呢？

仔細一看發現，若是沒有第一個 for 循環，也是能夠工做的，可是，這個 for 循環下來，若是 lastRun 的後面還有比較多的節點，那麼此次就是值得的。由於咱們只須要克隆 lastRun 前面的節點，後面的一串節點跟着 lastRun 走就是了，不須要作任何操做。

我以爲 Doug Lea 的這個想法也是挺有意思的，不過比較壞的狀況就是每次 lastRun 都是鏈表的最後一個元素或者很靠後的元素，那麼此次遍歷就有點浪費了。不過 Doug Lea 也說了，根據統計，若是使用默認的閾值，大約只有 1/6 的節點須要克隆。

get過程

相對於 put 來講，get 真的不要太簡單。

計算 hash 值，找到 segment 數組中的具體位置，或咱們前面用的「槽」
槽中也是一個數組，根據 hash 找到數組中具體的位置
到這裏是鏈表了，順着鏈表進行查找便可

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. 根據 hash 找到對應的 segment
    if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null && (tab = s.table) != null) { // 3. 找到segment 內部數組相應位置的鏈表，遍歷
        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; }

View Code

這裏用了個getObjectVolatile來保證讀的可見性。

併發問題分析

如今咱們已經說完了 put 過程和 get 過程，咱們能夠看到 get 過程當中是沒有加鎖的，那天然咱們就須要去考慮併發問題。

添加節點的操做 put 和刪除節點的操做 remove 都是要加 segment 上的獨佔鎖的，因此它們之間天然不會有問題，咱們須要考慮的問題就是 get 的時候在同一個 segment 中發生了 put 或 remove 操做。

1. put 操做的線程安全性。

初始化槽，這個咱們以前就說過了，使用了 CAS 來初始化 Segment 中的數組。
添加節點到鏈表的操做是插入到表頭的，因此，若是這個時候 get 操做在鏈表遍歷的過程已經到了中間，是不會影響的。固然，另外一個併發問題就是 get 操做在 put 以後，須要保證剛剛插入表頭的節點被讀取，這個依賴於 setEntryAt 方法中使用的 UNSAFE.putOrderedObject還有相關的volatile的field。
擴容。擴容是新建立了數組，而後進行遷移數據，最後面將 newTable 設置給屬性 table。因此，若是 get 操做此時也在進行，那麼也不要緊，若是 get 先行，那麼就是在舊的 table 上作查詢操做；而 put 先行，那麼 put 操做的可見性保證就是 table 使用了 volatile 關鍵字。

2. remove 操做的線程安全性。
remove 操做咱們沒有分析源碼，因此這裏說的讀者感興趣的話仍是須要到源碼中去求實一下的。

get 操做須要遍歷鏈表，可是 remove 操做會」破壞」鏈表。

若是 remove 破壞的節點 get 操做已通過去了，那麼這裏不存在任何問題。

若是 remove 先破壞了一個節點，分兩種狀況考慮。一、若是此節點是頭結點，那麼須要將頭結點的 next 設置爲數組該位置的元素，table 雖然使用了 volatile 修飾，可是 volatile 並不能提供數組內部操做的可見性保證，因此源碼中使用了 UNSAFE 來操做數組，請看方法 setEntryAt。二、若是要刪除的節點不是頭結點，它會將要刪除節點的後繼節點接到前驅節點中，這裏的併發保證就是 next 屬性是 volatile 的。

總之，首先segment分段所保證了單一的寫、刪是無併發危險的。

而後讀和這些更改性的操做呢，首先是經過table的volatile，而後涉及table中桶元素的替換（要訪問桶的第一個元素的時候），就利用setEntryAt方法中的unsafe的保證有序性的方法。

涉及讀非頭節點的，除了以上措施還有就是Entry中的next屬性也是volatile的。

後期補的：

由於這個table雖說是volatile的，但它裏面的元素是不能有volatile的效果的：Java數組在元素層面的元數據設計上的缺失，沒法表達元素是final、volatile等語義，數組元素就跟沒有標volatile的成員字段同樣，沒法保證線程之間可見性。

因此，咱們看到，這裏關於在數組裏面取元素的操做，都用的是getObjectVolatile之類的，經過這個來彌補數組的這個不足；

而後我看到好多關於set的方法，用的是putOrderedObject，大概是由於這個只是保證禁止指令重排序，開銷比putVolatile版本的小吧，並且應該是這裏只須要禁止重排序就能夠保證併發安全了吧。畢竟有segment鎖，並且get方法又有getObjectVolatile

3、ConcurrentHashMap——1.8

1.8的concurrentHashMap真心難懂，特別是擴容還有轉移方法……

這個版本的hashMap摒棄了Segment的概念，主要是採用CAS算法，底層用的是和1.8的HashMap同樣的數組+鏈表+紅黑樹的實現。

emmm關於紅黑樹就不在這裏講了

先講幾個重要的屬性和須要瞭解的東西

sizeCtl

這個是在ConcurrenthashMap中很重要的一個field，它在流程控制和邏輯代碼上起着重要的做用。

不一樣狀態，sizeCtl所表明的含義也有所不一樣。
- 未初始化：
  - sizeCtl=0：表示沒有指定初始容量。
  - sizeCtl>0：表示初始容量。
- 初始化中：
  - sizeCtl=-1,標記做用，告知其餘線程，正在初始化
- 正常狀態：
  - sizeCtl=0.75n ,擴容閾值
- 擴容中:
  - sizeCtl < 0 : 表示有其餘線程正在執行擴容
  - sizeCtl = (resizeStamp(n) << RESIZE_STAMP_SHIFT) + 2 :表示此時只有一個線程在執行擴容

ForwardingNode

一個用於鏈接兩個table的節點類。它包含一個nextTable指針，用於指向下一張表。並且這個節點的key value next指針所有爲null，它的hash值爲-1. 這裏面定義的find的方法是從nextTable裏進行查詢節點，而不是以自身爲頭節點進行查找。

大量的Unsafe和CAS

在ConcurrentHashMap中，隨處能夠看到U, 大量使用了U.compareAndSwapXXX的方法，這個方法是利用一個CAS算法實現無鎖化的修改值的操做（利用Unsafe來得到底層的相關支持CAS的方法），他能夠大大下降鎖代理的性能消耗。這個算法的基本思想就是不斷地去比較當前內存中的變量值與你指定的一個變量值是否相等，若是相等，則接受你指定的修改的值，不然拒絕你的操做。由於當前線程中的值已經不是最新的值，你的修改極可能會覆蓋掉其餘線程修改的結果。這一點與樂觀鎖，SVN的思想是比較相似的。

　　unsafe靜態塊

　　unsafe代碼塊控制了一些屬性的修改工做，好比最經常使用的SIZECTL 。在這一版本的concurrentHashMap中，大量應用來的CAS方法進行變量、屬性的修改工做。利用CAS進行無鎖操做，能夠大大提升性能。

private static final sun.misc.Unsafe U; private static final long SIZECTL; private static final long TRANSFERINDEX; private static final long BASECOUNT; private static final long CELLSBUSY; private static final long CELLVALUE; private static final long ABASE; private static final int ASHIFT; static { try { U = sun.misc.Unsafe.getUnsafe(); Class<?> k = ConcurrentHashMap.class; SIZECTL = U.objectFieldOffset (k.getDeclaredField("sizeCtl")); TRANSFERINDEX = U.objectFieldOffset (k.getDeclaredField("transferIndex")); BASECOUNT = U.objectFieldOffset (k.getDeclaredField("baseCount")); CELLSBUSY = U.objectFieldOffset (k.getDeclaredField("cellsBusy")); Class<?> ck = CounterCell.class; CELLVALUE = U.objectFieldOffset (ck.getDeclaredField("value")); Class<?> ak = Node[].class; ABASE = U.arrayBaseOffset(ak); int scale = U.arrayIndexScale(ak); if ((scale & (scale - 1)) != 0) throw new Error("data type scale not a power of two"); ASHIFT = 31 - Integer.numberOfLeadingZeros(scale); } catch (Exception e) { throw new Error(e); } }

View Code

　　三個關於table中桶的相關操做的核心方法

　　ConcurrentHashMap定義了三個原子操做，用於對指定位置的節點進行操做。正是這些原子操做保證了ConcurrentHashMap的線程安全。

//得到在i位置上的Node節點
    static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) { return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE); } //利用CAS算法設置i位置上的Node節點。之因此能實現併發是由於他指定了原來這個節點的值是多少 //在CAS算法中，會比較內存中的值與你指定的這個值是否相等，若是相等才接受你的修改，不然拒絕你的修改 //所以當前線程中的值並非最新的值，這種修改可能會覆蓋掉其餘線程的修改結果 有點相似於SVN
    static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i, Node<K,V> c, Node<K,V> v) { return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v); } //利用volatile方法設置節點位置的值
    static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) { U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v); }

View Code

從構造函數的初始化開始看吧

// 這構造函數裏，什麼都不幹
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; }

這個初始化方法有點意思，經過提供初始容量，計算了 sizeCtl，sizeCtl = 【 (1.5 * initialCapacity + 1)，而後向上取最近的 2 的 n 次方】。如 initialCapacity 爲 10，那麼獲得 sizeCtl 爲 16，若是 initialCapacity 爲 11，獲得 sizeCtl 爲 32。

emm如今這個sizeCtl好像衝當一個capacity的角色，在put方法中，initial表格的時候好像又會把這個sizeCtl變成相似一個threshold的角色，給擴容的時候作判斷。

而後直接經過put方法來更深刻吧

put：

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()); // 用於記錄相應鏈表的長度
    int binCount = 0; for (Node<K,V>[] tab = table;;) { Node<K,V> f; int n, i, fh; // 若是數組"空"，進行數組初始化
        if (tab == null || (n = tab.length) == 0) // 初始化數組，後面會詳細介紹
            tab = initTable(); // 找該 hash 值對應的數組下標，獲得第一個節點 f
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { // 若是數組該位置爲空， // 用一次 CAS 操做將這個新值放入其中便可，這個 put 操做差很少就結束了，能夠拉到最後面了 // 若是 CAS 失敗，那就是有併發操做，進到下一個循環就行了
            if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null))) break;                   // no lock when adding to empty bin
 } // hash 竟然能夠等於 MOVED，這個須要到後面才能看明白，不過從名字上也能猜到，確定是由於在擴容
        else if ((fh = f.hash) == MOVED) // 幫助數據遷移，這個等到看完數據遷移部分的介紹後，再理解這個就很簡單了
            tab = helpTransfer(tab, f); else { // 到這裏就是說，f 是該位置的頭結點，並且不爲空
 V oldVal = null; // 獲取數組該位置的頭結點的監視器鎖
            synchronized (f) { if (tabAt(tab, i) == f) { if (fh >= 0) { // 頭結點的 hash 值大於 0，說明是鏈表 // 用於累加，記錄鏈表的長度
                        binCount = 1; // 遍歷鏈表
                        for (Node<K,V> e = f;; ++binCount) { K ek; // 若是發現了"相等"的 key，判斷是否要進行值覆蓋，而後也就能夠 break 了
                            if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) { oldVal = e.val; if (!onlyIfAbsent) e.val = value; break; } // 到了鏈表的最末端，將這個新值放到鏈表的最後面
                            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; // 調用紅黑樹的插值方法插入新節點
                        if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key, value)) != null) { oldVal = p.val; if (!onlyIfAbsent) p.val = value; } } } } // binCount != 0 說明上面在作鏈表操做
            if (binCount != 0) { // 判斷是否要將鏈表轉換爲紅黑樹，臨界值和 HashMap 同樣，也是 8
                if (binCount >= TREEIFY_THRESHOLD) // 這個方法和 HashMap 中稍微有一點點不一樣，那就是它不是必定會進行紅黑樹轉換， // 若是當前數組的長度小於 64，那麼會選擇進行數組擴容，而不是轉換爲紅黑樹 // 具體源碼咱們就不看了，擴容部分後面說
 treeifyBin(tab, i); if (oldVal != null) return oldVal; break; } } } // 
    addCount(1L, binCount); return null; }

View Code

put是在一個for的無限循環中進行的，大概分爲這幾個狀況：

1. 表格爲空，初始化先。

2. 桶中沒有東西，那就直接用CAS來爲桶中的頭指針設值。

else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { // 若是數組該位置爲空， // 用一次 CAS 操做將這個新值放入其中便可，這個 put 操做差很少就結束了，能夠拉到最後面了 // 若是 CAS 失敗，那就是有併發操做，進到下一個循環就行了
            if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null))) break;                   // no lock when adding to empty bin
        }

3. else if ((fh = f.hash) == MOVED)，就幫助數據遷移。

4. 桶中不爲空，且不是處於遷移狀態，那麼就用synchronized來進行put操做。這裏麪包括找到key的替換操做；沒找到key的鏈表尾插入操做；數量達到閾值的樹化操做；原本就是樹結點交給樹來作put操做

put完後，出去循環，而後addCount方法，這個方法也是有點煩的，大概作了兩件事：

對 table 的長度加一。不管是經過修改 baseCount，仍是經過使用 CounterCell。當 CounterCell 被初始化了，就優先使用他，再也不使用 baseCount。
檢查是否須要擴容，或者是否正在擴容。若是須要擴容，就調用擴容方法，若是正在擴容，就幫助其擴容。

下面來慢慢看這些方法。

先是initialTable

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 設置爲 -1，表明搶到了鎖
        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { try { if ((tab = table) == null || tab.length == 0) { // DEFAULT_CAPACITY 默認初始容量是 16
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY; // 初始化數組，長度爲 16 或初始化時提供的長度
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n]; // 將這個數組賦值給 table，table 是 volatile 的
                    table = tab = nt; // 若是 n 爲 16 的話，那麼這裏 sc = 12 // 其實就是 0.75 * n
                    sc = n - (n >>> 2); } } finally { // 設置 sizeCtl 爲 sc，咱們就當是 12 吧
                sizeCtl = sc; } break; } } return tab; }

View Code

這個比較簡單，主要就是初始化一個合適大小的數組，而後會設置 sizeCtl，就是讓sizeCtl變成一個相似threshold的角色，給擴容的時候作判斷用的。

初始化方法中的併發問題是經過對 sizeCtl 進行一個 CAS 操做來控制的。

而後就是關於擴容、transfer的操做了，這個是真滴難懂，代碼還長，我暫時先記一下它的思路吧。

關於擴容的思路（來自：https://blog.csdn.net/varyall/article/details/81283231）

jdk8中，採用多線程擴容。整個擴容過程，經過CAS設置sizeCtl，transferIndex等變量協調多個線程進行併發擴容。

先介紹幾個相關的field：

nextTable：擴容時，把table中的元素遷移至nextTable，擴容時非空。

sizeCtl：上面講了，這裏放個圖：

transferIndex：擴容索引，表示已經分配給擴容線程的table數組索引位置。主要用來協調多個線程，併發安全地獲取遷移任務（hash桶）。

其實說白了，就是多個線程來一塊兒擴容，每一個線程要作的，就是把當前transferIndex到transferIndex - stride個桶的數據遷移到新table中去。（transferIndex一開始在數組尾巴那，往前挪）

看兩個圖理解下：

　　1 在擴容以前，transferIndex 在數組的最右邊。此時有一個線程發現已經到達擴容閾值，準備開始擴容。

　　2 擴容線程，在遷移數據以前，首先要將transferIndex左移（以cas的方式修改transferIndex=transferIndex-stride(要遷移hash桶的個數)），獲取遷移任務。每一個擴容線程都會經過for循環+CAS的方式設置transferIndex，所以能夠確保多線程擴容的併發安全。

forwardingNode：擴容索引，表示已經分配給擴容線程的table數組索引位置。主要用來協調多個線程，併發安全地獲取遷移任務（hash桶）。簡單地說，遷移完的桶就會被設爲這個結點，這個結點的hash值是-1，也就是常量MOVED的值。

看看擴容過程：

線程執行Put，發現容量要擴容了，這個時候的transferIndex = table.length = 32。
擴容線程A 以cas的方式修改transferindex=31-16=16 ,而後按照降序遷移table[31]--table[16]這個區間的hash桶。
遷移hash桶時，會將桶內的鏈表或者紅黑樹，按照必定算法，拆分紅2份，將其插入nextTable[i]和nextTable[i+n]（n是table數組的長度）。遷移完畢的hash桶,會被設置成ForwardingNode節點，以此告知訪問此桶的其餘線程，此節點已經遷移完畢。
此時，線程2訪問到了ForwardingNode節點，若是線程2執行的put或remove等寫操做，那麼就會先幫其擴容。若是線程2執行的是get等讀方法，則會調用ForwardingNode的find方法，去nextTable裏面查找相關元素。

　　5. 若是準備加入擴容的線程，發現如下狀況，放棄擴容，直接返回。

- 　　發現transferIndex=0,即全部node均已分配
- 　　發現擴容線程已經達到最大擴容線程數

總之擴容就是：多線程無鎖擴容的關鍵就是經過CAS設置sizeCtl與transferIndex變量，協調多個線程對table數組中的node進行遷移。

get的分析

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) { // 判斷頭結點是否就是咱們須要的節點
        if ((eh = e.hash) == h) { if ((ek = e.key) == key || (ek != null && key.equals(ek))) return e.val; } // 若是頭結點的 hash 小於 0，說明 正在擴容，或者該位置是紅黑樹
        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; }

View Code

get 方法歷來都是最簡單的，這裏也不例外：

計算 hash 值
根據 hash 值找到數組對應位置: (n – 1) & h
根據該位置處結點性質進行相應查找

若是該位置爲 null，那麼直接返回 null 就能夠了
若是該位置處的節點恰好就是咱們須要的，返回該節點的值便可
若是該位置節點的 hash 值小於 0，說明正在擴容（-1），或者是紅黑樹（-2），就用find來解決get獲取。（擴容的話若是爲-1，說明結點轉移了，就去nextTable裏面去get）
若是以上 3 條都不知足，那就是鏈表，進行遍歷比對便可

我本身對1.8的concurrentHashMap的併發分析：

1. 涉及寫的方面

若是寫的時候發現，table都仍是null，就要初始化，這個初始化顯然要考慮併發，這裏的併發主要經過對sizeCtl的CAS操做還有table的volatile來實現。擴容是個While循環，第一時間會判斷sizeCtl是否是小於0，是的話說明有線程在初始化這個table了，就yield；不然就能夠進行初始化操做，第一步就CAS改sizeCtl爲-1，若是失敗從新進入循環，成功的話就搞一個table賦值給類變量table咯，這個類變量table是volatile因此能夠保證其餘線程的可見性。
若是table中某個桶爲空，就直接用cas把新結點設爲桶的那個首節點，失敗的話會結合外層循環造成自旋CAS。咱們知道table雖然是volatile，但table中的元素的寫操做並不具備volatile特性，因此這裏是經過CAS操做來解決這個問題的。
若是table中某個桶不爲空，也就是下面是個鏈表或者是個紅黑樹，這個時候就要synchronized(桶的頭節點)了，這是put操做惟一一個用到鎖的地方，可見對比1.7好了不少，由於鎖的顆粒度從segment編導是一個bucket的位置，也就是說桶下面的東西的寫操做直接是確定不會有衝突的了。
若是須要擴容，則通過CAS設置sizeCtl與transferIndex變量，協調多個線程對table數組中的node進行遷移。

2. 涉及讀方面的操做

讀咱們能夠看到是徹底沒有鎖的，get能夠大體分爲三種狀況，1. 桶的頭節點直接就是那個想要的key，直接就返回這個node的val；2. 若是桶元素的hash值小於0，多是紅黑樹多是forwardingNode，交給這個node的find方法，可能去新的table中get，也可能交給紅黑樹get；3. 都不是，就在這個桶下面的鏈表中遍歷。

第一種狀況，直接讀桶，也就是讀Node[] table的某個元素，這個爲何能夠不加鎖？？若是讀的時候有人在put怎麼辦？這裏我以爲主要是經過這個tabAt()方法來解決的，咱們能夠看到tabAt方法是用getObjectVolatile來實現的，因此能夠保證讀對寫的可見性吧。
第二種狀況，另外一個table中find同樣的吧，紅黑樹不作分析……
第三種狀況，遍歷的時候，可能涉及到鏈表的增加，刪除什麼的；node結點元素的變化。鏈表變化，其實就next指針的變化，但next指針是volatile的，因此對寫操做是具備可見性的；而後node元素變化？也不怕，由於node的val也是volatile的，在多線程環境下線程A修改結點的val或者新增節點的時候是對線程B可見的。

4、參考文章：

http://www.importnew.com/28263.html——《Java7/8 中的 HashMap 和 ConcurrentHashMap 全解析》本文主要是根據這篇的主線來轉載的。

https://www.cnblogs.com/throwable/p/9139947.html——《JAVA中神奇的雙刃劍--Unsafe》

https://www.cnblogs.com/seyer/p/5819904.html——網上不少講解都這個版本

https://blog.csdn.net/varyall/article/details/81283231——《ConcurrentHashMap源碼分析（JDK8）擴容實現機制》這篇講擴容機制的，思路講得很清晰了，也有關鍵源碼的分析。