ConcurrentHashMap 线程安全原理

ConcurrentHashMap 是 Java 并发编程中最重要的线程安全容器。在 HashMap 一文中，我们知道 HashMap 不是线程安全的——多线程并发写入会导致数据丢失甚至死循环。ConcurrentHashMap 就是为解决这个问题而生的：它既保证了线程安全，又通过精细化的锁设计避免了全表加锁的性能瓶颈。

想象一个大型图书馆：Hashtable 的做法是整个图书馆同一时间只允许一个人进出（全表锁），而 ConcurrentHashMap 的做法是每个书架独立管理——你在 A 书架找书，我在 B 书架放书，互不干扰。JDK 1.7 和 1.8 分别用不同的方式实现了这种"分区管理"。

为什么不用 Hashtable 和 synchronizedMap

在 ConcurrentHashMap 之前，Java 提供了两种线程安全的 Map：

// 方案一：Hashtable（JDK 1.0）
Map<String, String> map = new Hashtable<>();

// 方案二：Collections.synchronizedMap（JDK 1.2）
Map<String, String> map = Collections.synchronizedMap(new HashMap<>());

这两者的问题是一样的——锁粒度太粗。它们对每个方法都加了 synchronized，锁住的是整个 Map 对象：

// Hashtable 源码（简化）
public synchronized V get(Object key) { ... }
public synchronized V put(K key, V value) { ... }

这意味着：

• 线程 A 在读 key1，线程 B 想读 key2 —— 也要等（读读互斥）
• 线程 A 在写 key1，线程 B 想写 key99 —— 也要等（明明操作的是不同的桶）
• 所有操作都串行执行，并发度 = 1

这就像整个图书馆只有一把大门钥匙，不管你去哪个书架，都要排队等钥匙。在高并发场景下，这种设计是不可接受的。

JDK 1.7 实现：分段锁

核心思想

JDK 1.7 的解决方案是分段锁（Segment Lock）：把一个大的 Map 拆成多个小的 HashMap，每个小 Map 有自己独立的锁。

继续用图书馆的比喻：不再用一把大门钥匙管理整个馆，而是把图书馆分成 16 个阅览区，每个区有自己的门锁。不同区可以同时服务不同的读者。

数据结构：两层数组 + 链表

JDK 1.7 的 ConcurrentHashMap 本质上是一个两层数组 + 单向链表的结构：

• 第一层数组：Segment[]（段数组，默认 16 个段）
• 第二层数组：每个 Segment 内部的 HashEntry[]（桶数组）
• 第三层链表：每个桶是一条 HashEntry 单向链表，处理 hash 冲突

// 第一层：Segment 数组
final Segment<K,V>[] segments;  // 默认 16 个段

// 每个 Segment 内部有自己的桶数组、计数器、阈值、负载因子
// 本质上就是一个独立的小型哈希表，同时继承 ReentrantLock 提供锁能力
static class Segment<K,V> extends ReentrantLock {
    transient volatile HashEntry<K,V>[] table;  // 第二层：桶数组
    transient int count;       // 元素个数
    transient int threshold;   // 扩容阈值
    final float loadFactor;    // 负载因子
}

// 链表节点（不可变单向链表）
static class HashEntry<K,V> {
    final int hash;
    final K key;
    volatile V value;              // volatile！保证读的可见性
    final HashEntry<K,V> next;     // final！节点创建后 next 指针不可变
}

整体结构如下：

第一层数组              第二层数组                链表
Segment[]              HashEntry[]             HashEntry → HashEntry → ...
┌─────────────┐
│ Segment[0]  │ ──→ ┌──────────┐
│ (锁独立)    │     │ table[0] │ ──→ Entry → Entry → null
│             │     │ table[1] │ ──→ null
│             │     │ table[2] │ ──→ Entry → null
│             │     └──────────┘
├─────────────┤
│ Segment[1]  │ ──→ ┌──────────┐
│ (锁独立)    │     │ table[0] │ ──→ Entry → null
│             │     │ table[1] │ ──→ Entry → Entry → Entry → null
│             │     └──────────┘
├─────────────┤
│ ...         │     （共 16 个 Segment，每个都有自己的桶数组）
├─────────────┤
│ Segment[15] │ ──→ ┌──────────┐
│ (锁独立)    │     │ table[0] │ ──→ ...
│             │     └──────────┘
└─────────────┘

HashEntry.next 为什么是 final 的？

这是一个重要的设计决策。next 被声明为 final，意味着节点一旦创建，它指向哪个下一节点就永远不可改变。这带来两个影响：

