深入揭秘 ConcurrentHashMap：JDK7 到 JDK8 並發優化的演進之路

引言

本文主要從原始碼層次探討 ConcurrentHashMap 的並發控制實現，全文可以濃縮為下面的內容，也可以用來回答問題《ConcurrentHashMap 是如何實現執行緒安全支持高並發的？》

在 JDK 7 中，ConcurrentHashMap 是使用陣列加鏈表的結構實現的。陣列又分為 HashEntry 陣列和 Segment 陣列。其中 Segment 陣列使用 ReentrantLock 控制並發，HashEntry 就是存放鍵值對的。在 JDK 7 中為每一段桶分配一個 segment 分段鎖，當需要操作某一個桶中的資料時，需要先取得對應的分段鎖。

在 JDK 8 中，ConcurrentHashMap 是使用的陣列加鏈表加紅黑樹實現的。透過 volatile + cas （樂觀鎖）和 synchronized （悲觀鎖）實現並發控制的。鎖的粒度減少到對桶頭加鎖，並發度更高。並且鏈表可以優化為紅黑樹，查詢效率也比 JDK 7 的高；其次 HashMap 擴容的時候，其他執行緒可以協助擴容。

1. (JDK 8) ConcurrentHashMap

ConcurrentHashMap 的重要屬性和內部類

// 默認為 0
// 當初始化時, 為 -1
// 當擴容時, 為 -(1 + 擴容執行緒數)
// 當初始化或擴容完成後，為下一次的擴容的閾值大小
private transient volatile int sizeCtl;
// 整個 ConcurrentHashMap 就是 Node[]
static class Node<K,V> implements Map.Entry<K,V> {}
// hash 表
transient volatile Node<K,V>[] table;
// 擴容時的新 hash 表
private transient volatile Node<K,V>[] nextTable;
// 擴容時如果某個 bin 遷移完畢, 用 ForwardingNode 作為舊 table bin 的頭結點
static final class ForwardingNode<K,V> extends Node<K,V> {}
// 用在 compute 以及 computeIfAbsent 時, 用來佔位, 計算完成後替換為普通 Node
static final class ReservationNode<K,V> extends Node<K,V> {}
// 作為 treebin 的頭節點, 存儲 root 和 first
static final class TreeBin<K,V> extends Node<K,V> {}
// 作為 treebin 的節點, 存儲 parent, left, right
static final class TreeNode<K,V> extends Node<K,V> {}

可以發現在 JDK 8 的 ConcurrentHashMap 中，底層陣列就是 Node[]，存放的都是 Node 或者它的子類。相對於 HashMap 來說，ConcurrentHashMap 在陣列中用 ForwardingNode 和 TreeBin 當哨兵。

sizeCtl 的作用

• sizeCtl < 0 且「戳匹配」說明正在擴容：本執行緒可幫忙（sc+1）並調用 transfer(tab, nt)
• sizeCtl >= 0 且超過閾值：當前執行緒成為主擴容者，設置 sizeCtl = (rs<<shift)+2 後進入 transfer
• 擴容完成後，sizeCtl 會被恢復為新閾值

ForwardingNode 的哨兵作用：

1. 標記 hash 值所映射的鏈表已經遷移完畢
2. 查詢時，遇到 forwardingnode 就會去 nextTable 中進行查找

1.1 擴容機制和樹化機制

1. 擴容機制

   • 容量（capacity）
     • HashMap 底層陣列 table 的長度，始終為 2 的冪次方
   • 負載因子（load factor）
     • loadFactor，默認值 0.75
   • 閾值（threshold）
     • 已存放元素個數超過 threshold = capacity * loadFactor 時，就觸發擴容，每次擴容後重新計算
   • 初次插入
     • 當第一次調用 put 時，如果 table 為空，則會先調用 resize()，默認初始化容量為 16（或用戶指定的初始容量向上取到最近的 2 的冪）
   • 插入後元素數超過閾值
     • 每次 put 後，若 size > threshold，則執行 resize() 擴容

