ConcurrentHashMap源码解析：并发安全的哈希表实现

一、ConcurrentHashMap概述

在Java开发中，我们经常需要使用哈希表来存储键值对数据。JDK提供了多种Map实现，其中HashMap是最常用的，但它在多线程环境下是线程不安全的；Hashtable虽然线程安全，但采用全表锁导致并发效率极低。

**ConcurrentHashMap**应运而生，它是java.util.concurrent包下的并发哈希表实现，专为高并发场景设计。其核心价值在于：在保证线程安全的前提下，通过精妙的设计实现了高效的并发访问。

从JDK 1.5首次引入到现在，ConcurrentHashMap的实现经历了多次重大优化：

• JDK 1.7及之前：采用"分段锁(Segment)"机制
• JDK 1.8及之后：改用"CAS + synchronized"的锁机制，性能大幅提升
• JDK 9及以上：进一步优化了并发控制和内存效率

本文将基于JDK 1.8的实现进行解析，因为这是目前应用最广泛的版本。

二、ConcurrentHashMap核心特性及源码证明

1. 细粒度锁机制实现线程安全

特性描述：
ConcurrentHashMap通过对哈希表中的每个"桶(Bucket)"进行独立加锁，而非锁定整个哈希表，实现了细粒度的并发控制。这意味着多个线程可以同时操作不同的桶，大幅提升了并发效率。

核心概念解释：

• 哈希桶(Bucket)：哈希表的基本存储单元，每个桶本质上是一个链表或红黑树的头节点
• CAS(Compare-And-Swap)：一种无锁原子操作，通过比较内存值与预期值，若相等则更新为新值
• synchronized：Java中的同步关键字，此处用于对桶的头节点加锁

源码证明：

final V putVal(K key, V value, boolean onlyIfAbsent) {
    // ConcurrentHashMap不允许key或value为null
    if (key == null || value == null) throw new NullPointerException();
    // 计算哈希值，通过spread方法减少哈希冲突
    int hash = spread(key.hashCode());
    int binCount = 0;
    // 循环直到插入成功（CAS操作可能失败需要重试）
    for (Node<K,V>[] tab = table;;) {
        Node<K,V> f; int n, i, fh;
        // 如果哈希表未初始化，则初始化
        if (tab == null || (n = tab.length) == 0)
            tab = initTable();
        // 如果目标桶为空，尝试用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;  // 插入成功，退出循环
        }
        // 如果遇到ForwardingNode，说明正在扩容，帮助扩容
        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;
                            // 找到相同key的节点
                            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;
                            }
                        }
                    }
                    // 红黑树节点（哈希值为负数且不是MOVED）
                    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;
}

代码解析：

1. 当桶为空时，使用casTabAt进行CAS操作，无锁插入新节点
2. 当桶不为空时，仅对桶的头节点f进行synchronized加锁
3. 不同桶的操作可以并行执行，只有同一桶的操作才会互斥
4. 这种设计将锁的粒度从整个哈希表缩小到单个桶，理论上并发度等于桶的数量

2. 无锁读操作与volatile保证

特性描述：
ConcurrentHashMap的读操作全程无需加锁，通过volatile关键字保证内存可见性，实现了高效的读写并发。这是它相比Hashtable性能优异的重要原因。

核心概念解释：

• 内存可见性：当一个线程修改了共享变量的值，其他线程能够立即看到修改后的结果
• volatile：Java关键字，用于保证变量的内存可见性，禁止指令重排序
• Unsafe：JDK内部的工具类，提供了硬件级别的原子操作

源码证明：

// Node节点的定义
static class Node<K,V> implements Map.Entry<K,V> {
    final int hash;
    final K key;
    // val和next都用volatile修饰，保证可见性
    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;
    }
    // 省略其他方法...
}

// 获取元素的方法
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;
        }
        // 哈希值为负数，可能是红黑树或ForwardingNode
        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;
}

// 安全地获取桶中的节点（使用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);
}

代码解析：

1. Node类的val和next字段都用volatile修饰，确保修改后能立即被其他线程看到
2. tabAt方法通过Unsafe.getObjectVolatile获取节点，保证了读取到的是最新值
3. 整个get方法没有任何锁操作，实现了无锁读
4. 读操作与写操作可以并行进行，读操作之间也可以完全并行