1. 插入只能用头插法：新节点必须插在链表头部（new HashEntry(key, value, currentHead)），因为无法修改已有节点的 next 指针
2. 删除需要复制：删除链表中间的节点时，需要把该节点之前的所有节点重新复制一遍来重建链表

这么做的好处是读操作更安全：由于 next 不可变，读线程遍历链表时不会遇到"next 突然指向了别的地方"的情况，天然保证了遍历的一致性。

put 流程：两次定位

put(key, value) 流程：

1. 计算 key 的 hash 值
2. 第一次定位：hash 的高位 → 确定落在哪个 Segment
   segmentIndex = (hash >>> segmentShift) & segmentMask
3. 对该 Segment 加锁（segment.lock()）
4. 第二次定位：hash 的低位 → 确定 Segment 内的桶位置
   bucketIndex = hash & (table.length - 1)
5. 在桶的链表中查找/插入（和 HashMap 一样）
6. 释放锁（segment.unlock()）

关键点：只有操作同一个 Segment 的线程才会互相阻塞。操作不同 Segment 的线程完全并行。

get 流程：无需加锁

get 操作不需要加锁。因为 HashEntry 的 value 和 next 都用 volatile 修饰，根据 Java 内存模型的 happens-before 规则，写线程对 volatile 变量的修改，对读线程立刻可见。

get(key) 流程：

1. 计算 hash → 定位 Segment → 定位桶
2. 遍历链表，通过 equals() 匹配 key
3. 直接读取 value（volatile 保证可见性）
4. 全程无锁

并发度与局限

并发度：默认 16 个 Segment，最多支持 16 个线程同时写入不同 Segment。

局限性：

这些局限促使 JDK 1.8 对 ConcurrentHashMap 进行了彻底重写。

JDK 1.8 实现：CAS + synchronized

设计思路的转变

JDK 1.8 彻底抛弃了 Segment 分段锁，改为一层哈希表结构（和 HashMap 完全一样），锁的粒度从"一组桶"细化到"单个桶"。

用图书馆的比喻来说：不再把图书馆分成 16 个阅览区各设门锁，而是给每一排书架装一把锁。你锁 A 排放书，完全不影响我在 B 排找书。

数据结构

// 和 HashMap 1.8 完全一样：数组 + 链表 + 红黑树
transient volatile Node<K,V>[] table;

// 新数组（扩容时使用）
private transient volatile Node<K,V>[] nextTable;

// 控制变量：初始化和扩容的核心协调器
private transient volatile int sizeCtl;

// 节点结构
static class Node<K,V> {
    final int hash;
    final K key;
    volatile V val;            // 注意：volatile！
    volatile Node<K,V> next;   // 注意：volatile！
}

对比 HashMap 的 Node，ConcurrentHashMap 的 Node 有两个关键区别：

sizeCtl：多功能的控制变量

sizeCtl 是 ConcurrentHashMap 中最巧妙的设计之一——一个变量在不同阶段承担不同含义：

这种"一变量多含义"的设计虽然增加了理解难度，但避免了额外的 volatile 变量开销。在并发场景下，每个 volatile 变量都有内存可见性代价，减少 volatile 变量数量是一种有意的性能优化。

put 全流程：源码级解析

put 操作是理解 ConcurrentHashMap 并发设计的核心。整个流程是一个自旋循环（for 死循环），根据桶的状态选择不同的处理策略：

final V putVal(K key, V value, boolean onlyIfAbsent) {
    // ① 不允许 null key 或 null 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 = initTable();

        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null)
            // ⑤ 情况二：桶为空 → CAS 无锁写入
            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 {
            // ⑦ 情况四：桶不为空 → synchronized 锁桶头
            V oldVal = null;
            synchronized (f) {
                if (tabAt(tab, i) == 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;
                                break;
                            }
                            Node<K,V> pred = e;
                            if ((e = e.next) == null) {
                                pred.next = new Node<>(hash, key, value);
                                break;
                            }
                        }
                    } else if (f instanceof TreeBin) {
                        // 红黑树操作
                        binCount = 2;
                        // ...
                    }
                }
            }
            if (binCount != 0) {
                if (binCount >= TREEIFY_THRESHOLD)
                    treeifyBin(tab, i);  // 链表 → 红黑树
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    addCount(1L, binCount);  // ⑧ 更新计数
    return null;
}

下面逐一讲解每种情况的设计逻辑。

情况一：数组未初始化（initTable）

