詳解ConCurrentHashMap源碼（jdk1.8）

• 2022 年 8 月 12 日
• 筆記

ConCurrentHashMap是一個支援高並發集合，常用的集合之一，在jdk1.8中ConCurrentHashMap的結構和操作和HashMap都很類似：

• 數據結構基於數組+鏈表/紅黑樹。
• get通過計算hash值後取模數組長度確認索引來查詢元素。
• put方法也是先找索引位置，然後不存在就直接添加，存在相同key就替換。
• 擴容都是創建新的table數組，原來的數據轉移到新的table數組中。

唯一不同的是，HashMap不支援並發操作，ConCurrentHashMap是支援並發操作的。所以ConCurrentHashMap的設計也比HashMap也複雜的多，通過閱讀ConCurrentHashMap的源碼，也更加了解一些並發的操作,比如：

• volatile 執行緒可見性
• CAS 樂觀鎖
• synchronized 同步鎖/悲觀鎖

詳見HashMap相關文章：

詳解HashMap源碼解析（上）

詳解HashMap源碼解析（下）

數據結構

ConCurrentHashMap是由數組+鏈表/紅黑樹組成的：

其中左側部分是一個哈希表,通過hash演算法確定元素在數組的下標:

transient volatile Node<K,V>[] table;

鏈表是為了解決hash衝突,當發生衝突的時候。採用鏈表法,將元素添加到鏈表的尾部。其中Node節點存儲數據：

 static class Node<K,V> implements Map.Entry<K,V> {
    final int hash;
    final K key;
    volatile V val;
    volatile Node<K,V> next;

    Node(int hash, K key, V val, Node<K,V> next) {
        this.hash = hash;
        this.key = key;
        this.val = val;
        this.next = next;
    }
}

Node節點包含：

• hash hash值
• key 值
• value 值
• next next指針

主要屬性欄位

// 最大容量
int MAXIMUM_CAPACITY = 1 << 30;

// 初始化容量
int DEFAULT_CAPACITY = 16

// 控制數組初始化或者擴容，為負數時，表示數組正在初始化或者擴容。-1表示正在初始化。其他情況-n表示n執行緒正在擴容。
private transient volatile int sizeCtl;

// 裝載因子
float LOAD_FACTOR = 0.75f

// 鏈表長度為 8 轉成紅黑樹
int TREEIFY_THRESHOLD = 8

// 紅黑樹長度小於6退化成鏈表
int UNTREEIFY_THRESHOLD = 6;

獲取數據get

public V get(Object key) {
    Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
    // 計算hash值
    int h = spread(key.hashCode());
    // 判斷 tab 不為空並且 tab對應的下標不為空 
    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;
        }
        // eh < 0 表示遇到擴容
        else if (eh < 0)
            return (p = e.find(h, key)) != null ? p.val : null;
        // 遍歷鏈表，直到遍歷key相等的值    
        while ((e = e.next) != null) {
            if (e.hash == h &&
                ((ek = e.key) == key || (ek != null && key.equals(ek))))
                return e.val;
        }
    }
    return null;
}

• 獲取數據流程：
  • 調用spread獲取hash值,通過(n - 1) & h取余獲取數組下標的數據。
  • 首節點符合就返回數據。
  • eh<0表示遇到了擴容，會調用正在擴容節點ForwardingNode的find方法，查找該節點，匹配就返回。
  • 遍歷鏈表，匹配到數據就返回。
  • 以上都不符合，返回null。

get如何實現執行緒安全

get方法裡面沒有使用到鎖，那是如何實現執行緒安全。主要使用到了volatile。

• volatile

一個執行緒對共享變數的修改，另外一個執行緒能夠立刻看到，我們稱為可見性。

cpu運行速度比記憶體速度快很多，為了均衡和記憶體之間的速度差異，增加了cpu快取,如果在cpu快取中存在cpu需要數據，說明命中了cpu快取，就不經過訪問記憶體。如果不存在，則要先把記憶體的數據載入到cpu快取中，在返回給cpu處理器。

在多核cpu的伺服器中，每個cpu都有自己的快取，cpu之間的快取是不共享的。 當多個執行緒在不同的cpu上執行時，比如下圖中，執行緒A操作的是cpu-1上的快取，執行緒B操作的是cpu-2上的快取，這個時候，執行緒A對變數V的操作對於執行緒B是不可見的。

但是一個變數被volatile聲明，它的意思是：

告訴編譯器，對這個變數的讀寫，不能使用cpu快取，必須從記憶體中讀取或者寫入。