这种设计的优势在于：读操作是ConcurrentHashMap中最频繁的操作之一，无锁化极大提升了整体性能。

3. 支持多线程并发扩容

特性描述：
当哈希表中的元素数量达到阈值时，需要进行扩容（resize）操作。ConcurrentHashMap支持多个线程共同参与扩容，避免了单线程扩容的性能瓶颈，这是其高并发设计的重要体现。

核心概念解释：

• 扩容(resize)：将哈希表的容量翻倍，重新计算所有元素的哈希值并迁移到新的位置
• ForwardingNode：特殊的节点类型，用于标记正在扩容的桶，引导读写操作到新表
• transferIndex：用于协调多线程扩容的索引变量，标记尚未迁移的桶范围

源码证明：

// 扩容的核心方法
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
    int n = tab.length, stride;
    // 根据CPU核心数计算每个线程处理的桶数量
    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) { // 处理OOM
            sizeCtl = Integer.MAX_VALUE;
            return;
        }
        nextTable = nextTab;
        transferIndex = n; // 迁移索引初始化为原表长度
    }
    int nextn = nextTab.length;
    // 创建ForwardingNode，用于标记已迁移的桶
    ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
    boolean advance = true;
    boolean finishing = false; // 标记是否完成扩容
    
    // 多线程分工处理不同的桶
    for (int i = 0, bound = 0;;) {
        Node<K,V> f; int fh;
        // 分配当前线程需要处理的桶范围
        while (advance) {
            int nextIndex, nextBound;
            if (--i >= bound || finishing)
                advance = false;
            else if ((nextIndex = transferIndex) <= 0) {
                i = -1;
                advance = false;
            }
            // CAS更新transferIndex，获取待处理的桶范围
            else if (U.compareAndSwapInt
                     (this, TRANSFERINDEX, nextIndex,
                      nextBound = (nextIndex > stride ?
                                   nextIndex - stride : 0))) {
                bound = nextBound;
                i = nextIndex - 1;
                advance = false;
            }
        }
        if (i < 0 || i >= n || i + n >= nextn) {
            int sc;
            // 所有桶处理完毕，更新table为新表
            if (finishing) {
                nextTable = null;
                table = nextTab;
                sizeCtl = (n << 1) - (n >>> 1); // 更新阈值
                return;
            }
            // 减少扩容线程计数
            if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
                if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
                    return;
                finishing = advance = true;
                i = n; // 重新检查所有桶
            }
        }
        // 处理空桶
        else if ((f = tabAt(tab, i)) == null)
            advance = casTabAt(tab, i, null, fwd);
        // 遇到ForwardingNode，说明该桶已迁移
        else if ((fh = f.hash) == MOVED)
            advance = true; // 继续处理下一个桶
        else {
            // 对当前桶加锁，进行迁移
            synchronized (f) {
                if (tabAt(tab, i) == f) {
                    Node<K,V> ln, hn;
                    // 处理链表节点
                    if (fh >= 0) {
                        int runBit = fh & n;
                        Node<K,V> lastRun = f;
                        // 分离链表为高低两部分
                        for (Node<K,V> p = f.next; p != null; p = p.next) {
                            int b = p.hash & n;
                            if (b != runBit) {
                                runBit = b;
                                lastRun = p;
                            }
                        }
                        if (runBit == 0) {
                            ln = lastRun;
                            hn = null;
                        } else {
                            hn = lastRun;
                            ln = null;
                        }
                        // 重新构建链表
                        for (Node<K,V> p = f; p != lastRun; p = p.next) {
                            int ph = p.hash; K pk = p.key; V pv = p.val;
                            if ((ph & n) == 0)
                                ln = new Node<K,V>(ph, pk, pv, ln);
                            else
                                hn = new Node<K,V>(ph, pk, pv, hn);
                        }
                        // 放入新表的对应位置
                        setTabAt(nextTab, i, ln);
                        setTabAt(nextTab, i + n, hn);
                        // 标记当前桶已迁移
                        setTabAt(tab, i, fwd);
                        advance = true;
                    }
                    // 处理红黑树节点
                    else if (f instanceof TreeBin) {
                        TreeBin<K,V> t = (TreeBin<K,V>)f;
                        TreeNode<K,V> lo = null, loTail = null;
                        TreeNode<K,V> hi = null, hiTail = null;
                        int lc = 0, hc = 0;
                        // 分离红黑树为高低两部分
                        for (Node<K,V> e = t.first; e != null; e = e.next) {
                            int h = e.hash;
                            TreeNode<K,V> p = (TreeNode<K,V>)e;
                            if ((h & n) == 0) {
                                if ((p.prev = loTail) == null)
                                    lo = p;
                                else
                                    loTail.next = p;
                                loTail = p;
                                lc++;
                            } else {
                                if ((p.prev = hiTail) == null)
                                    hi = p;
                                else
                                    hiTail.next = p;
                                hiTail = p;
                                hc++;
                            }
                        }
                        // 根据节点数量决定是转为链表还是保留红黑树
                        ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
                            (hc != 0) ? new TreeBin<K,V>(lo) : t;
                        hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
                            (lc != 0) ? new TreeBin<K,V>(hi) : t;
                        // 放入新表的对应位置
                        setTabAt(nextTab, i, ln);
                        setTabAt(nextTab, i + n, hn);
                        // 标记当前桶已迁移
                        setTabAt(tab, i, fwd);
                        advance = true;
                    }
                }
            }
        }
    }
}