2. 樹化機制

   • 鏈表轉化為紅黑樹的兩個前提條件
     • 鏈表長度超過閾值：當向同一個桶插入第 9 個元素（也就是鏈表長度 > TREEIFY_THRESHOLD，常量值 8）時，滿足第一條件
     • 表的容量足夠大：只有在當前 table.length ≥ MIN_TREEIFY_CAPACITY（常量值 64）時，才真正將鏈表樹化；否則不樹化，而是先觸發一次擴容（resize），讓更多的桶分攤這些節點

ConcurrentHashMap 為什麼要加容量檢查？
對於小表，鏈表查找開銷尚小，擴容更有利於分散節點；等表足夠大，才值得投入樹化的維護成本。

HashMap 和 ConcurrentHashMap 擴容機制和樹化的總結：

• TREEIFY_THRESHOLD = 8：鏈表長度超過 8 時嘗試樹化
• MIN_TREEIFY_CAPACITY = 64：只有當表長 ≥ 64 時才做樹化，否則先擴容
• UNTREEIFY_THRESHOLD = 6：樹中節點 ≤ 6 時退化為鏈表
• 擴容拆分：樹節點在擴容時按高/低位拆分，分別樹化或鏈表化

1.2 重要方法

// 獲取 Node[] 中第 i 個 Node
static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {}

// cas 修改 Node[] 中第 i 個 Node 的值, c 為舊值, v 為新值
static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i, Node<K,V> c, Node<K,V> v) {}

// 直接修改 Node[] 中第 i 個 Node 的值, v 為新值
static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {}

1.3 構造器分析

可以看到 ConcurrentHashMap 使用的是懶惰初始化，在構造方法中僅僅計算了 table 的大小，以後在第一次使用時才會真正創建。

public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) {
    if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
        throw new IllegalArgumentException();
    if (initialCapacity < concurrencyLevel) // Use at least as many bins
        initialCapacity = concurrencyLevel; // as estimated threads
    long size = (long)(1.0 + (long)initialCapacity / loadFactor);
    // tableSizeFor 仍然是保證計算的大小是 2^n, 即 16,32,64 ...
    int cap = (size >= (long)MAXIMUM_CAPACITY) ?
        MAXIMUM_CAPACITY : tableSizeFor((int)size);
    this.sizeCtl = cap;
}

1.4 get 流程

get 實現了無鎖讀取，但是並不保證一致性讀取。ForwardingNode 發布前讀舊桶，發布後讀新表。

public V get(Object key) {
    Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
    // spread 把用戶的 hashCode() 做一次高低位混洗，降低糟糕哈希（低位全 0）的碰撞概率；並保證結果是正數（配合位運算定位桶位）
    int h = spread(key.hashCode());
    // tabAt(...) 不是普通陣列讀，而是經由 Unsafe 的帶內存語義的讀取，保證是 happens‑before 且不加鎖獲取
    if ((tab = table) != null && (n = tab.length) > 0 &&
        (e = tabAt(tab, (n - 1) & h)) != null) {
        // 如果頭結點已經是要查找的 key
        if ((eh = e.hash) == h) {
            if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                // 桶頭命中 O(1) 的時間複雜度
                return e.val;
        }
        // hash 為負數表示該 bin 在擴容中或是 treebin, 這時調用 find 方法來查找
        else if (eh < 0)
            // find 函數會處理正在進行遷移的場景 以及 正常紅黑樹的場景
            return (p = e.find(h, key)) != null ? p.val : null;
        // 正常遍歷鏈表, 用 equals 比較
        while ((e = e.next) != null) {
            if (e.hash == h &&
                ((ek = e.key) == key || (ek != null && key.equals(ek))))
                return e.val;
        }
    }
    return null;
}

get 方法的源碼分析：首先判斷桶頭是否符合、再判斷是否為特殊桶（紅黑樹或者是 ForwardingNode）、走到這裡就說明是鏈表判斷了，直接循環遍歷判斷。