上面的變數V被volatile聲明，執行緒A在cup-1中修改了數據，會直接寫到記憶體中，不會寫入到cpu快取中。而執行緒B無法從cpu快取讀取變數，需要從主記憶體拉取數據。

• 總結:
  • 使用volatile關鍵字的變數會將修改的變數強制寫入記憶體中。
  • 其他執行緒讀取變數時，會直接從記憶體中讀取變數。

volatile在get應用

• table哈希表

transient volatile Node<K,V>[] table;

使用volatile聲明數組，表示引用地址是volatile而不是數組元素是volatile。

既然不是數組元素被修飾成volatile，那實現執行緒安全在看Node節點。

• Node節點

static class Node<K,V> implements Map.Entry<K,V> {
    final int hash;
    final K key;
    volatile V val;
    volatile Node<K,V> next;
}

其中val和next都用了volatile修飾，在多執行緒環境下，執行緒A修改節點val或者新增節點對別人執行緒是可見的。
所以get方法使用無鎖操作是可以保證執行緒安全。

既然volatile修飾數組對get操作沒有效果，那加在volatile上有什麼目的呢？

是為了數組在擴容的時候對其他執行緒具有可見性。

• jdk 1.8 的get操作不使用鎖，主要有兩個方面：
  • Node節點的val和next都用volatile修飾，保證執行緒修改或者新增節點對別人執行緒是可見的。
  • volatile修飾table數組，保證數組在擴容時其它執行緒是具有可見性的。

添加數據put

put(K key, V value)直接調用putVal(key, value, false)方法。

public V put(K key, V value) {
    return putVal(key, value, false);
}

putVal()方法：

final V putVal(K key, V value, boolean onlyIfAbsent) {
        // key或者value為空，報空指針錯誤
        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為空或者長度為0，初始化table
                tab = initTable();
            // 使用volatile查找索引下的數據
            else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
                // 索引位置沒有數據，使用cas添加數據
                if (casTabAt(tab, i, null,
                             new Node<K,V>(hash, key, value, null)))
                    break;                   // no lock when adding to empty bin
            }
            // MOVED表示數組正在進行數組擴容，當前進行也參加到數組複製
            else if ((fh = f.hash) == MOVED)
                tab = helpTransfer(tab, f);
            else {
                V oldVal = null;
                // 數組不在擴容和也有值，說明數據下標處有值
                // 鏈表中有數據，使用synchronized同步鎖
                synchronized (f) {
                    if (tabAt(tab, i) == f) {
                        // 為鏈表
                        if (fh >= 0) {
                            binCount = 1;
                            // 遍歷鏈表
                            for (Node<K,V> e = f;; ++binCount) {
                                K ek;
                                // hash 以及key相同，替換value值
                                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;
                                }
                            }
                        }
                        // 紅黑樹，TreeBin哈希值固定為-2
                        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;
                            }
                        }
                    }
                }
                if (binCount != 0) {
                    // 鏈錶轉紅黑樹
                    if (binCount >= TREEIFY_THRESHOLD)
                        treeifyBin(tab, i);
                    if (oldVal != null)
                        return oldVal;
                    break;
                }
            }
        }
        addCount(1L, binCount);
        return null;
    }

• 添加數據流程：
  • 判斷key或者value為null都會報空指針錯誤。
  • 計算hash值,然後開啟沒有終止條件的循環。
  • 如果table數組為null,初始化數組。
  • 數組table不為空，通過volatile找到數組對應下標是否為空，為空就使用CAS添加頭結點。
  • 節點的hash=-1表示數組正在擴容，一起進行擴容操作。
  • 以上不符合，說明索引處有值，使用synchronized鎖住當前位置的節點，防止被其他執行緒修改。
    • 如果是鏈表，遍歷鏈表，匹配到相同的key替換value值。如果鏈表找不到，就添加到鏈表尾部。
    • 如果是紅黑樹，就添加到紅黑樹中。
  • 節點的鏈表個數大於8,鏈表就轉成紅黑樹。

ConcurrentHashMap鍵值對為什麼都不能為null,而HashMap就可以？

通過get獲取數據時，如果獲取的數據是null，就無法判斷，是put時的value為null，還是找個key就沒做過映射。HashMap是非並發的，可以通過contains(key)判斷，而支援並發的ConcurrentHashMap在調用contains方法和get方法的時候，map可能已經不同了。參考

如果數組table為空調用initTable初始化數組：

