ConcurrentHashMap 源码解析

简介

ConcurrentHashMap 位于 java.util.concurrent 包下，是 HashMap 的线程安全版本，专为高并发场景设计。

• JDK 1.7 及之前：采用「分段锁（Segment）」方案。内部维护一个 Segment 数组，每个 Segment 本质是一个小的 HashMap 并继承自 ReentrantLock，默认 16 个分段，理论上支持 16 个线程同时写（锁的粒度是「段」）。
• JDK 1.8 之后：废弃了 Segment，改为「数组 + 链表 / 红黑树 + CAS + synchronized」的实现，锁的粒度进一步细化到「桶（bin）的头节点」，并发度不再受固定分段数限制，性能大幅提升。

与 HashMap 不同，ConcurrentHashMap 不允许 null 键和 null 值（否则会抛出 NullPointerException）——这是为了避免在并发场景下无法区分「key 不存在」和「value 为 null」。

注意：ConcurrentHashMap 的迭代器是「弱一致（weakly consistent）」的：迭代过程中允许其他线程修改，不会抛出 ConcurrentModificationException，但也不保证一定看到最新的值。

底层数据结构分析

JDK 1.8 的 ConcurrentHashMap 整体结构与 HashMap 类似：一个 Node 数组（哈希桶），每个桶存放链表或红黑树。区别在于它全程不使用全局锁，而是依靠以下三件套保证并发安全：

1. volatile 变量：table、next、val 等字段用 volatile 修饰，保证可见性。
2. CAS（Compare-And-Swap）：初始化、空桶插入、计数等无竞争场景用 Unsafe.compareAndSwap* 原子操作。
3. synchronized 块：发生冲突（桶非空）时，只对当前桶的头节点加 synchronized，锁粒度极细。

// 节点定义：与 HashMap.Node 最大的区别是 val 和 next 都用 volatile 修饰
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;
    }
}

此外还有几个特殊节点类型：

• TreeBin：红黑树的根节点（hash 固定为 TREEBIN = -2），内部持有一棵红黑树，并代理读写。
• TreeNode：红黑树节点（并不直接挂在桶上，而是被 TreeBin 包裹）。
• ForwardingNode：扩容时的「转发节点」（hash 为 MOVED = -1），标记某个桶已经迁移到新表，并指向 nextTable，让读线程能转发到新表、写线程能参与扩容。
• ReservationNode：占位节点（hash 为 RESERVED = -3），用于 computeIfAbsent 等场景的临时占位。

核心成员变量

// 哈希桶数组，第一次插入时才懒初始化
transient volatile Node<K,V>[] table;

// 扩容时用的新表，容量为旧表的 2 倍
private transient volatile Node<K,V>[] nextTable;

// 基础计数器（无竞争时直接累加这里）
private transient volatile long baseCount;

// 控制变量，是整个并发控制的「大脑」：
//   > 0    ：初始化完成后的扩容阈值（或初始容量）
//   = 0    ：未初始化
//   = -1   ：正在初始化
//   < -1   ：正在扩容，值为 -(1 + 正在参与扩容的线程数)
private transient volatile int sizeCtl;

// 扩容时，多个线程领取任务用的下标（从后往前分配桶区间）
private transient volatile int transferIndex;

// 计数单元格数组（竞争激烈时在 baseCount 之外分担计数）
private transient volatile CounterCell[] counterCells;

// 自旋锁标记，用于保护 counterCells 的初始化 / 扩容
private transient volatile int cellsBusy;

源码分析

初始化：initTable

table 是懒初始化的，首次 put 时通过 initTable 完成。它用 CAS 把 sizeCtl 从 0 改成 -1 来抢到「初始化资格」，避免多个线程重复建表。

private final Node<K,V>[] initTable() {
    Node<K,V>[] tab; int sc;
    while ((tab = table) == null || tab.length == 0) {
        // sizeCtl < 0 说明已经有线程在初始化或扩容，让出 CPU 自旋等待
        if ((sc = sizeCtl) < 0)
            Thread.yield();
        // CAS 把 sizeCtl 设为 -1，抢到初始化资格
        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
            try {
                if ((tab = table) == null || tab.length == 0) {
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY; // 默认 16
                    @SuppressWarnings("unchecked")
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                    table = tab = nt;
                    sc = n - (n >>> 2); // 阈值 = 容量 * 0.75
                }
            } finally {
                sizeCtl = sc; // 恢复为扩容阈值
            }
            break;
        }
    }
    return tab;
}

写入：put / putVal

put 的核心逻辑是 putVal，是理解 ConcurrentHashMap 并发设计的重中之重。