● 如果 table 不為空且長度大於 0 且索引位置有元素
  ○ if 頭節點 key 的 hash 值相等
    ■ 頭節點的 key 指向同一地址或者 equals
      ● 返回 value
  ○ else if 頭節點的 hash 為負數（bin 在擴容或者是 treebin）
    ■ 調用 find 方法查找
  ○ 走到這裡說明是正常鏈表遍歷，進入循環（e 不為空）
    ■ 節點 key 的 hash 值相等，且 key 指向同一地址或 equals
      ● 返回 value
● 返回 null

1.5 put 流程

以下陣列簡稱（table），鏈表簡稱（bin）。

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();
    // 其中 spread 方法會綜合高位低位, 具有更好的 hash 性
    int hash = spread(key.hashCode());
    int binCount = 0;
    for (Node<K,V>[] tab = table;;) {
        // f 是鏈表頭節點
        // fh 是鏈表頭結點的 hash
        // i 是鏈表在 table 中的下標
        Node<K,V> f; int n, i, fh;
        // 要創建 table
        if (tab == null || (n = tab.length) == 0)
            // 初始化 table 使用了 cas, 無需 synchronized 創建成功, 進入下一輪循環
            tab = initTable();
        // 要創建鏈表頭節點
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
            // 添加鏈表頭使用了 cas, 無需 synchronized
            if (casTabAt(tab, i, null,
                         new Node<K,V>(hash, key, value, null)))
                break;
        }
        // 幫忙擴容
        else if ((fh = f.hash) == MOVED)
            // 幫忙之後, 進入下一輪循環
            tab = helpTransfer(tab, f);
        else {
            V oldVal = null;
            // 鎖住鏈表頭節點
            synchronized (f) {
                // 再次確認鏈表頭節點沒有被移動
                if (tabAt(tab, i) == f) {
                    // 鏈表
                    if (fh >= 0) {
                        binCount = 1;
                        // 遍歷鏈表
                        for (Node<K,V> e = f;; ++binCount) {
                            K ek;
                            // 找到相同的 key    (更新機制)
                            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;
                            // 已經是最後的節點了, 新增 Node, 追加至鏈表尾   (插入機制)
                            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;
                        // putTreeVal 會看 key 是否已經在樹中, 是, 則返回對應的 TreeNode
                        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)
                    // 如果鏈表長度 >= 樹化閾值(8), 進行鏈表轉為紅黑樹
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    // 增加 size 計數
    addCount(1L, binCount);
    return null;
}
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();
        // 嘗試將 sizeCtl 設置為 -1（表示初始化 table）
        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
            // 獲得鎖, 創建 table, 這時其它執行緒會在 while() 循環中 yield 直至 table 創建
            try {
                if ((tab = table) == null || tab.length == 0) {
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                    table = tab = nt;
                    sc = n - (n >>> 2);
                }
            } finally {
                sizeCtl = sc;
            }
            break;
        }
    }
    return tab;
}
// check 是之前 binCount 的個數 （addCount函數作用不僅是更新Map的Size數量，還能進行擴容的判斷！）
private final void addCount(long x, int check) {
    CounterCell[] as; long b, s;
    if (
        // 已經有了 counterCells, 向 cell 累加
        (as = counterCells) != null ||
        // 還沒有, 向 baseCount 累加
        !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)
    ) {
        CounterCell a; long v; int m;
        boolean uncontended = true;
        if (
            // 還沒有 counterCells
            as == null || (m = as.length - 1) < 0 ||
            // 還沒有 cell
            (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
            // cell cas 增加計數失敗
            !(uncontended = U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))
        ) {
            // 創建累加單元陣列和cell, 累加重試
            fullAddCount(x, uncontended);
            return;
        }
        if (check <= 1)
            return;
        // 獲取元素個數
        s = sumCount();
    }
    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;
                // newtable 已經創建了，幫忙擴容
                if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                    transfer(tab, nt);
            }
            // 需要擴容，這時 newtable 未創建
            else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                         (rs << RESIZE_STAMP_SHIFT) + 2))
                transfer(tab, null);
            s = sumCount();
        }
    }
}

