ConcurrentHashMap 源码分析

JavaJuice约 1288 字大约 4 分钟

ConcurrentHashMap 源码分析

1. 概述

ConcurrentHashMap 是 Java 集合框架中一个线程安全的 Map 实现类，主要用于高并发场景。相比 HashMap，ConcurrentHashMap 采用了锁分段和无锁优化策略，在多线程环境下提供高效的键值对操作。ConcurrentHashMap 的实现因 JDK 版本不同而有所变化，以下以 Java 8 为主要分析版本。

2. ConcurrentHashMap 的存储结构

Java 8 中的 ConcurrentHashMap 采用了以下存储结构：

数组 + 链表 + 红黑树 的组合。
每个桶（table 的一个槽）存储的是 Node 对象，当发生哈希冲突时，链表或红黑树会存储冲突的元素。
当链表长度超过阈值（TREEIFY_THRESHOLD=8）时，链表会转化为红黑树；当冲突降低到阈值以下时，红黑树会重新转化为链表。

3. 主要成员变量

transient volatile Node<K,V>[] table;  // 主存储表，存储键值对
private transient volatile int sizeCtl; // 控制初始化和扩容的标志位

table：存储键值对的主数组，每个元素是一个链表的头节点或红黑树的根节点。
sizeCtl：控制数组的初始化和扩容。
- -1：表示当前正在初始化。
- 负数：高 16 位记录扩容戳，低 16 位表示正在扩容的线程数。
- 正数：表示下一次扩容的触发阈值。

4. 核心方法

4.1 初始化（`initTable` 方法）

ConcurrentHashMap 的初始化使用了懒加载策略，表在第一次插入元素时才真正分配内存。以下是核心逻辑：

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;
                    @SuppressWarnings("unchecked")
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                    table = tab = nt;
                    sc = n - (n >>> 2); // 扩容阈值为 0.75 * n
                }
            } finally {
                sizeCtl = sc; // 更新扩容阈值
            }
            break;
        }
    }
    return tab;
}

使用 CAS 确保只有一个线程能够初始化表，其余线程通过 Thread.yield 让出 CPU。
初始化完成后更新 sizeCtl 为扩容阈值。

4.2 插入（`put` 方法）

put 方法是 ConcurrentHashMap 的核心方法，以下是主要实现逻辑：

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;
        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<>(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;
                            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, 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) {
                if (binCount >= TREEIFY_THRESHOLD) // 链表转为红黑树
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    addCount(1L, binCount); // 更新大小并检查是否需要扩容
    return null;
}

关键步骤：

计算哈希值：
使用 spread 方法对键的哈希值进行扰动，尽量保证哈希值的高位和低位都有变化。
插入节点：
- 如果目标桶为空，直接插入新节点（使用 CAS 保证线程安全）。
- 如果目标桶存在链表，使用 synchronized 锁住该桶并插入。
- 如果目标桶为红黑树，调用红黑树的插入方法。
扩容检查：
插入完成后，更新计数并检查是否需要扩容。如果需要扩容，则触发 helpTransfer 方法。

4.3 扩容（`resize` 方法）

当表中元素数量超过阈值时，会触发扩容操作，将表容量扩展为原来的两倍。

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
    int n = tab.length, stride;
    if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
        stride = MIN_TRANSFER_STRIDE; // 最小步长
    if (nextTab == null) {            // 初始化扩容表
        try {
            @SuppressWarnings("unchecked")
            Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
            nextTab = nt;
        } catch (Throwable ex) {      // OOME
            sizeCtl = Integer.MAX_VALUE;
            return;
        }
        nextTable = nextTab;
        transferIndex = n;
    }
    int nextn = nextTab.length;
    Node<K,V> f;
    while (transferIndex > 0) {
        // 遍历老表，将每个槽的链表或红黑树重新分配到新表中
    }
    table = nextTab;
    sizeCtl = (n << 1) - (n >>> 1); // 更新扩容阈值
}

5. 总结

线程安全性：
- 使用 CAS 和 synchronized 实现高效的线程安全。
- 扩容和插入过程均允许多线程协作，避免锁定整个表。
性能优化：
- 分段锁改进：相比 Java 7 的 Segment 分段锁，Java 8 使用了细粒度锁（对链表节点加锁），进一步提升了并发性能。
- 红黑树优化：当冲突链表过长时转为红黑树，减少查找时间复杂度。
适用场景：
ConcurrentHashMap 适用于高并发场景下的键值对存储，尤其在多线程环境中频繁读写操作时，能够提供较好的性能表现。