final V putVal(K key, V value, boolean onlyIfAbsent) {
    if (key == null || value == null) throw new NullPointerException(); // 不允许 null
    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();                 // 1. 表为空先初始化
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
            // 2. 目标桶为空：CAS 直接塞入，无需加锁（无竞争快速路径）
            if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null)))
                break;
        }
        else if ((fh = f.hash) == MOVED)
            tab = helpTransfer(tab, f);         // 3. 正在扩容：帮忙迁移数据
        else {
            // 4. 桶非空且有竞争：只对「头节点 f」加 synchronized
            V oldVal = null;
            synchronized (f) {
                if (tabAt(tab, i) == f) {       // 二次检查，防止 f 已被替换
                    if (fh >= 0) {              // 链表
                        binCount = 1;
                        for (Node<K,V> e = f;; ++binCount) {
                            K ek;
                            if (e.hash == hash &&
                                ((ek = e.key) == key || (ek != null && key.equals(ek)))) {
                                oldVal = e.val;
                                if (!onlyIfAbsent)
                                    e.val = value; // val 是 volatile，写立即可见
                                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;
                        }
                    }
                }
            }
            if (binCount != 0) {
                // 链表长度达到 8 时转为红黑树
                if (binCount >= TREEIFY_THRESHOLD)
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    // 5. 更新元素计数（baseCount / counterCells），可能触发扩容
    addCount(1L, binCount);
    return null;
}

要点：

• 读写分离加锁：空桶用 CAS（无锁），非空桶才 synchronized(f) 且只锁头节点，所以不同桶的写可以完全并行。
• spread 扰动：(h ^ (h >>> 16)) & HASH_BITS，HASH_BITS = 0x7fffffff 保证 hash 非负，避免与 MOVED、TREEBIN 等特殊标记冲突。

读取：get

get 全程不加锁，完全依赖 volatile 读和原子读（tabAt 内部用 Unsafe.getObjectVolatile）保证可见性，因此读性能极高。

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;
        }
        else if (eh < 0)                        // 负数 hash：红黑树 / 扩容转发节点
            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;
}

注意 e.find(h, key) 的多态行为：ForwardingNode.find 会去 nextTable 里查找（扩容期间读不阻塞）；TreeBin.find 会在红黑树中查找。

计数与扩容：addCount / transfer

元素总数不再用一个 size 字段，而是 baseCount + 所有 CounterCell 的 value 之和（sumCount），避免计数成为性能瓶颈。

• addCount：先用 CAS 累加到 baseCount；若竞争激烈（CAS 失败），再累加到 counterCells 中的某个单元格。size() / mappingCount() 就是把这两部分加起来，是一个近似值，但性能好且并发安全的计数。
• transfer（扩容）：新表容量为旧表的 2 倍。通过共享的 transferIndex 把旧表桶区间「从后往前」分配给参与的线程，每个线程认领一段并迁移；旧桶迁移完会替换为 ForwardingNode，从而让其他线程读 / 写时知道要去新表。sizeCtl 用 -(1 + 参与线程数) 记录正在扩容的线程数。

与 HashMap / Hashtable 的对比

维度	HashMap	Hashtable	ConcurrentHashMap (JDK8)
线程安全	否	是（方法全 synchronized）	是（CAS + 桶级 synchronized）
锁粒度	无	整个表（方法级）	单个桶头节点
null 键值	允许	不允许	不允许
并发度	1	1	高（按桶并行）
数据结构	数组+链表/红黑树	数组+链表	数组+链表/红黑树+CAS
size()	O(1) 精确	O(1) 精确	O(并发) 近似（弱一致）

Hashtable 因为锁的是整个方法，并发下性能极差，已被 ConcurrentHashMap 全面取代；Collections.synchronizedMap 同样是全局锁，仅适合并发度很低的场景。

经典 Demo

import java.util.concurrent.ConcurrentHashMap;

public class ConcurrentHashMapDemo {
    public static void main(String[] args) throws InterruptedException {
        ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();

        // 1. 基础读写（不允许 null）
        map.put("a", 1);
        System.out.println(map.get("a")); // 1

        // 2. 原子复合操作：putIfAbsent / computeIfAbsent
        map.putIfAbsent("b", 2);              // 仅当不存在时才放入
        map.computeIfAbsent("c", k -> 3);     // 不存在则计算并放入
        System.out.println(map);              // {a=1, b=2, c=3}

        // 3. 多线程并发写入示例
        int threadCount = 10;
        Thread[] threads = new Thread[threadCount];
        for (int t = 0; t < threadCount; t++) {
            threads[t] = new Thread(() -> {
                for (int i = 0; i < 100; i++) {
                    // 线程安全地累加计数（merge 是原子复合操作）
                    map.merge("counter", 1, Integer::sum);
                }
            });
            threads[t].start();
        }
        for (Thread th : threads) {
            th.join();
        }

        // 4. 并发累加结果应为 1000，且不会丢更新
        System.out.println(map.get("counter")); // 1000
    }
}