ConcurrentHashMap详解：原理、实现与并发控制
ConcurrentHashMap是Java并发包(java.util.concurrent)中提供的线程安全哈希表实现，它是对HashMap的线程安全版本，但比Hashtable或Collections.synchronizedMap()有更好的并发性能。本文将全面解析ConcurrentHashMap的设计原理、实现机制和最佳实践。
一、核心特性与设计演进
1. 基本特性
线程安全：支持多线程并发访问而不需要外部同步
高并发：通过分段锁或CAS操作实现高并发性能
弱一致性：迭代器反映的是创建时的状态，不保证反映所有更新
不允许null键/值：与HashMap不同，不允许null键或值
2. 版本演进
Java 7实现(分段锁)：
将整个哈希表分成多个段(Segment)，每个段是一个独立的哈希表
每个段有自己的锁，不同段可以并发操作
默认创建16个段，并发级别为16
Java 8实现(CAS + synchronized)：
取消了分段锁设计
使用CAS(Compare-And-Swap) + synchronized实现更细粒度的锁
当发生哈希冲突时，只锁住冲突的链表或红黑树的头节点
引入红黑树处理哈希冲突，当链表长度超过阈值(默认为8)时转换为红黑树
二、核心数据结构与参数
1. 数据结构
ConcurrentHashMap在JDK8中的数据结构为数组+链表+红黑树：
默认数组大小：16，负载因子：0.75
链表长度 ≥ 8时转换为红黑树(树化)
树节点数 ≤ 6时退化为链表(反树化)
2. 重要参数
initialCapacity：初始容量，默认16
loadFactor：负载因子，默认0.75f
concurrencyLevel：并发级别(Java7)，Java8中仅用于兼容性
TREEIFY_THRESHOLD：链表转红黑树的阈值，默认为8
UNTREEIFY_THRESHOLD：红黑树转链表的阈值，默认为6
三、并发控制机制
1. Java 8中的并发控制
CAS：用于无竞争情况下的快速操作
synchronized：当发生哈希冲突时，锁住链表或树的头节点
volatile：保证变量的可见性
sizeCtl：控制表初始化和扩容的特殊标志
2. 为什么使用synchronized而不是ReentrantLock
性能考量：Java 8优化后的synchronized在单bucket竞争时性能更好
内存占用：synchronized的JVM内置锁更节省内存
JIT优化：锁消除/锁膨胀等优化更成熟
四、关键操作实现原理
1. put操作实现
put操作的核心逻辑如下：
计算key的hash值
如果表未初始化，则初始化表
如果定位到的桶为空，尝试CAS插入新节点
如果桶正在迁移(ForwardingNode)，帮助迁移
否则对桶的头节点加synchronized锁：
如果是链表，遍历查找并插入/更新
如果是树，按照红黑树方式插入
判断是否需要树化或扩容
源码为：
/*** Maps the specified key to the specified value in this table.* Neither the key nor the value can be null.** <p> The value can be retrieved by calling the <tt>get</tt> method* with a key that is equal to the original key.** @param key key with which the specified value is to be associated* @param value value to be associated with the specified key* @return the previous value associated with <tt>key</tt>, or* <tt>null</tt> if there was no mapping for <tt>key</tt>* @throws NullPointerException if the specified key or value is null*/
public V put(K key, V value) {Segment<K,V> s;if (value == null)throw new NullPointerException();int hash = hash(key);// hash 值无符号右移 28位（初始化时获得），然后与 segmentMask=15 做与运算// 其实也就是把高4位与segmentMask（1111）做与运算int j = (hash >>> segmentShift) & segmentMask;if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck(segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment// 如果查找到的 Segment 为空，初始化s = ensureSegment(j);return s.put(key, hash, value, false);
}/*** Returns the segment for the given index, creating it and* recording in segment table (via CAS) if not already present.** @param k the index* @return the segment*/
@SuppressWarnings("unchecked")
private Segment<K,V> ensureSegment(int k) {final Segment<K,V>[] ss = this.segments;long u = (k << SSHIFT) + SBASE; // raw offsetSegment<K,V> seg;// 判断 u 位置的 Segment 是否为nullif ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {Segment<K,V> proto = ss[0]; // use segment 0 as prototype// 获取0号 segment 里的 HashEntry<K,V> 初始化长度int cap = proto.table.length;// 获取0号 segment 里的 hash 表里的扩容负载因子，所有的 segment 的 loadFactor 是相同的float lf = proto.loadFactor;// 计算扩容阀值int threshold = (int)(cap * lf);// 创建一个 cap 容量的 HashEntry 数组HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) { // recheck// 再次检查 u 位置的 Segment 是否为null，因为这时可能有其他线程进行了操作Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);// 自旋检查 u 位置的 Segment 是否为nullwhile ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))== null) {// 使用CAS 赋值，只会成功一次if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))break;}}}return seg;
}上面探究了获取 Segment 段和初始化 Segment 段的操作。最后一行的 Segment 的 put 方法还没有查看，继续分析:
final V put(K key, int hash, V value, boolean onlyIfAbsent) {// 获取 ReentrantLock 独占锁，获取不到，scanAndLockForPut 获取。HashEntry<K,V> node = tryLock() ? null : scanAndLockForPut(key, hash, value);V oldValue;try {HashEntry<K,V>[] tab = table;// 计算要put的数据位置int index = (tab.length - 1) & hash;// CAS 获取 index 坐标的值HashEntry<K,V> first = entryAt(tab, index);for (HashEntry<K,V> e = first;;) {if (e != null) {// 检查是否 key 已经存在，如果存在，则遍历链表寻找位置，找到后替换 valueK k;if ((k = e.key) == key ||(e.hash == hash && key.equals(k))) {oldValue = e.value;if (!onlyIfAbsent) {e.value = value;++modCount;}break;}e = e.next;}else {// first 有值没说明 index 位置已经有值了，有冲突，链表头插法。if (node != null)node.setNext(first);elsenode = new HashEntry<K,V>(hash, key, value, first);int c = count + 1;// 容量大于扩容阀值，小于最大容量，进行扩容if (c > threshold && tab.length < MAXIMUM_CAPACITY)rehash(node);else// index 位置赋值 node，node 可能是一个元素，也可能是一个链表的表头setEntryAt(tab, index, node);++modCount;count = c;oldValue = null;break;}}} finally {unlock();}return oldValue;
}2. get操作实现
get操作的特点：
完全无锁，依赖volatile保证可见性
通过两次hash定位到具体元素
如果在桶上直接返回
如果是红黑树则按树方式查找
否则按链表遍历查找
源码为：
public V get(Object key) {Segment<K,V> s; // manually integrate access methods to reduce overheadHashEntry<K,V>[] tab;int h = hash(key);long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;// 计算得到 key 的存放位置if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&(tab = s.table) != null) {for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile(tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);e != null; e = e.next) {// 如果是链表，遍历查找到相同 key 的 value。K k;if ((k = e.key) == key || (e.hash == h && key.equals(k)))return e.value;}}return null;
}3. 扩容机制
ConcurrentHashMap的扩容机制非常精巧：
当元素数量超过容量*负载因子时触发扩容
扩容时创建新表(大小为原表2倍)
采用多线程协助扩容机制，提高扩容效率
扩容期间可以同时进行查询操作
通过ForwardingNode标记已迁移的桶
源码为：
private void rehash(HashEntry<K,V> node) {HashEntry<K,V>[] oldTable = table;// 老容量int oldCapacity = oldTable.length;// 新容量，扩大两倍int newCapacity = oldCapacity << 1;// 新的扩容阀值threshold = (int)(newCapacity * loadFactor);// 创建新的数组HashEntry<K,V>[] newTable = (HashEntry<K,V>[]) new HashEntry[newCapacity];// 新的掩码，默认2扩容后是4，-1是3，二进制就是11。int sizeMask = newCapacity - 1;for (int i = 0; i < oldCapacity ; i++) {// 遍历老数组HashEntry<K,V> e = oldTable[i];if (e != null) {HashEntry<K,V> next = e.next;// 计算新的位置，新的位置只可能是不变或者是老的位置+老的容量。int idx = e.hash & sizeMask;if (next == null) // Single node on list// 如果当前位置还不是链表，只是一个元素，直接赋值newTable[idx] = e;else { // Reuse consecutive sequence at same slot// 如果是链表了HashEntry<K,V> lastRun = e;int lastIdx = idx;// 新的位置只可能是不变或者是老的位置+老的容量。// 遍历结束后，lastRun 后面的元素位置都是相同的for (HashEntry<K,V> last = next; last != null; last = last.next) {int k = last.hash & sizeMask;if (k != lastIdx) {lastIdx = k;lastRun = last;}}// ，lastRun 后面的元素位置都是相同的，直接作为链表赋值到新位置。newTable[lastIdx] = lastRun;// Clone remaining nodesfor (HashEntry<K,V> p = e; p != lastRun; p = p.next) {// 遍历剩余元素，头插法到指定 k 位置。V v = p.value;int h = p.hash;int k = h & sizeMask;HashEntry<K,V> n = newTable[k];newTable[k] = new HashEntry<K,V>(h, p.key, v, n);}}}}// 头插法插入新的节点int nodeIndex = node.hash & sizeMask; // add the new nodenode.setNext(newTable[nodeIndex]);newTable[nodeIndex] = node;table = newTable;
}五、与其他Map的比较
1. 与HashMap、Hashtable的比较
| 特性 | HashMap | Hashtable | ConcurrentHashMap |
|---|---|---|---|
| 线程安全 | 否 | 是 | 是 |
| 锁粒度 | 无锁 | 全表锁 | 分段锁/桶锁 |
| 允许null键/值 | 是 | 否 | 否 |
| 迭代器弱一致性 | 否 | 否 | 是 |
| 性能 | 最高 | 最低 | 高 |
2. 线程安全保证的差异
HashMap：在多线程环境下可能出现数据丢失或环形链表(JDK7)
Hashtable：全表锁导致性能低下
Collections.synchronizedMap：方法级别同步，性能较差
ConcurrentHashMap：细粒度锁，高并发性能
六、使用场景与注意事项
1. 适用场景
高并发环境下的键值存储
需要线程安全且高性能的Map实现
读多写少的场景(因为读操作完全无锁)
2. 注意事项
复合操作问题 ：
ConcurrentHashMap只能保证单个操作的原子性
复合操作(如"检查然后执行")需要额外同步
// 不安全的复合操作
if (!map.containsKey(key)) {map.put(key, value);
}// 安全的替代方案
map.putIfAbsent(key, value);size()的弱一致性：
size()方法返回的是近似值，不是精确值
精确计数需要全局锁，采用分片计数(baseCount + CounterCell[])牺牲精确性换取并发度
迭代器的弱一致性 ：
迭代器反映的是创建时的状态
不会抛出ConcurrentModificationException
七、源码解析与设计亮点
1. 核心数据结构
// 主哈希表，volatile保证可见性
transient volatile Node<K,V>[] table;// 扩容时用，代表扩容后的数组
private transient volatile Node<K,V>[] nextTable;// 控制表初始化和扩容的字段
private transient volatile int sizeCtl;2. 节点类型
基础节点(链表)：
static class Node<K,V> implements Map.Entry<K,V> {final int hash; // 不可变hash值final K key; // 不可变keyvolatile V val; // volatile保证可见性volatile Node<K,V> next; // volatile保证可见性
}树节点(红黑树)：
static final class TreeNode<K,V> extends Node<K,V> {TreeNode<K,V> parent; // 红黑树父节点TreeNode<K,V> left;TreeNode<K,V> right;TreeNode<K,V> prev; // 链表前驱(用于删除)boolean red;
}3. 设计亮点
锁粒度优化：仅锁单个bucket头节点(Java 7锁Segment)
无锁读路径：所有读操作都不加锁，依赖volatile和final保证可见性
协作式扩容：多线程共同完成数据迁移
状态机设计：通过MOVED、TREEBIN等特殊hash值实现状态转换
缓存友好：使用@Contended避免伪共享
八、常见问题解答
1. 为什么ConcurrentHashMap不允许null值？
ConcurrentHashMap不允许null值主要是因为并发环境下难以区分"键不存在"和"键对应的值为null"这两种情况。在非并发Map中，可以通过containsKey()来解决，但在并发环境下，这两个操作的原子性无法保证。
2. ConcurrentHashMap如何保证扩容期间的一致性？
ForwardingNode机制：已迁移的bucket会被标记，读操作自动跳转新表
写操作协作：遇到ForwardingNode的写入线程会先协助扩容
3. ConcurrentHashMap的size()为何不精确？
精确计数需要全局锁，这会严重影响并发性能。ConcurrentHashMap采用分片计数(baseCount + CounterCell[])的方式，牺牲精确性换取并发度。
九、最佳实践与性能优化
合理设置初始容量：避免频繁扩容
选择合适的并发级别：在Java 8中主要影响初始容量
利用原子性方法：如putIfAbsent、compute等
避免不必要的复合操作：使用内置的原子方法替代
读多写少场景性能最佳：因为读操作完全无锁
总结
ConcurrentHashMap是Java并发编程中最重要的并发容器之一，它通过精细的锁设计、CAS操作和volatile变量，在保证线程安全的同时提供了接近HashMap的性能。从Java 7的分段锁到Java 8的CAS+synchronized优化，ConcurrentHashMap的设计演进体现了Java并发编程技术的不断进步。理解其实现原理和适用场景，对于开发高性能并发应用至关重要。