先 spread 之後進行取桶 → 若 HashMap 未初始化，則執行 initTable（CAS 抢令牌） → 陣列當前的桶空，就用 CAS 放頭結點 → 桶在遷移則幫助遷移 helpTransfer → 否則對 bin 頭加鎖，鏈表插尾或樹插 → 可能 treeify（但容量小先擴容）。

用分段計數 addCount 增量並透過 sizeCtl 觸發或協助擴容（ForwardingNode + 協作搬桶）。

ConcurrentHashMap使用這兩種手段來保證執行緒安全主要是一種權衡的考量，在某些操作中使用synchronized，還是使用CAS，主要是根據鎖競爭程度來判斷的。

比如：在putVal中，如果計算出來的hash槽沒有存放元素，那麼就可以直接使用CAS來進行設置值，這是因為在設置元素的時候，因為hash值經過了各種擾動後，造成hash碰撞的機率較低，那麼我們可以預測使用較少的自旋來完成具體的hash落槽操作。

當發生了hash碰撞的時候說明容量不夠用了或者已經有大量執行緒訪問了，因此這時使用synchronized來處理hash碰撞比CAS效率要高，因為發生了hash碰撞大概率來說是執行緒競爭比較強烈。

1.6 size 計算流程

size 計算實際發生在 put，remove 改變集合元素的操作之中。

• 沒有競爭發生，向 baseCount 累加計數
• 有競爭發生，新建 counterCells，向其中的一個 cell 累加計
  • counterCells 初始有兩個 cell
  • 如果計數競爭比較激烈，會創建新的 cell 來累加計數

public int size() {
    long n = sumCount();
    return ((n < 0L) ? 0 :
            (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
            (int)n);
}
final long sumCount() {
    CounterCell[] as = counterCells; CounterCell a;
    // 將 baseCount 計數與所有 cell 計數累加
    long sum = baseCount;
    if (as != null) {
        for (int i = 0; i < as.length; ++i) {
            if ((a = as[i]) != null)
                sum += a.value;
        }
    }
    return sum;
}

1.7 總結

Java 8 陣列（Node） +（鏈表 Node | 紅黑樹 TreeNode ） 。JDK 8 的 ConcurrentHashMap 中，底層陣列就是 Node[]，存放的都是 Node 或者它的子類，在陣列中用 ForwardingNode 和 TreeBin 當哨兵，用 sizeCtl 標識擴容狀態。

以下陣列稱之為（table），鏈表簡稱（bin）

• 初始化，使用 cas 來保證並發安全，懶惰初始化 table
• 樹化，當 table.length < 64 時，先嘗試擴容，超過 64 時，並且 bin.length > 8 時，會將鏈表樹化，樹化過程 會用 synchronized 鎖住鏈表頭
• put，如果該 bin 尚未創建，只需要使用 cas 創建 bin；如果已經有了，鎖住鏈表頭進行後續 put 操作，元素添加至 bin 的尾部
• get，無鎖操作僅需要保證可見性，所以 value 是被 volatile 修飾的。首先判斷桶頭；在判斷是否為紅黑樹或者是正在擴容 ForwardingNode，那麼它會讓 get 操作在新 table 進行搜索；最後才是時間複雜度高的鏈表遍歷
• 擴容，擴容時以 bin 為單位進行，需要對 bin 進行 synchronized，但這時妙的是其它競爭執行緒也不是無事可做，它們會幫助把其它 bin 進行擴容，擴容時平均只有 1/6 的節點會把複製到新 table 中

(JDK 7)ConcurrentHashMap

它維護了一個 segment 陣列，每個 segment 對應一把鎖（就是繼承 ReentrantLock）。

• 優點：如果多個執行緒訪問不同的 segment，實際是沒有衝突的，這與 jdk8 中是類似的
• 缺點：Segments 陣列默認大小為 16，這個容量初始化指定後就不能改變了，並且不是懶惰初始化

2.1 構造器分析

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;
    // ssize 必須是 2^n, 即 2, 4, 8, 16 ... 表示了 segments 陣列的大小
    int sshift = 0;
    int ssize = 1;
    while (ssize < concurrencyLevel) {
        ++sshift;
        ssize <<= 1;
    }
    // segmentShift 默認是 32 - 4 = 28
    this.segmentShift = 32 - sshift;
    // segmentMask 默認是 15 即 0000 0000 0000 1111
    this.segmentMask = ssize - 1;
    if (initialCapacity > MAXIMUM_CAPACITY)
        initialCapacity = MAXIMUM_CAPACITY;
    int c = initialCapacity / ssize;
    if (c * ssize < initialCapacity)
        ++c;
    int cap = MIN_SEGMENT_TABLE_CAPACITY;
    while (cap < c)
        cap <<= 1;
    // 創建 segments and segments[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];
    UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
    this.segments = ss;
}

