`ConcurrentHashMap` 并发哈希表源码

概述

ConcurrentHashMap 是 Java 并发包中最重要的数据结构之一，提供了线程安全的哈希表实现。JDK 7 使用分段锁（Segment + ReentrantLock）实现并发，JDK 8+ 采用 CAS + synchronized 的方式，显著提升了并发性能。

相比 Hashtable 的全表锁，ConcurrentHashMap 只锁住单个桶，支持更高的并发度。同时，它保留了 HashMap 的数组+链表+红黑树结构，在保持 O(1) 平均查找性能的同时保证了线程安全。

本文基于 OpenJDK 21 源码，从 CAS 基础操作开始，逐层深入到插入、扩容、计数等核心并发机制。

本文基于 OpenJDK 21 源码分析，关键实现会对比 JDK 7 到 JDK 21 的版本差异。

核心源码解析

① JDK 8+ 的 `CAS + synchronized` 实现

java

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

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

    // Unsafe 操作
    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);
    }

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

    static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
        U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
    }
}

tabAt() 使用 getObjectVolatile 保证可见性
casTabAt() 使用 compareAndSwapObject 原子更新
setTabAt() 设置值并立即对其他线程可见

② `put(K, V)` 的 `putVal()` 流程

java

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();
    int hash = spread(key.hashCode());
    int binCount = 0;
    for (Node<K,V>[] tab = table;;) {
        Node<K,V> f; int n, i, fh; K fk; V fv;
        if (tab == null || (n = tab.length) == 0)
            tab = initTable();                       // 懒加载初始化
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
            if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value)))
                break;                               // CAS 插入空桶
        }
        else if ((fh = f.hash) == MOVED)
            tab = helpTransfer(tab, f);               // 正在扩容 → 协助扩容
        else {
            V oldVal = null;
            synchronized (f) {                        // 锁住桶头节点
                if (tabAt(tab, i) == f) {             // 二次检查
                    if (fh >= 0) {                     // 链表模式
                        for (Node<K,V> e = f;;) {
                            // ... 遍历链表查找/插入
                        }
                    } else if (f instanceof TreeBin) { // 红黑树模式
                        // ... putTreeVal
                    }
                }
            }
            // ...
        }
    }
    addCount(1L, binCount);
    return null;
}

核心流程：

null 检查：ConcurrentHashMap 不允许 null 键和 null 值
空桶 CAS 插入：桶为空 → casTabAt() 无锁插入
扩容协助：检测到迁移标记 → helpTransfer() 参与扩容
桶锁插入：非空桶 → synchronized(f) 加锁 → 链表/红黑树插入

③ `sizeCtl` 的多种状态

sizeCtl 是 ConcurrentHashMap 最核心的控制字段，用 volatile 修饰：

java

private transient volatile int sizeCtl;

值	含义
`-1`	正在初始化
`-(1+n)`	`n` 个线程正在扩容（高 16 位为扩容标识戳）
`=0`	未初始化（默认值）
`>0`	正常扩容阈值（`capacity * loadFactor`）

java

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(); // 其他线程正在初始化
        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
            try {
                if ((tab = table) == null || tab.length == 0) {
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                    table = tab = (Node<K,V>[])new Node<?,?>[n];
                    sizeCtl = n - (n >>> 2); // 阈值 = n * 0.75
                }
            } finally {
                sizeCtl = sc;
            }
            break;
        }
    }
    return tab;
}

④ `transfer()` 的并发扩容

transfer() 是 ConcurrentHashMap 最复杂的并发扩容方法：

java

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
    int n = tab.length, stride;
    // 计算每个线程处理的桶数（最小 16）
    if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
        stride = MIN_TRANSFER_STRIDE;

    if (nextTab == null) { // 初始化新表
        nextTab = (Node<K,V>[])new Node<?,?>[n << 1];
    }

    ForwardingNode<K,V> fwd = new ForwardingNode<>(nextTab);

    // 分区迁移
    for (int i = 0, bound = 0;;) {
        // ... 分配任务区间
        for (int j = i; j > bound; --j) {
            Node<K,V> f = tabAt(tab, j);
            if (f == null) {
                // 空桶 → CAS 设置 ForwardingNode
                if (casTabAt(tab, j, null, fwd))
                    break;
            } else if ((fh = f.hash) == MOVED) {
                // 该桶已被其他线程迁移
            } else {
                synchronized (f) { // 锁住桶迁移
                    // 链表/红黑树拆分（同 HashMap）
                    // 迁移完成后设置 ForwardingNode
                }
            }
        }
    }
}

stride：每个线程负责的桶数，CPU 多时减少
ForwardingNode：标记已迁移完成的桶
其他线程通过 helpTransfer() 检测到 ForwardingNode 后参与扩容

⑤ `helpTransfer()` 的多线程协作

java

final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
    Node<K,V>[] nextTab; int sc;
    if (tab != null && (f instanceof ForwardingNode) &&
        (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {
        int rs = resizeStamp(tab.length);
        while (nextTab == nextTable && table == tab &&
               (sc = sizeCtl) < 0) {
            if ((sc >>> RESIZE_STAMP_SHIFT) != rs ||
                sc == rs + 1 || sc == rs + MAX_RESIZERS ||
                transferIndex <= 0)
                break;
            if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
                transfer(tab, nextTab);
                break;
            }
        }
        return nextTab;
    }
    return table;
}

resizeStamp(n) 生成扩容标识戳，确保只有同一批扩容的线程协作
CAS 增加 sizeCtl 中的扩容线程计数

⑥ `CounterCell` 的计数累加

java