private final Node<K,V>[] initTable() {
    Node<K,V>[] tab; int sc;
    // table 為 null
    while ((tab = table) == null || tab.length == 0) {
       
        if ((sc = sizeCtl) < 0)
         // sizeCtl<0表示其它執行緒正在初始化數組數組，當前執行緒需要讓出CPU
            Thread.yield(); // lost initialization race; just spin
        // 調用CAS初始化table表    
        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
            try {
                if ((tab = table) == null || tab.length == 0) {
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                    @SuppressWarnings("unchecked")
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                    table = tab = nt;
                    sc = n - (n >>> 2);
                }
            } finally {
                sizeCtl = sc;
            }
            break;
        }
    }
    return tab;
}

initTable判斷sizeCtl值，如果sizeCtl為-1表示有其他執行緒正在初始化數組，當前執行緒調用Thread.yield讓出CPU。而正在初始化數組的執行緒通過Unsafe.compareAndSwapInt方法將sizeCtl改成-1。

initTable最外層一直使用while循環，而非if條件判斷，就是確保數組可以初始化成功。

數組初始化成功之後，再執行添加的操作，調用tableAt通過volatile的方式找到(n-1)&hash處的bin節點。

• 如果為空，使用CAS添加節點。
• 不為空，需要使用synchronized鎖，索引對應的bin節點，進行添加或者更新操作。

Insertion (via put or its variants) of the first node in an
empty bin is performed by just CASing it to the bin. This is
by far the most common case for put operations under most
key/hash distributions. Other update operations (insert,
delete, and replace) require locks. We do not want to waste
the space required to associate a distinct lock object with
each bin, so instead use the first node of a bin list itself as
a lock. Locking support for these locks relies on builtin
"synchronized" monitors.

如果f的hash值為-1，說明當前f是ForwaringNode節點，意味著有其它執行緒正在擴容，則一起進行擴容操作。

完成添加或者更新操作之後，才執行break終止最外層沒有終止條件的for循環，確保數據可以添加成功。

最後執行addCount方法。

private final void addCount(long x, int check) {
    CounterCell[] as; long b, s;
    // 利用CAS更新baseCoount
    if ((as = counterCells) != null ||
        !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
        CounterCell a; long v; int m;
        boolean uncontended = true;
        if (as == null || (m = as.length - 1) < 0 ||
            (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
            !(uncontended =
              U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
            fullAddCount(x, uncontended);
            return;
        }
        if (check <= 1)
            return;
        s = sumCount();
    }
    // check >= 0,需要檢查是否需要進行擴容操作
    if (check >= 0) {
        Node<K,V>[] tab, nt; int n, sc;
        while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
               (n = tab.length) < MAXIMUM_CAPACITY) {
            int rs = resizeStamp(n);
            if (sc < 0) {
                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))
                    transfer(tab, nt);
            }
            else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                         (rs << RESIZE_STAMP_SHIFT) + 2))
                transfer(tab, null);
            s = sumCount();
        }
    }
}

擴容transfer

什麼時候會擴容

*插入一個新的節點：

• 新增節點，所在的鏈表元素個數達到閾值8，則會調用treeifyBin把鏈錶轉成紅黑樹，在轉成之前，會判斷數組長度小於MIN_TREEIFY_CAPACITY,默認是64,觸發擴容。
• 調用put方法，在結尾addCount方法記錄元素個數，並檢查是否進行擴容，數組元素達到閾值時，觸發擴容。