代码解析：

1. 扩容时将原表分为多个段，每个线程通过CAS获取一段连续的桶进行迁移
2. 迁移完成的桶会被标记为ForwardingNode，引导新的读写操作访问新表
3. 新表容量是原表的2倍，元素根据哈希值重新分配到新表的两个可能位置
4. 当有新线程访问正在扩容的哈希表时，会自动参与到扩容过程中（helpTransfer方法）

这种多线程协同扩容的机制，避免了单线程扩容在大数据量下的性能瓶颈，是ConcurrentHashMap高并发能力的重要保障。

4. 链表与红黑树的自适应转换

特性描述：
ConcurrentHashMap会根据桶中元素的数量，自动在链表和红黑树之间进行转换。当元素较少时使用链表（插入快），当元素较多时转为红黑树（查询快），兼顾了插入和查询性能。

核心概念解释：

• 链表：一种线性数据结构，插入快但查询慢（时间复杂度O(n)）
• 红黑树：一种自平衡二叉查找树，查询快（时间复杂度O(log n)）但插入稍慢
• 树化阈值：当桶中元素数量超过该阈值（默认8）时，将链表转为红黑树
• 反树化阈值：当桶中元素数量低于该阈值（默认6）时，将红黑树转回链表

源码证明：

// 树化阈值：当桶中元素数量达到8时，考虑转为红黑树
static final int TREEIFY_THRESHOLD = 8;

// 反树化阈值：当桶中元素数量低于6时，转回链表
static final int UNTREEIFY_THRESHOLD = 6;

// 最小树化容量：哈希表容量至少达到64才会进行树化，否则会先扩容
static final int MIN_TREEIFY_CAPACITY = 64;

// 将链表转为红黑树的方法
private final void treeifyBin(Node<K,V>[] tab, int index) {
    Node<K,V>[] t; int n, sc;
    if (tab != null) {
        // 如果哈希表容量小于MIN_TREEIFY_CAPACITY，先进行扩容而非树化
        if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
            tryPresize(n << 1);
        else if ((t = tab) != null && (sc = t.length) > 0) {
            // 锁定当前桶
            synchronized (t[index]) {
                if (t[index] == tabAt(t, index)) {
                    Node<K,V> hd = null, tl = null;
                    // 将链表节点转为树节点
                    for (Node<K,V> e = tabAt(t, index); e != null; e = e.next) {
                        TreeNode<K,V> p = new TreeNode<K,V>(e.hash, e.key, e.val, null, null);
                        if ((p.prev = tl) == null)
                            hd = p;
                        else
                            tl.next = p;
                        tl = p;
                    }
                    // 将桶的头节点设置为TreeBin（红黑树的包装类）
                    setTabAt(t, index, new TreeBin<K,V>(hd));
                }
            }
        }
    }
}