構造完成，如下圖所示

JDK 7 的 ConcurrentHashMap 沒有實現懶惰初始化，所以空間佔用相對來說不是很友好。

其中this.segmentShift和this.segmentMask的作用是決定將 key 的 hash 結果匹配到哪個 segment。

例如：根據某一 hash 值求 segment 位置，先將高位向低位移動 this.segmentShift 位。

結果再與 this.segmentMask 做位於運算，最終得到 1010 即下標為 10 的 segment。

2.2 put 流程

當第一次創建 Segment 的時候，需要保證安全性。JDK 7 使用的是 CAS 來確保創建 segement 是安全的。

public V put(K key, V value) {
    Segment<K,V> s;
    if (value == null)
        throw new NullPointerException();
    int hash = hash(key);
    // 計算出 segment 下標
    int j = (hash >>> segmentShift) & segmentMask;

    // 獲得 segment 對象, 判斷是否為 null, 是則創建該 segment
    if ((s = (Segment<K,V>)UNSAFE.getObject 
         (segments, (j << SSHIFT) + SBASE)) == null) {
        // 這時不能確定是否真的為 null, 因為其它執行緒也發現該 segment 為 null,
        // 因此在 ensureSegment 里用 cas 方式保證該 segment 安全性
        s = ensureSegment(j);
    }
    // 進入 segment 的 put 流程
    return s.put(key, hash, value, false);
}

segment 繼承了可重入鎖（ReentrantLock），它的 put 方法為：

final V put(K key, int hash, V value, boolean onlyIfAbsent) {
    // 嘗試加鎖
    HashEntry<K,V> node = tryLock() ? null :
    // 如果不成功, 進入 scanAndLockForPut 流程
    // 如果是多核 cpu 最多 tryLock 64 次, 進入 lock 流程
    // 在嘗試期間, 還可以順便看該節點在鏈表中有沒有, 如果沒有順便創建出來
    scanAndLockForPut(key, hash, value);

    // 執行到這裡 segment 已經被成功加鎖, 可以安全執行
    V oldValue;
    try {
        HashEntry<K,V>[] tab = table;
        int index = (tab.length - 1) & hash;
        HashEntry<K,V> first = entryAt(tab, index);
        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 {
                // 新增
                // 1) 之前等待鎖時, node 已經被創建, next 指向鏈表頭
                if (node != null)
                    node.setNext(first);
                else
                    // 2) 創建新 node
                    node = new HashEntry<K,V>(hash, key, value, first);
                int c = count + 1; 
                // 3) 擴容
                if (c > threshold && tab.length < MAXIMUM_CAPACITY)
                    rehash(node);
                else
                    // 將 node 作為鏈表頭
                    setEntryAt(tab, index, node);
                ++modCount;
                count = c;
                oldValue = null;
                break;
            }
        }
    } finally {
        unlock();
    }
    return oldValue;
}

2.3 rehash 流程

發生在 put 中，因為此時已經獲得了鎖，因此 rehash 時不需要考慮執行緒安全。