ConcurrentHashMap 采用懒初始化——构造函数不创建数组，第一次 put 时才初始化。

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();  // 其他线程正在初始化，让出 CPU
        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
            // CAS 将 sizeCtl 设为 -1，表示"我来初始化"
            try {
                if ((tab = table) == null || tab.length == 0) {
                    // 双重检查后创建数组
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                    Node<K,V>[] nt = new Node[n];
                    table = tab = nt;
                    sc = n - (n >>> 2);  // 等价于 0.75n，即扩容阈值
                }
            } finally {
                sizeCtl = sc;  // 恢复为扩容阈值
            }
            break;
        }
    }
    return tab;
}

设计要点：

1. CAS 抢占初始化权：多个线程同时 put 时，只有一个线程能 CAS 成功（sizeCtl 从 ≥0 改为 -1），其他线程 yield 让出 CPU 等待
2. 双重检查：CAS 成功后再检查一次 table == null，防止极端情况下的重复初始化
3. 阈值计算：n - (n >>> 2) = n - n/4 = 0.75n，用位运算替代浮点乘法

情况二：桶为空——CAS 无锁写入

如果目标桶是空的（还没有任何节点），直接用 CAS 原子操作写入新节点，完全不需要加锁：

// tabAt：通过 Unsafe 直接读取数组元素（volatile 语义）
// casTabAt：通过 CAS 原子地设置数组元素
if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value)))
    break;  // 成功！
// CAS 失败说明其他线程抢先写入了，进入下一轮循环重新判断

为什么空桶可以不加锁？

因为 CAS 本身就是原子操作。当多个线程同时向同一个空桶写入时，只有一个线程的 CAS 会成功（把 null 替换为新节点），其他线程的 CAS 会失败（发现桶已经不是 null），然后进入下一轮循环，走"桶不为空"的逻辑。CAS 失败的代价很低——不需要线程上下文切换，只是重试一次循环。

情况三：正在扩容——帮助迁移

如果发现桶头节点的 hash 值是 MOVED（-1），说明这个桶已经被迁移到新数组了。当前线程不会傻等，而是主动加入扩容，帮助迁移其他桶的数据。这个机制会在后面的"扩容"章节详细讲解。

情况四：桶不为空——synchronized 锁桶头

当桶已经有节点时，需要遍历链表或操作红黑树，这些操作不是原子的，必须加锁。但锁的范围仅限于这个桶的头节点：

synchronized (f) {  // f 是桶头节点
    if (tabAt(tab, i) == f) {  // 双重检查：确认头节点没有被其他线程改过
        // 链表或红黑树操作...
    }
}

synchronized 内部的双重检查：为什么拿到锁之后还要再检查 tabAt(tab, i) == f？因为在等待锁的过程中，其他线程可能已经完成了扩容，把这个桶的数据迁移到了新数组，头节点已经变成了 ForwardingNode。如果不检查就直接操作，可能会往旧数组写数据，导致数据丢失。

三种同步策略的设计哲学

这三种策略的选择遵循一个原则：能不加锁就不加锁，必须加锁就锁最小的范围。空桶用 CAS，非空桶只锁一个桶头节点（不锁整个数组），读操作靠 volatile 完全免锁。

get 操作为什么不需要加锁

ConcurrentHashMap 的 get 操作全程无锁，这是它高性能的关键之一。能做到无锁的原因有三层保障：

第一层：volatile 数组引用

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

table 本身是 volatile 的，这保证了扩容后新数组的引用对所有线程立即可见。

第二层：volatile 节点字段

static class Node<K,V> {
    final int hash;       // final → 不可变
    final K key;          // final → 不可变
    volatile V val;       // volatile → 写后立即可见
    volatile Node<K,V> next;  // volatile → 链表变化立即可见
}

• key 和 hash 是 final 的，创建后不可变，天然线程安全
• val 是 volatile 的，写线程修改后，读线程立即可见
• next 是 volatile 的，链表的结构变化（插入新节点）对读线程立即可见

第三层：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);
}

tabAt 使用 Unsafe.getObjectVolatile 直接从内存读取数组元素，绕过 CPU 缓存，保证读到的是最新值。普通的数组访问 tab[i] 可能读到 CPU 缓存中的旧值，而 getObjectVolatile 会插入一条**内存屏障（load barrier）**指令，强制从主内存读取。

这三层保障的效果是：get 操作总能读到一个一致的快照——要么是最新值，要么是某个时刻的完整值，不会读到"写了一半"的中间状态。

扩容机制：多线程并行迁移

ConcurrentHashMap 的扩容是整个实现中最复杂也最精巧的部分。与 HashMap 单线程扩容不同，ConcurrentHashMap 支持多个线程同时参与数据迁移，大幅加速扩容过程。