不使用加鎖的，支援多執行緒擴容。利用並發處理減少擴容帶來性能的影響。

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
    int n = tab.length, stride;
    if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
        stride = MIN_TRANSFER_STRIDE; // subdivide range
    if (nextTab == null) {            // initiating
        try {
            @SuppressWarnings("unchecked")
            // 創建nextTab，容量為原來容量的兩倍
            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 = nextTab;
        transferIndex = n;
    }
    int nextn = nextTab.length;
    // 創建 ForwardingNode 節點，作為標記位，表明當前位置已經做過桶處理
    ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
    // advance = true 表明該節點已經處理過了
    boolean advance = true;
    boolean finishing = false; // to ensure sweep before committing nextTab
    for (int i = 0, bound = 0;;) {
        Node<K,V> f; int fh;
        // 控制 --i，遍歷原hash表中的節點
        while (advance) {
            int nextIndex, nextBound;
            if (--i >= bound || finishing)
                advance = false;
            else if ((nextIndex = transferIndex) <= 0) {
                i = -1;
                advance = false;
            }
            // 用CAS計算得到的transferIndex
            else if (U.compareAndSwapInt
                     (this, TRANSFERINDEX, nextIndex,
                      nextBound = (nextIndex > stride ?
                                   nextIndex - stride : 0))) {
                bound = nextBound;
                i = nextIndex - 1;
                advance = false;
            }
        }
        // 將原數組中的節點賦值到新的數組中，nextTab賦值給table，清空nextTable。
        if (i < 0 || i >= n || i + n >= nextn) {
            int sc;
            // 所有節點完成複製工作，
            if (finishing) {
                nextTable = null;
                table = nextTab;
                // 設置新的閾值為原來的1.5倍
                sizeCtl = (n << 1) - (n >>> 1);
                return;
            }
            // 利用CAS方法更新擴容的閾值，sizeCtl減一，說明新加入一個執行緒參與到擴容中
            if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
                if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
                    return;
                finishing = advance = true;
                i = n; // recheck before commit
            }
        }
        // 遍歷的節點為null，則放入到ForwardingNode指針節點
        else if ((f = tabAt(tab, i)) == null)
            advance = casTabAt(tab, i, null, fwd);
        // f.hash==-1表示遍歷到ForwardingNode節點，說明該節點已經處理過了    
        else if ((fh = f.hash) == MOVED)
            advance = true; // already processed
        else {
            // 節點加鎖
            synchronized (f) {
                if (tabAt(tab, i) == f) {
                    Node<K,V> ln, hn;
                    // fh>=0，表示為鏈表節點
                    if (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) {
                                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);
                        }
                        // 在nextTable i 位置處插入鏈表
                        setTabAt(nextTab, i, ln);
                        // 在nextTable i+n 位置處插入鏈表
                        setTabAt(nextTab, i + n, hn);
                        // 在table i的位置處插上ForwardingNode，表示該節點已經處理過
                        setTabAt(tab, i, fwd);
                        // 可以執行 --i的操作，再次遍歷節點
                        advance = true;
                    }
                    // TreeBin紅黑樹，按照紅黑樹處理，處理邏輯和鏈表類似
                    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;
                            }
                        }
                        // 擴容後樹節點的個數<=6,紅黑樹轉成鏈表
                        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;
                    }
                }
            }
        }
    }
}

擴容過程有的複雜，主要涉及到多執行緒的並發擴容，ForwardingNode的作用就是支援擴容操作，將已經處理過的節點和空節點置為ForwardingNode，並發處理時多個執行緒處理ForwardingNode表示已經處理過了，就往後遍歷。

總結

• ConcurrentHashMap是基於數組+鏈表/紅黑樹的數據結構，添加、刪除、更新都是先通過計算key的hash值確定數據的索引值，這和HashMap是類似的，只不過ConcurrentHashMap針對並發做了更多的處理。

• get方法獲取數據，先計算hash值再再和數組長度取余操作獲取索引位置。

  • 通過volatile關鍵字找到table保證多執行緒環境下，數組擴容具有可見性,而Node節點中val和next指針都使用volatile修飾保證數據修改後別的執行緒是可見的。這就保證了ConcurrentHashMap的執行緒安全性。
  • 如果遇到數組擴容，就參與到擴容中。
  • 首節點值匹配到數據就直接返回數據，否則就遍歷鏈表或者紅黑樹，直到匹配到數據。

• put方法添加或者更新數據。

  • 如果key或value為空，就報錯。這是因為在調用get方法獲取數據為null,無法判斷是獲取的數據為null，還是對應的key就不存在映射,HashMap可以通過contains(key)判斷，而ConcurrentHashMap在多執行緒環境下調用contains和get方法的時候，map可能就不同了。
  • 如果table數組為空，先初始化數組，先通過sizeCtl控制並發，如果小於0表示有別的執行緒正在初始化數組，就讓出CPU,否則使用CAS將sizeCtl設置成-1。
  • 初始化數組之後，如果節點為空，使用CAS添加節點。
  • 不為空，就鎖住該節點，進行添加或者更新操作。

• transfer擴容

  • 在新增一個節點時，鏈表個數達到閾值8，會將鏈錶轉成紅黑樹，在轉成之前，會先判斷數組長度小於64,會觸發擴容。還有集合個數達到閾值時也會觸發擴容。
  • 擴容數組的長度是原來數組的兩倍。
  • 為了支援多執行緒擴容創建ForwardingNode節點作為標記位，如果遍歷到該節點，說明已經做過處理。
  • 遍歷賦值原來的數據給新的數組。

參考

為什麼ConcurrentHashMap的讀操作不需要加鎖

Java並發——ConcurrentHashMap(JDK 1.8）