private final void addCount(long x, int check) {
    CounterCell[] cs; long b, s;
    if ((cs = counterCells) != null ||
        !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
        CounterCell c; long v; int m;
        boolean uncontended = true;
        if (cs == null || (m = cs.length - 1) < 0 ||
            (c = cs[ThreadLocalRandom.getProbe() & m]) == null ||
            !(uncontended =
              U.compareAndSwapLong(c, CELLVALUE, v = c.value, v + x))) {
            fullAddCount(x, uncontended); // CAS 竞争时的降级
            return;
        }
        // ...
    }
}

baseCount：主计数器（无竞争时 CAS 更新）
CounterCell[]：竞争时的分拆计数器（减少 CAS 冲突）
ThreadLocalRandom.getProbe() 为每个线程选择不同的 CounterCell

⑦ `TreeBin` 的读写锁

TreeBin 是 ConcurrentHashMap 中红黑树的封装，实现了轻量级读写锁：

java

static final class TreeBin<K,V> extends Node<K,V> {
    volatile TreeNode<K,V> root;
    volatile TreeNode<K,V> first;
    volatile int lockState; // 0=unlocked, 1=writer, 2=reader

    static final int WRITER = 1;
    static final int READER = 4;

    final TreeNode<K,V> find(int h, Object k) {
        if (root != null) {
            TreeNode<K,V> e = first;
            while (e != null) {
                int dir, ph; K ek;
                if ((ph = e.hash) == h &&
                    ((ek = e.key) == k || (ek != null && k.equals(ek))))
                    return e;
                if (++dir < 0) // 读锁标记
                    Thread.yield();
                // 无锁遍历（基于 volatile root）
            }
        }
        return null;
    }
}

读操作无锁（依靠 volatile 保证可见性）
写操作使用 synchronized 或 CAS lockState
读多写少的场景下性能极优

⑧ `size()` 的 `sumCount()`

java

public int size() {
    long n = sumCount();
    return (n < 0L) ? 0 : (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE : (int)n;
}

final long sumCount() {
    CounterCell[] cs = counterCells;
    long sum = baseCount;
    if (cs != null) {
        for (CounterCell c : cs)
            if (c != null)
                sum += c.value;
    }
    return sum;
}

遍历 CounterCell[] + baseCount 求和
不精确但高效（最终一致而非强一致）

⑨ `computeIfAbsent()` 的 `ReservationNode`

java

public V computeIfAbsent(K key, Function<? super K, ? extends V> mappingFunction) {
    int h = spread(key.hashCode());
    V val = null;
    for (Node<K,V>[] tab = table;;) {
        // ...
        if ((f = tabAt(tab, i = (n - 1) & h)) == null) {
            Node<K,V> r = new ReservationNode<K,V>();
            synchronized (r) { // 占位锁
                if (casTabAt(tab, i, null, r)) {
                    binCount = 1;
                    Node<K,V> node = null;
                    try {
                        val = mappingFunction.apply(key);
                        if (val != null) {
                            node = new Node<K,V>(h, key, val);
                        }
                    } finally {
                        setTabAt(tab, i, node);
                    }
                }
            }
        }
        // ...
    }
}

ReservationNode：占位节点，防止其他线程同时执行相同的 computeIfAbsent
保证 mappingFunction 只执行一次
没有 synchronized 的地毯式锁定

⑩ `keySet()` / `values()` / `entrySet()` 的视图

java

public KeySetView<K,V> keySet() {
    KeySetView<K,V> ks;
    return (ks = keySet) != null ? ks : (keySet = new KeySetView<K,V>(this, null));
}

KeySetView 是 ConcurrentHashMap 的静态内部类
视图上的操作（如 contains()、remove()）直接委托给 ConcurrentHashMap
迭代器是弱一致的（weakly consistent），不抛 ConcurrentModificationException

总结

ConcurrentHashMap 的 JDK 8+ 实现是并发编程的典范：

细粒度锁：CAS 无锁 + synchronized 桶锁，远优于 JDK 7 的分段锁
并发扩容：多线程协同迁移，ForwardingNode 标记已迁移桶
弱一致性：遍历时反映部分最新状态，不保证强一致
高性能计数：CounterCell 分拆减少 CAS 竞争
null 禁止：不允许 null 键/值，确保安全

在需要线程安全的哈希表场景中，ConcurrentHashMap 是首选方案。

ConcurrentHashMap 并发哈希表源码 ​

概述 ​

核心源码解析 ​

① JDK 8+ 的 CAS + synchronized 实现 ​

② put(K, V) 的 putVal() 流程 ​

③ sizeCtl 的多种状态 ​

④ transfer() 的并发扩容 ​

⑤ helpTransfer() 的多线程协作 ​

⑥ CounterCell 的计数累加 ​

⑦ TreeBin 的读写锁 ​

⑧ size() 的 sumCount() ​

⑨ computeIfAbsent() 的 ReservationNode ​

⑩ keySet() / values() / entrySet() 的视图 ​

总结 ​

`ConcurrentHashMap` 并发哈希表源码

概述

核心源码解析

① JDK 8+ 的 `CAS + synchronized` 实现

② `put(K, V)` 的 `putVal()` 流程

③ `sizeCtl` 的多种状态

④ `transfer()` 的并发扩容

⑤ `helpTransfer()` 的多线程协作

⑥ `CounterCell` 的计数累加

⑦ `TreeBin` 的读写锁

⑧ `size()` 的 `sumCount()`

⑨ `computeIfAbsent()` 的 `ReservationNode`

⑩ `keySet()` / `values()` / `entrySet()` 的视图

总结