Java并发编程——ConcurrentHashMap源码深度剖析与实战应用

1. 为什么需要ConcurrentHashMap？

在日常开发中，我们经常会遇到需要统计用户访问次数的场景。假设我们用普通的HashMap来实现这个功能，代码可能是这样的：

public class HashMapDemo { private static final HashMap<String, Integer> USER_ACCESS_COUNT = new HashMap<>(); public static void main(String[] args) { Integer count = USER_ACCESS_COUNT.get("tom"); if (count == null) { USER_ACCESS_COUNT.put("tom", 1); } else { USER_ACCESS_COUNT.put("tom", count + 1); } } }

这段代码在单线程环境下运行没有问题，但在多线程环境下就会出大问题。想象一下，如果两个线程同时执行这段代码，都发现"tom"的计数为null，就会同时执行put操作，最终结果可能不是我们期望的2，而是1。这就是典型的线程安全问题。

1.1 HashMap的线程安全问题

HashMap在多线程环境下主要存在两个问题：

1. 数据覆盖：多个线程同时put时可能导致数据被覆盖
2. 死循环：在扩容时可能形成环形链表，导致CPU飙升

我曾在生产环境遇到过因为HashMap导致的CPU100%问题，排查了大半天才发现是并发扩容导致的死循环。这种问题一旦发生，系统基本就瘫痪了。

1.2 HashTable的局限性

为了解决线程安全问题，Java早期提供了HashTable，它通过在方法上加synchronized关键字来保证线程安全。但这种方式锁的粒度太大，相当于把整个哈希表锁住了，所有操作都要串行执行，性能非常差。

public class HashTableDemo { private static final Hashtable<String, Integer> USER_ACCESS_COUNT = new Hashtable<>(); public static void main(String[] args) { Integer count = USER_ACCESS_COUNT.get("tom"); if (count == null) { USER_ACCESS_COUNT.put("tom", 1); } else { USER_ACCESS_COUNT.put("tom", count + 1); } } }

1.3 ConcurrentHashMap的优势

ConcurrentHashMap通过更细粒度的锁机制解决了上述问题：

1. 分段锁技术（JDK7）：将数据分成多个段，每个段独立加锁
2. CAS+synchronized（JDK8）：只在操作特定桶时加锁，其他线程可以继续操作其他桶

这种设计使得ConcurrentHashMap在读操作时完全不需要加锁，写操作也只需要锁住部分数据，大大提高了并发性能。实测下来，在高并发场景下，ConcurrentHashMap的吞吐量能达到HashTable的几十倍。

2. ConcurrentHashMap的核心设计

2.1 JDK7 vs JDK8实现对比

JDK7和JDK8的ConcurrentHashMap实现有显著不同：

JDK8的实现更加简洁高效，去掉了Segment的概念，直接使用Node数组作为存储结构，锁的粒度也从段级别降低到了桶级别。

2.2 关键属性解析

先来看几个核心属性：

transient volatile Node<K,V>[] table; // 哈希桶数组 private transient volatile Node<K,V>[] nextTable; // 扩容时使用 private transient volatile int sizeCtl; // 控制标识符

sizeCtl这个属性特别重要，它承担了多个角色：

• -1：表示正在初始化
• -N：表示有N-1个线程正在进行扩容
• 正数：表示下一次扩容的阈值

这种一个变量表示多种状态的设计非常巧妙，我在实际开发中也经常借鉴这种思路。

2.3 哈希算法

ConcurrentHashMap使用spread方法对key的hashCode进行再哈希：

static final int spread(int h) { return (h ^ (h >>> 16)) & HASH_BITS; }

这个算法将高位特征扩散到低位，同时保证结果总是正数（最高位为0）。这样可以减少哈希冲突，同时避免特殊哈希值导致的问题。

3. 核心方法源码解析

3.1 put方法实现

put方法是ConcurrentHashMap最核心的方法之一，我们来看JDK8的实现：

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<K,V>(hash, key, value, null))) break; // CAS成功插入新节点 } 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<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; } } } } if (binCount != 0) { if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); // 链表转红黑树 if (oldVal != null) return oldVal; break; } } } addCount(1L, binCount); // 更新计数 return null; }

put方法的执行流程可以总结为：

1. 计算key的哈希值
2. 如果表为空，先初始化
3. 如果目标桶为空，直接CAS插入
4. 如果桶正在迁移，先协助迁移
5. 否则锁住桶的头节点，执行插入
6. 插入后检查是否需要树化
7. 最后更新元素计数

3.2 get方法实现

get方法不需要加锁，因为Node的val和next都是volatile的：

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; } 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; }

get方法的执行流程：

1. 计算key的哈希值
2. 找到对应的桶
3. 如果是链表，遍历查找
4. 如果是树，调用树的查找方法
5. 找到返回value，否则返回null

3.3 扩容机制

ConcurrentHashMap的扩容是个复杂的过程，支持多线程协同扩容：

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) { sizeCtl = Integer.MAX_VALUE; return; } nextTable = nextTab; transferIndex = n; } // ... 省略部分代码 }

扩容过程的关键点：

1. 创建新表，大小是原表的2倍
2. 将原表分成多个区间，每个线程负责一个区间
3. 迁移数据时采用高低位拆分法
4. 迁移完成后用新表替换旧表

我曾在一次性能优化中观察到，ConcurrentHashMap的扩容确实能充分利用多核CPU的优势，多个线程同时迁移数据，速度比单线程快很多。

4. 高级API使用技巧

4.1 computeIfAbsent

这个方法特别适合做缓存，我经常用它来实现本地缓存：

ConcurrentHashMap<String, Object> cache = new ConcurrentHashMap<>(); public Object getFromCache(String key) { return cache.computeIfAbsent(key, k -> { // 如果key不存在，执行这个函数计算value return expensiveOperation(k); }); }

4.2 merge

merge方法非常适合做计数统计：

ConcurrentHashMap<String, Integer> counts = new ConcurrentHashMap<>(); public void increment(String key) { counts.merge(key, 1, Integer::sum); }

4.3 forEach

并行遍历：

map.forEach(2, (k, v) -> System.out.println(k + "=" + v));

5. 实战案例分析

让我们回到文章开头的用户访问统计场景，用ConcurrentHashMap实现一个线程安全的计数器：

public class AccessCounter { private final ConcurrentHashMap<String, AtomicLong> counter = new ConcurrentHashMap<>(); public void increment(String user) { counter.computeIfAbsent(user, k -> new AtomicLong()).incrementAndGet(); } public long getCount(String user) { return counter.getOrDefault(user, new AtomicLong()).get(); } }

这个实现有几个优点：

1. 线程安全
2. 高性能
3. 代码简洁
4. 避免了不必要的对象创建

在实际项目中，这种计数器可以用来统计各种业务指标，比如页面PV、接口调用次数等。