ConcurrentHashMap源码深度剖析:这才是线程安全的正确姿势

ConcurrentHashMap (JDK 7)│├─► Segment[]   ← 继承 ReentrantLock（可重入锁）│    ││    ├─► Segment 0│    │    └─► HashEntry[]  ← 真正的数据数组│    │         ├─► HashEntry → HashEntry → HashEntry（链表）│    │         └─► ...│    ││    ├─► Segment 1│    │    └─► HashEntry[] │    │         └─► ...│    ││    └─► Segment 2│         └─► ...└─► ...

// JDK 7 ConcurrentHashMap 的核心参数public class ConcurrentHashMap<K,V> {    // 默认初始容量：16    static final int DEFAULT_INITIAL_CAPACITY = 16;    // 默认负载因子：0.75    static final float DEFAULT_LOAD_FACTOR = 0.75f;    // 默认并发级别：16（决定了Segment数组的长度）    static final int DEFAULT_CONCURRENCY_LEVEL = 16;    // Segment数组，长度是并发级别    final Segment<K,V>[] segments;    // 一个Segment内部的数据结构    static final class Segment<K,V> extends ReentrantLock {        transient volatile HashEntry<K,V>[] table;        transient int count;        // 元素个数        transient int modCount;      // 修改次数        transient int threshold;     // 扩容阈值        final float loadFactor;      // 负载因子    }}

定位元素需要两次哈希：key │ ▼第一次哈希：找到对应的 Segment │   int segmentIndex = (hash >>> segmentShift) & segmentMask │ ▼第二次哈希：找到 Segment 内的 HashEntry 数组索引 │   int entryIndex = hash & (table.length - 1) │ ▼遍历链表找到目标元素

put(key, value)    │    ▼计算key的hash值    │    ▼根据hash找到对应的Segment    │    ▼尝试获取Segment的锁    ├───── 获取成功 ────► 执行写入    │    └───── 获取失败 ────► 自旋等待 + 阻塞等待（自旋次数达到阈值）                                  │                                  ▼                           获取锁后执行写入

// JDK 7 put 方法核心逻辑（简化）public V put(K key, V value) {    int hash = hash(key);    // 找到对应的Segment    Segment<K,V> segment = segmentForHash(hash);    // 调用Segment的put方法    return segment.put(key, hash, value, false);}// Segment内部的put方法V put(K key, int hash, V value, boolean onlyIfAbsent) {    // 尝试获取锁（继承自ReentrantLock）    lock();    try {        // 写入逻辑...        HashEntry<K,V>[] tab = table;        int index = hash & (tab.length - 1);        HashEntry<K,V> first = tab[index];        // 遍历链表，更新或插入        for (HashEntry<K,V> e = first; e != null; e = e.next) {            if (e.hash == hash && key.equals(e.key)) {                V oldValue = e.value;                e.value = value;                return oldValue;            }        }        // 插入新节点（头插法）        tab[index] = new HashEntry<>(key, hash, first, value);        count++;        return null;    } finally {        unlock(); // 释放锁    }}

优点：├─► 并发度高：默认16个Segment，最多支持16个线程同时写入├─► 读操作基本无锁（HashEntry的value和next是volatile）缺点：├─► 内存开销大：两级数据结构，每个Segment都是一个独立的锁├─► 并发级别固定：创建后不能改变Segment数量├─► 扩容复杂：每个Segment独立扩容└─► 某些操作需要锁所有Segment（如size()、containsValue()）

ConcurrentHashMap (JDK 8)│├─► Node<K,V>[] table   ← 数据数组（和HashMap一样）│    ││    ├─► 桶0: null│    ││    ├─► 桶1: Node → Node → Node（链表，synchronized锁头节点）│    ││    ├─► 桶2: TreeNode ← 红黑树（synchronized锁根节点）│    ││    └─► 桶3: ForwardingNode ← 扩容时的特殊节点│├─► sizeCtl   ← 控制标识（非常重要）│    ├─► -1: 正在初始化│    ├─► -N: 有N-1个线程正在扩容│    ├─► 正数: 下一次扩容的阈值│    └─► 0: 默认值，未初始化│└─► 其他辅助结构：CounterCell[]（用于高并发计数）