// TreeBin类中的查找方法（红黑树查找）
final Node<K,V> find(int h, Object k) {
    if (k == null)
        return null;
    for (Node<K,V> e = first; e != null;) {
        int s; K ek;
        if (((s = lockState) & (WAITER | WRITER)) != 0) {
            if (e.hash == h &&
                ((ek = e.key) == k || (ek != null && k.equals(ek))))
                return e;
            e = e.next;
        }
        else if (U.compareAndSwapInt(this, LOCKSTATE, s,
                                     s | READER)) {
            TreeNode<K,V> p;
            try {
                p = ((TreeNode<K,V>)e).findTreeNode(h, k, null);
            } finally {
                Thread w;
                if (U.getAndAddInt(this, LOCKSTATE, -READER) == (READER | WRITER) &&
                    (w = waiter) != null)
                    LockSupport.unpark(w);
            }
            return p;
        }
    }
    return null;
}

代码解析：

1. 当桶中元素数量达到TREEIFY_THRESHOLD（8）时，会触发treeifyBin方法
2. treeifyBin方法会先检查哈希表容量，如果小于MIN_TREEIFY_CAPACITY（64），会先进行扩容而非树化
3. 树化过程会将链表节点转为红黑树节点，并通过TreeBin类来管理红黑树
4. 红黑树的查找操作通过findTreeNode方法实现，时间复杂度为O(log n)
5. 当红黑树中元素数量减少到UNTREEIFY_THRESHOLD（6）以下时，会转回链表

这种自适应的数据结构转换，使得ConcurrentHashMap在不同数据量下都能保持较好的性能特性。

5. 弱一致性迭代器

特性描述：
ConcurrentHashMap的迭代器具有弱一致性（Weakly Consistent），它不会抛出ConcurrentModificationException，允许在迭代过程中修改映射，但可能不会反映所有的并发修改。

核心概念解释：

• 快速失败(Fail-Fast)：如HashMap的迭代器，当发现并发修改时会立即抛出异常
• 弱一致性(Weakly Consistent)：迭代器可以看到迭代开始前的所有修改，可能看到部分迭代过程中的修改，但不会抛出异常

源码证明：

// 迭代器的实现
private abstract class HashIterator {
    Node<K,V> next;        // 下一个要返回的节点
    Node<K,V> current;     // 当前节点
    int expectedModCount;  // 用于快速失败的修改计数（但实际未使用）
    int index;             // 当前桶索引
    Node<K,V>[] t;         // 当前哈希表
    
    HashIterator() {
        expectedModCount = modCount;
        Node<K,V>[] tab = table;
        current = next = null;
        index = 0;
        t = tab;
    }
    
    public final boolean hasNext() {
        // 检查是否还有下一个节点
        return next != null;
    }
    
    final Node<K,V> nextNode() {
        Node<K,V>[] t;
        Node<K,V> e = next;
        // 虽然检查了modCount，但实际中不会抛出异常，因为迭代器是弱一致性的
        if (modCount != expectedModCount)
            throw new ConcurrentModificationException();
        if (e == null)
            throw new NoSuchElementException();
        // 如果当前节点的next为null，寻找下一个非空桶
        if ((next = (current = e).next) == null && (t = table) != null) {
            do {} while (index < t.length && (next = tabAt(t, index++)) == null);
        }
        return e;
    }
    
    public final void remove() {
        Node<K,V> p = current;
        if (p == null)
            throw new IllegalStateException();
        if (modCount != expectedModCount)
            throw new ConcurrentModificationException();
        current = null;
        K key = p.key;
        // 调用ConcurrentHashMap的remove方法
        removeNode(hash(key), key, null, false, false);
        expectedModCount = modCount;
    }
}

代码解析：

1. 迭代器基于创建时的哈希表快照进行遍历
2. 迭代过程中允许对映射进行修改，但迭代器可能看不到这些修改
3. 不会像HashMap那样严格检查并发修改并抛出异常
4. 迭代器的remove方法是安全的，通过调用ConcurrentHashMap的removeNode实现