扩容触发

当 put 操作完成后，addCount 方法会检查元素总数是否超过阈值（容量 × 0.75）。如果超过，就触发扩容。

核心组件

扩容机制依赖三个关键组件：

分段认领机制

想象一条流水线上有 64 个零件需要搬运。不是一个工人从头搬到尾，而是多个工人各自认领一段：

旧数组（64 个桶）：
[0] [1] [2] ... [15] | [16] ... [31] | [32] ... [47] | [48] ... [63]
                                                             ↑
                                                       transferIndex = 64（起始）

Thread-1 认领 [48-63]：transferIndex CAS 改为 48
Thread-2 认领 [32-47]：transferIndex CAS 改为 32
Thread-3 认领 [16-31]：transferIndex CAS 改为 16

每个线程独立迁移自己的那段桶，互不干扰。

每个线程通过 CAS 操作原子地修改 transferIndex 来"认领"一段桶。步长（stride）的计算考虑了 CPU 核数：

// 步长计算：至少 16 个桶
int stride = (NCPU > 1) ? (n >>> 3) / NCPU : n;
if (stride < MIN_TRANSFER_STRIDE)  // MIN_TRANSFER_STRIDE = 16
    stride = MIN_TRANSFER_STRIDE;

单个桶的迁移过程

每个桶的迁移逻辑和 HashMap 类似——利用 hash 的高位 bit 判断元素去新数组的哪个位置：

旧容量 n = 16，新容量 = 32

对于旧桶 i = 5 中的每个节点，检查 (hash & n)：
├── == 0 → 留在新数组的位置 5（低位链表 ln）
└── != 0 → 移到新数组的位置 5 + 16 = 21（高位链表 hn）

// transfer 方法中的链表拆分（简化）
synchronized (f) {
    // 把旧桶的链表拆成两条：低位链表和高位链表
    Node<K,V> ln = null, hn = null;
    for (Node<K,V> e = f; e != null; e = e.next) {
        if ((e.hash & n) == 0)
            ln = new Node<>(e.hash, e.key, e.val, ln);  // 低位
        else
            hn = new Node<>(e.hash, e.key, e.val, hn);  // 高位
    }
    // 写入新数组
    setTabAt(nextTab, i, ln);       // 低位链表放到位置 i
    setTabAt(nextTab, i + n, hn);   // 高位链表放到位置 i + n
    // 旧桶放入 ForwardingNode，标记为"已迁移"
    setTabAt(tab, i, fwd);
}

ForwardingNode：路标节点

当一个桶的数据迁移完成后，旧桶中会放入一个 ForwardingNode（hash = -1），它有两个作用：

1. 路标：告诉其他线程"这个桶已经搬走了，去新数组找"
2. 协助触发器：其他线程的 put 操作发现 ForwardingNode 时，会主动加入扩容

// ForwardingNode 的结构
static final class ForwardingNode<K,V> extends Node<K,V> {
    final Node<K,V>[] nextTable;  // 指向新数组

    ForwardingNode(Node<K,V>[] tab) {
        super(MOVED, null, null, null);  // hash = MOVED = -1
        this.nextTable = tab;
    }

    // find 方法：在新数组中查找
    Node<K,V> find(int h, Object k) {
        return nextTable[...];  // 重定向到新数组
    }
}

扩容全景图

时间线：
──────────────────────────────────────────────────────────→

Thread-1（触发扩容）:
  创建 nextTable → 认领 [48-63] → 迁移中...
                                   ↓ 旧桶[48] 放入 ForwardingNode
                                   ↓ 旧桶[49] 放入 ForwardingNode
                                   ↓ ...
                                   ↓ 完成 [48-63] → 认领 [0-15] → 迁移中...

Thread-2（put 操作，发现 ForwardingNode）:
  put(key) → 桶头是 ForwardingNode → helpTransfer()
           → 认领 [32-47] → 迁移中...

Thread-3（put 操作，发现 ForwardingNode）:
  put(key) → 桶头是 ForwardingNode → helpTransfer()
           → 认领 [16-31] → 迁移中...

所有桶迁移完毕后：
  table = nextTable  ← 切换到新数组
  nextTable = null   ← 清空

迁移过程中的 get 操作：
  ├── 旧桶有数据 → 直接读旧桶
  └── 旧桶是 ForwardingNode → 通过 fwd.find() 去新数组读

这种设计保证了：

