1. 为什么需要ConcurrentHashMap在日常开发中我们经常会遇到需要统计用户访问次数的场景。假设我们用普通的HashMap来实现这个功能代码可能是这样的public class HashMapDemo { private static final HashMapString, 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在多线程环境下主要存在两个问题数据覆盖多个线程同时put时可能导致数据被覆盖死循环在扩容时可能形成环形链表导致CPU飙升我曾在生产环境遇到过因为HashMap导致的CPU100%问题排查了大半天才发现是并发扩容导致的死循环。这种问题一旦发生系统基本就瘫痪了。1.2 HashTable的局限性为了解决线程安全问题Java早期提供了HashTable它通过在方法上加synchronized关键字来保证线程安全。但这种方式锁的粒度太大相当于把整个哈希表锁住了所有操作都要串行执行性能非常差。public class HashTableDemo { private static final HashtableString, 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通过更细粒度的锁机制解决了上述问题分段锁技术JDK7将数据分成多个段每个段独立加锁CASsynchronizedJDK8只在操作特定桶时加锁其他线程可以继续操作其他桶这种设计使得ConcurrentHashMap在读操作时完全不需要加锁写操作也只需要锁住部分数据大大提高了并发性能。实测下来在高并发场景下ConcurrentHashMap的吞吐量能达到HashTable的几十倍。2. ConcurrentHashMap的核心设计2.1 JDK7 vs JDK8实现对比JDK7和JDK8的ConcurrentHashMap实现有显著不同特性JDK7JDK8数据结构Segment数组HashEntry链表Node数组链表/红黑树锁机制分段锁ReentrantLockCASsynchronized并发度Segment数量固定动态变化扩容分段扩容协助扩容哈希冲突链表链表红黑树JDK8的实现更加简洁高效去掉了Segment的概念直接使用Node数组作为存储结构锁的粒度也从段级别降低到了桶级别。2.2 关键属性解析先来看几个核心属性transient volatile NodeK,V[] table; // 哈希桶数组 private transient volatile NodeK,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 (NodeK,V[] tab table;;) { NodeK,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 NodeK,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 (NodeK,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; } NodeK,V pred e; if ((e e.next) null) { pred.next new NodeK,V(hash, key, value, null); break; } } } else if (f instanceof TreeBin) { // 树节点 NodeK,V p; binCount 2; if ((p ((TreeBinK,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方法的执行流程可以总结为计算key的哈希值如果表为空先初始化如果目标桶为空直接CAS插入如果桶正在迁移先协助迁移否则锁住桶的头节点执行插入插入后检查是否需要树化最后更新元素计数3.2 get方法实现get方法不需要加锁因为Node的val和next都是volatile的public V get(Object key) { NodeK,V[] tab; NodeK,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方法的执行流程计算key的哈希值找到对应的桶如果是链表遍历查找如果是树调用树的查找方法找到返回value否则返回null3.3 扩容机制ConcurrentHashMap的扩容是个复杂的过程支持多线程协同扩容private final void transfer(NodeK,V[] tab, NodeK,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) NodeK,V[] nt (NodeK,V[])new Node?,?[n 1]; nextTab nt; } catch (Throwable ex) { sizeCtl Integer.MAX_VALUE; return; } nextTable nextTab; transferIndex n; } // ... 省略部分代码 }扩容过程的关键点创建新表大小是原表的2倍将原表分成多个区间每个线程负责一个区间迁移数据时采用高低位拆分法迁移完成后用新表替换旧表我曾在一次性能优化中观察到ConcurrentHashMap的扩容确实能充分利用多核CPU的优势多个线程同时迁移数据速度比单线程快很多。4. 高级API使用技巧4.1 computeIfAbsent这个方法特别适合做缓存我经常用它来实现本地缓存ConcurrentHashMapString, Object cache new ConcurrentHashMap(); public Object getFromCache(String key) { return cache.computeIfAbsent(key, k - { // 如果key不存在执行这个函数计算value return expensiveOperation(k); }); }4.2 mergemerge方法非常适合做计数统计ConcurrentHashMapString, 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 ConcurrentHashMapString, 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(); } }这个实现有几个优点线程安全高性能代码简洁避免了不必要的对象创建在实际项目中这种计数器可以用来统计各种业务指标比如页面PV、接口调用次数等。