弱一致性迭代器的设计，平衡了迭代性能和并发安全性，适合不需要强一致性的场景。

三、ConcurrentHashMap适用场景

1. 高并发读写场景：

   • 缓存系统：如实现本地缓存，支持多线程并发读写
   • 计数器：多线程环境下统计各种指标的计数
   • 会话存储：Web应用中存储用户会话信息

2. 需要避免集合迭代异常的场景：

   • 日志收集：多线程写入日志，同时有线程读取日志
   • 实时数据展示：后台线程更新数据，UI线程读取并展示数据

3. 需要高效并发处理的场景：

   • 分布式锁实现：利用putIfAbsent方法实现简单的分布式锁
   • 任务调度：存储和管理多线程环境下的任务信息

4. 替代Hashtable和同步包装的场景：

   • 任何使用Collections.synchronizedMap(new HashMap<>())的地方，都可以用ConcurrentHashMap替代以获得更好的并发性能

四、使用ConcurrentHashMap的注意事项

1. 复合操作的原子性问题：
   单个方法（如put、get）是线程安全的，但多个方法组合的复合操作则不是原子的：

   // 非原子操作，存在线程安全问题
   if (map.containsKey(key)) {
       map.put(key, value);
   }
   // 应使用原子方法
   map.putIfAbsent(key, value);
   

   ConcurrentHashMap提供了一些原子方法：putIfAbsent、remove、replace等，应优先使用这些方法。

2. key和value不能为null：
   与HashMap不同，ConcurrentHashMap不允许key或value为null，否则会抛出NullPointerException。这是因为在并发环境下，null值无法区分是"键不存在"还是"值为null"。

3. 迭代器的弱一致性限制：
   迭代器可能不会反映迭代过程中的所有修改，如需看到所有修改，需要重新获取迭代器。同时，迭代器不支持add、set等修改操作，只能调用remove。

4. size()方法的局限性：
   size()方法返回的是一个估计值，而非精确值。这是因为在并发环境下，获取精确值需要全表锁定，影响性能。如果需要精确计数，应使用mappingCount()方法，它返回一个long类型的估计值，精度更高。

5. 内存消耗较大：
   为了支持并发，ConcurrentHashMap比HashMap消耗更多内存。在内存受限的场景下，需要权衡内存和并发性能。

6. 不适合强一致性需求：
   ConcurrentHashMap提供的是最终一致性，而非强一致性。如果应用场景要求强一致性（如金融交易），可能需要额外的同步机制。

五、总结

ConcurrentHashMap是Java并发编程中的杰作，它通过精妙的设计平衡了线程安全和并发性能，其核心特性与实现逻辑可总结如下：

1. 细粒度锁机制：采用"CAS + synchronized"组合，对单个哈希桶（Bucket）加锁而非全表锁，使不同桶的操作可并行执行，大幅提升并发效率。

2. 无锁化读操作：借助volatile关键字和Unsafe工具类，读操作全程无锁，通过内存可见性保证读取最新值，实现高效的读写并发。

3. 多线程协同扩容：支持多个线程共同参与扩容（resize），通过transferIndex分配任务范围，用ForwardingNode标记已迁移桶，避免单线程扩容瓶颈。

4. 自适应数据结构：根据桶中元素数量，自动在链表（插入快）和红黑树（查询快）间转换，通过树化阈值（8）和反树化阈值（6）实现动态调整，兼顾不同数据量下的性能。

5. 弱一致性迭代器：迭代器基于创建时的快照遍历，允许并发修改且不抛出ConcurrentModificationException，平衡了迭代性能与安全性。

在适用场景上，ConcurrentHashMap是高并发读写场景（如缓存、计数器）、需避免迭代异常场景的首选，可替代Hashtable和同步包装的HashMap以提升性能。但使用时需注意：复合操作需用原子方法、key/value不可为null、迭代器弱一致性、size()为估计值等局限性。

理解ConcurrentHashMap的设计思想——以细粒度锁、无锁编程、多线程协同等机制实现"安全与性能并存"，不仅能更好地应用它，更能为解决其他并发问题提供宝贵的思路。

---

Studying will never be ending.

▲如有纰漏，烦请指正~~