// JDK 8 ConcurrentHashMap 的核心字段public class ConcurrentHashMap<K,V> {    // 存储数据的数组    transient volatile Node<K,V>[] table;    // 扩容时的新数组（正常时为null）    private transient volatile Node<K,V>[] nextTable;    // 控制标识（多用途）    private transient volatile int sizeCtl;    // 计数辅助数组（用于高并发下的计数）    private transient volatile CounterCell[] counterCells;    // 基础计数器    private transient volatile long baseCount;    // 链表转红黑树的阈值（和HashMap一样）    static final int TREEIFY_THRESHOLD = 8;    // 红黑树转链表的阈值    static final int UNTREEIFY_THRESHOLD = 6;    // 转红黑树的最小数组长度    static final int MIN_TREEIFY_CAPACITY = 64;}

put(key, value)    │    ▼计算hash值（扰动处理）    │    ▼进入死循环（保证一定能插入成功）    │    ├───── case 1: table为空 ────► 初始化数组（CAS设置sizeCtl）    │    ├───── case 2: 目标桶为空 ────► CAS直接插入新节点（无锁）    │    ├───── case 3: 正在扩容 ────► 当前线程帮忙扩容（协同）    │    └───── case 4: 桶不为空 ────► synchronized锁住桶的头节点                                     │                                     ▼                              ├─► 如果是链表：遍历插入/更新                              │                              └─► 如果是红黑树：调用树的插入方法    │    ▼    插入完成后，检查链表长度是否达到8        如果达到8且数组长度<64，触发扩容        如果达到8且数组长度≥64，转红黑树

// JDK 8 putVal 方法核心逻辑final V putVal(K key, V value, boolean onlyIfAbsent) {    if (key == null || value == null) throw new NullPointerException();    // 1. 计算hash值（扰动处理）    int hash = spread(key.hashCode());    int binCount = 0;    // 2. 死循环，直到操作成功    for (Node<K,V>[] tab = table;;) {        Node<K,V> f; int n, i, fh;        // 3. 如果table为空，初始化        if (tab == null || (n = tab.length) == 0)            tab = initTable();        // 4. 如果目标桶为空，CAS插入        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {            if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null)))                break;                   // 插入成功，退出循环        }        // 5. 如果正在扩容，帮忙扩容        else if ((fh = f.hash) == MOVED)            tab = helpTransfer(tab, f);        // 6. 桶不为空，需要加锁操作        else {            V oldVal = null;            // 锁住桶的头节点（粒度极细）            synchronized (f) {                // 再次检查头节点是否被修改（double-check）                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<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;                        }                    }                }            }            // 7. 插入完成后，检查是否需要转红黑树            if (binCount != 0) {                if (binCount >= TREEIFY_THRESHOLD)                    treeifyBin(tab, i);                if (oldVal != null)                    return oldVal;                break;            }        }    }    // 8. 计数增加    addCount(1L, binCount);    return null;}

get(key) 全程无锁！    │    ▼计算hash值    │    ▼找到对应的桶    │    ├───── 桶为空 ────► 返回null    │    ├───── 桶的第一个节点就是要找的 ────► 直接返回value    │    ├───── 正在扩容（节点hash=MOVED） ────► 去新数组找    │    └───── 链表/红黑树 ────► 遍历查找

// get方法核心逻辑（无锁）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;        }        // 正在扩容，去nextTable找        else if (eh < 0)            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;}

触发扩容的条件：1. 链表长度达到8，但数组长度小于642. 元素个数超过阈值（sizeCtl）扩容过程：    │    ▼第一个扩容的线程创建nextTable（新数组）    │    ▼将sizeCtl设置为负数（表示正在扩容）    │    ▼从后往前将每个桶迁移到nextTable    │    ├───── 迁移完一个桶，在原位置放ForwardingNode（hash=MOVED）    │    └───── 其他线程发现是MOVED，会帮忙迁移其他桶    │    ▼所有桶迁移完成后，table指向nextTable    │    ▼恢复sizeCtl为正数（新阈值）

// 扩容时的桶迁移（多线程协同）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;    // 第一个扩容线程初始化nextTable    if (nextTab == null) {        try {            @SuppressWarnings("unchecked")            Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];            nextTab = nt;        } catch (Throwable ex) {            sizeCtl = Integer.MAX_VALUE;            return;        }        nextTable = nextTab;        transferIndex = n;    }    // 每个线程领取自己的任务区间，从后往前迁移    // 迁移完一个桶，放ForwardingNode    // 其他线程看到ForwardingNode，就去帮忙迁移其他桶}

publicintsize() {    long n = sumCount();  // 累加baseCount和CounterCell    return ((n < 0L) ? 0 : (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE : (int)n);}final longsumCount() {    CounterCell[] as = counterCells; CounterCell a;    long sum = baseCount;    if (as != null) {        for (int i = 0; i < as.length; ++i) {            if ((a = as[i]) != null)                sum += a.value;        }    }    return sum;}