private void rehash(HashEntry<K,V> node) {
    HashEntry<K,V>[] oldTable = table;
    int oldCapacity = oldTable.length;
    int newCapacity = oldCapacity << 1;
    threshold = (int)(newCapacity * loadFactor);
    HashEntry<K,V>[] newTable =
        (HashEntry<K,V>[]) new HashEntry[newCapacity];
    int sizeMask = newCapacity - 1;
    for (int i = 0; i < oldCapacity ; i++) {
        HashEntry<K,V> e = oldTable[i];
        if (e != null) {
            HashEntry<K,V> next = e.next;
            int idx = e.hash & sizeMask;
            if (next == null) // Single node on list
                newTable[idx] = e;
            else { // Reuse consecutive sequence at same slot
                HashEntry<K,V> lastRun = e;
                int lastIdx = idx;
                // 過一遍鏈表, 將盡可能把 rehash 後 idx 不變的節點重用
                for (HashEntry<K,V> last = next;
                     last != null;
                     last = last.next) {
                    int k = last.hash & sizeMask;
                    if (k != lastIdx) {
                        lastIdx = k;
                        lastRun = last;
                    }
                }
                newTable[lastIdx] = 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);
                }
            }
        }
    }
    // 擴容完成, 才加入新的節點
    int nodeIndex = node.hash & sizeMask; // add the new node
    node.setNext(newTable[nodeIndex]);
    newTable[nodeIndex] = node;

    // 替換為新的 HashEntry table
    table = newTable;
}

附，調試代碼：

public static void main(String[] args) {
    ConcurrentHashMap<Integer, String> map = new ConcurrentHashMap<>();
    for (int i = 0; i < 1000; i++) {
        int hash = hash(i);
        int segmentIndex = (hash >>> 28) & 15;
        if (segmentIndex == 4 && hash % 8 == 2) {
            System.out.println(i + "\t" + segmentIndex + "\t" + hash % 2 + "\t" + hash % 4 +
                               "\t" + hash % 8);
        }
    }
    map.put(1, "value");
    map.put(15, "value"); // 2 擴容為 4 15 的 hash%8 與其他不同
    map.put(169, "value");
    map.put(197, "value"); // 4 擴容為 8
    map.put(341, "value");
    map.put(484, "value");
    map.put(545, "value"); // 8 擴容為 16
    map.put(912, "value");
    map.put(941, "value");
    System.out.println("ok");
}
private static int hash(Object k) {
    int h = 0;
    if ((0 != h) && (k instanceof String)) {
        return sun.misc.Hashing.stringHash32((String) k);
    }
    h ^= k.hashCode();
    // Spread bits to regularize both segment and index locations,
    // using variant of single-word Wang/Jenkins hash.
    h += (h << 15) ^ 0xffffcd7d;
    h ^= (h >>> 10);
    h += (h << 3);
    h ^= (h >>> 6);
    h += (h << 2) + (h << 14);
    int v = h ^ (h >>> 16);
    return v;
}

2.4 get 流程

get 時並未加鎖，用了 UNSAFE 方法保證了可見性。擴容過程中，get 先發生就從舊表取內容，get 後發生就從新表取內容。

public V get(Object key) {
    Segment<K,V> s; // manually integrate access methods to reduce overhead
    HashEntry<K,V>[] tab;
    int h = hash(key);
    // u 為 segment 對象在陣列中的偏移量
    long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
    // s 即為 segment
    if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
        (tab = s.table) != null) {
        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;
}

2.5 size 計算流程

• 計算元素個數前，先不加鎖計算兩次，如果前後兩次結果如一樣，認為個數正確返回
• 如果不一樣，進行重試，重試次數超過 3，將所有 segment 鎖住，重新計算個數返回

public int size() {
    // Try a few times to get accurate count. On failure due to
    // continuous async changes in table, resort to locking.
    final Segment<K,V>[] segments = this.segments;
    int size;
    boolean overflow; // true if size overflows 32 bits
    long sum; // sum of modCounts
    long last = 0L; // previous sum
    int retries = -1; // first iteration isn't retry
    try {
        for (;;) {
            if (retries++ == 

原文出處：https://juejin.cn/post/7542868304472145960