--- url: https://www.chenlj.top/md/java-base/collections/ConcurrentHashMap.md --- # 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`，锁粒度极细。 ```java // 节点定义：与 HashMap.Node 最大的区别是 val 和 next 都用 volatile 修饰 static class Node implements Map.Entry { final int hash; final K key; volatile V val; // 可见性保证 volatile Node next; // 可见性保证 Node(int hash, K key, V val, Node 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` 等场景的临时占位。 ## 核心成员变量 ```java // 哈希桶数组，第一次插入时才懒初始化 transient volatile Node[] table; // 扩容时用的新表，容量为旧表的 2 倍 private transient volatile Node[] 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 来抢到「初始化资格」，避免多个线程重复建表。 ```java private final Node[] initTable() { Node[] 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[] nt = (Node[])new Node[n]; table = tab = nt; sc = n - (n >>> 2); // 阈值 = 容量 * 0.75 } } finally { sizeCtl = sc; // 恢复为扩容阈值 } break; } } return tab; } ``` ### 写入：put / putVal `put` 的核心逻辑是 `putVal`，是理解 `ConcurrentHashMap` 并发设计的重中之重。 ```java 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[] tab = table;;) { Node 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(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 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 pred = e; if ((e = e.next) == null) { pred.next = new Node(hash, key, value, null); break; } } } else if (f instanceof TreeBin) { // 红黑树 Node p; binCount = 2; if ((p = ((TreeBin)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`）保证可见性，因此读性能极高。 ```java public V get(Object key) { Node[] tab; Node 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 ```java import java.util.concurrent.ConcurrentHashMap; public class ConcurrentHashMapDemo { public static void main(String[] args) throws InterruptedException { ConcurrentHashMap 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 } } ```