• put 不会被扩容阻塞：发现扩容就帮忙，帮完再继续自己的 put
• get 不受扩容影响：不管数据在旧数组还是新数组，都能正确读到
• 多线程加速：N 个线程一起迁移，总耗时约为单线程的 1/N

size() 的计数原理

在并发环境下，统计元素个数是一个看似简单实则困难的问题。如果用一个全局计数器，所有线程的 put/remove 都要争抢这一个变量，就会成为性能瓶颈。

ConcurrentHashMap 采用了和 LongAdder 相同的思想——分散竞争。

计数组件

// 主计数器（低竞争时使用）
private transient volatile long baseCount;

// 分散计数器（高竞争时使用）
private transient volatile CounterCell[] counterCells;

// CounterCell 结构
static final class CounterCell {
    volatile long value;
}

addCount：计数更新流程

put 完成后调用 addCount(1L, ...)：

1. 先尝试 CAS 更新 baseCount
   ├── 成功 → 结束（低并发时走这条路，零额外开销）
   └── 失败 → 说明有竞争，进入分散计数

2. 从 counterCells 数组中根据线程 hash 选一个 CounterCell
   ├── CAS 更新该 Cell 的 value → 成功则结束
   └── 失败 → 重新选一个 Cell 或扩容 counterCells 数组

用比喻来说：这就像超市收银。人少的时候（低并发），一个收银台（baseCount）就够了。人多的时候（高并发），开多个收银台（CounterCell[]），每个顾客去不同的收银台结账，最后汇总营收。

size()：汇总求和

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

final long sumCount() {
    CounterCell[] as = counterCells;
    long sum = baseCount;  // 从 baseCount 开始
    if (as != null) {
        for (CounterCell a : as)
            if (a != null)
                sum += a.value;  // 加上每个 Cell 的值
    }
    return sum;
}

注意：size() 返回的是一个近似值。因为在求和过程中，其他线程可能正在 put/remove，导致 sum 可能已经过时。这是 ConcurrentHashMap 的设计取舍——用精确性换取了性能。如果需要精确计数，应该在外部加锁。

不允许 null key/value 的原因

HashMap 允许 null key 和 null value，但 ConcurrentHashMap 严格禁止。这不是随意的限制，而是由并发环境的本质决定的。

问题的本质是语义歧义：

V value = map.get(key);
if (value == null) {
    // 在单线程中，可以用 containsKey 区分：
    // 情况 1：key 不存在 → containsKey 返回 false
    // 情况 2：key 存在但 value 就是 null → containsKey 返回 true

    // 但在并发环境中：
    if (map.containsKey(key)) {
        // 走到这里时，另一个线程可能已经 remove 了这个 key
        // 你以为"key 存在，value 是 null"
        // 实际上 key 已经不存在了
    }
}

在单线程的 HashMap 中，get 返回 null 可以通过 containsKey 来消歧。但在并发环境中，containsKey 和 get 不是原子操作——它们之间可能有其他线程修改了 Map。Doug Lea（ConcurrentHashMap 的作者）选择直接禁止 null，从源头消除这种歧义。

为什么从 ReentrantLock 改为 synchronized

JDK 1.7 的 Segment 继承自 ReentrantLock，而 JDK 1.8 改用了 synchronized。这看起来是"退步"（synchronized 传统上被认为比 ReentrantLock 慢），但实际上是深思熟虑的选择：

在 JDK 1.8 的场景中，锁只保护一个桶的操作（通常很快），持有时间极短。在这种"短持有时间"的场景下，synchronized 的性能已经足够好，而它带来的内存节省和代码简洁性优势更加明显。

JDK 1.7 vs 1.8 全面对比

ConcurrentHashMap vs Hashtable

**弱一致性（fail-safe）**说明：ConcurrentHashMap 的迭代器基于"快照"思想，不会抛出 ConcurrentModificationException。但迭代过程中的修改可能部分可见、部分不可见。如果业务要求强一致迭代，需要在外部加锁。

常用的原子复合操作

ConcurrentHashMap 提供了一组原子性的复合操作，避免了"先检查再执行"的竞态条件：

// ❌ 非原子操作（有竞态条件）
if (!map.containsKey(key)) {
    map.put(key, value);  // 两步之间可能有其他线程插入
}

// ✅ 原子操作
map.putIfAbsent(key, value);  // 原子地"不存在则插入"

// 使用 compute 做线程安全的词频统计
ConcurrentHashMap<String, Integer> wordCount = new ConcurrentHashMap<>();
words.forEach(word ->
    wordCount.compute(word, (k, v) -> v == null ? 1 : v + 1)
);

使用场景选型