AsyncCacheJVM 级 SingleFlight 实现原理与最佳实践本文基于CaffeineAsyncCacheJava 8适用于需要高并发下防止缓存击穿、避免重复加载的场景。一、核心目标同一 Key 只加载一次在高并发场景下缓存失效瞬间可能出现缓存击穿Cache Breakdown大量线程同时发现缓存缺失同时去 DB / RPC 加载同一份数据。AsyncCache的目标就是无论多少线程并发访问同一个 Key 只触发一次加载逻辑。二、核心实现原理1. 原子性保障ConcurrentHashMap.computeIfAbsentAsyncCache底层依赖ConcurrentHashMap核心逻辑等价于CompletableFutureVfuturemap.computeIfAbsent(key,k-{// ✅ 同一时刻只有一个线程能进入此处CompletableFutureVfnewCompletableFuture();executor.execute(()-{try{VvalueloadFromDb(k);f.complete(value);}catch(Throwablet){f.completeExceptionally(t);map.remove(k,f);// 加载失败允许重试}});returnf;});关键保证特性说明原子性computeIfAbsent对同一 key 所在桶加锁保证创建 Future 是原子操作可见性Node.value和next用volatile修饰Happens-Before 规则保证对其他线程立即可见唯一性同一 key 永远只创建一个CompletableFuture等待机制未抢到锁的线程直接拿到已有 Future自然等待结果✅这就是 JVM 级的 SingleFlight 实现2. Java 8 的并发控制CAS synchronized桶级锁⚠️重要更正Java 8 中ConcurrentHashMap已废弃 JDK 7 的 Segment 分段锁Striped Locking改用CAS 无锁 synchronized桶级锁的混合策略。JDK 7 vs JDK 8 对比维度JDK 7已淘汰JDK 8当前主流数据结构Segment[]HashEntry[] 链表Node[] 链表 / 红黑树锁机制ReentrantLock分段锁Striped LockingCAS synchronized桶级锁锁粒度Segment 级别默认 16 段单个桶的头节点并发度 数组长度读操作volatile保证可见性volatile保证可见性完全无锁写操作先获取 Segment 锁先 CAS 尝试失败再synchronized锁桶扩容单个 Segment 独立扩容多线程协同扩容ForwardingNode标记哈希冲突退化纯链表 O(n)链表 ≥ 8 且容量 ≥ 64 时转红黑树 O(log n)computeIfAbsent的执行流程Java 8线程 T1、T2、T3 同时调用 computeIfAbsent(sameKey, ...) │ ▼ ① 计算 hash定位到桶下标 i │ ▼ ② 桶为空── 是 ──→ CAS 直接插入无锁快路径 │ 否 ▼ ③ synchronized 锁住桶的头节点 │ ▼ ④ 再次检查 key 是否仍不存在double check │ ▼ ⑤ 只有一个线程执行 mappingFunction │ ▼ ⑥ 释放锁其他线程拿到同一个 Future核心要点CAS 优先无竞争时完全无锁性能极高synchronized锁桶头节点只有发生哈希冲突时才加锁且锁粒度极小锁升级机制JVM 会自动将synchronized从偏向锁 → 轻量级锁 → 重量级锁逐步升级绝大多数场景停留在轻量级锁阶段不同桶之间完全无竞争并发度约等于桶数组长度默认 16可随扩容增长为什么 Java 8 选择synchronized而非ReentrantLock对比项ReentrantLockJDK 7synchronizedJDK 8锁粒度Segment 级较粗桶头节点级更细JVM 优化无特殊优化偏向锁、轻量级锁、自旋、锁消除、锁粗化内存开销每个 Segment 一个锁对象锁信息内嵌在对象头中零额外对象可中断支持lockInterruptibly()不支持但缓存场景不需要公平性可配置不可配置但 FIFO 等待已足够实际性能较好更优尤其高并发 短临界区结论在ConcurrentHashMap这种锁持有时间极短、不需要条件变量和中断特性的场景下synchronized经过 JVM 优化后性能全面优于ReentrantLock且零内存开销。3. 自动清理与内存安全Future 完成后引用由 GC 自动回收加载失败时主动remove(key)避免空值缓存永久阻塞内存泄漏三、推荐写法✅ 基础用法最推荐publicStringgetData(Stringkey){returnasyncCache.get(key,(k,exec)-CompletableFuture.supplyAsync(()-loadDataFromDb(k),exec)).join();}execCaffeine 内置ForkJoinPool生产环境建议自定义线程池join()阻塞等待结果适合非响应式服务✅ 带超时保护防止线程堆积publicStringgetDataWithTimeout(Stringkey){returnasyncCache.get(key,(k,exec)-CompletableFuture.supplyAsync(()-loadDataFromDb(k),exec).completeOnTimeout(fallback,2,TimeUnit.SECONDS)).join();}✅ 防止DB 慢查询RPC 无限阻塞线程池被打爆✅ 防止缓存污染加载失败不缓存publicStringgetDataSafe(Stringkey){returnasyncCache.get(key,(k,exec)-CompletableFuture.supplyAsync(()-loadDataFromDb(k),exec).exceptionally(ex-{asyncCache.synchronous().invalidate(k);thrownewRuntimeException(ex);})).join();}非常重要否则失败结果会被缓存导致后续请求全部失败。五、原子更新 Value如果你是想做CAS 风格更新cache.asMap().compute(key,(k,oldFuture)-{if(oldFuturenull){returnCompletableFuture.completedFuture(init);}returnoldFuture.thenApply(v-v_updated);});⚠️ 注意AsyncCache中 Value 是CompletableFuture更新成本较高通常不建议频繁使用六、并发测试验证publicstaticvoidmain(String[]args){AsyncCacheString,StringcacheCaffeine.newBuilder().buildAsync();AtomicIntegerloadCountnewAtomicInteger(0);ListCompletableFutureStringfuturesIntStream.range(0,10).parallel().mapToObj(i-cache.get(sameKey,(key,exec)-{System.out.println(Thread.currentThread().getName() loading...);loadCount.incrementAndGet();returnCompletableFuture.supplyAsync(()-Data-key);})).toList();futures.get(0).whenComplete((v,ex)-{System.out.println(Result: v);});System.out.println(Load count loadCount.get());// ✅ 永远是 1}典型输出ForkJoinPool.commonPool-worker-1 loading... Result: Data-sameKey Load count 1✅完美证明仅一个线程执行加载逻辑七、与 Redis 分布式锁对比维度AsyncCacheJVM 级Redis Lock分布式作用范围单 JVM 内跨 JVM / 跨机器锁机制CAS synchronized桶级锁SETNX / Redlock性能⭐⭐⭐⭐⭐纳秒~微秒级⭐⭐毫秒级 网络 IO复杂度低开箱即用高需处理超时、死锁、脑裂网络 IO无有每次加锁至少 1 次 RTT适用场景单机本地缓存防击穿分布式协调、跨服务互斥✅结论能使用AsyncCache解决的场景不要用 Redis 锁。二者不是替代关系而是互补——分布式场景仍需 Redis单机高并发场景AsyncCache是更优解。八、避坑指南❌ 坑 1在computeIfAbsent的 Lambda 中再次操作同一个 Map// 错误示范可能导致死锁map.computeIfAbsent(key,k-{map.put(otherKey,someValue);// ⚠️ Lambda 内可能持有桶锁再次操作可能死锁returncomputeValue();});✅正确做法Lambda 内只做纯计算不涉及任何 Map 写操作。❌ 坑 2mappingFunction返回nullcomputeIfAbsent的mappingFunction不允许返回 null否则抛出NullPointerException。// 错误cache.get(key,k-null);// NPE!// 正确返回包装类型或 Optionalcache.get(key,k-CompletableFuture.completedFuture(null));// OK[citation:6]❌ 坑 3mappingFunction执行时间过长computeIfAbsent在执行 Lambda 时持有桶锁虽然时间极短如果 Lambda 内做耗时操作会阻塞同一桶的其他操作。✅正确做法Lambda 内只创建CompletableFuture实际计算交给异步线程// ✅ 推荐Lambda 立即返回 Future计算异步执行cache.get(key,(k,exec)-{returnCompletableFuture.supplyAsync(()-loadDataFromDb(k),exec);});九、总结CaffeineAsyncCache通过ConcurrentHashMap.computeIfAbsent实现了 JVM 级的 SingleFlight。在 Java 8 中底层依赖 CAS 无锁 synchronized桶级锁的混合策略以极低的锁开销保证同一 Key 只加载一次。核心要点速查要点一句话说明原子性来源ConcurrentHashMap.computeIfAbsentJava 8CAS synchronized桶级锁锁粒度单个桶的头节点不同桶之间零竞争无锁路径桶为空时 CAS 直接插入完全无锁等待机制所有线程共享同一个CompletableFuture失败处理主动remove允许重试超时保护completeOnTimeout防止永久阻塞分布式场景仍需 Redis / DB 层协调最大陷阱Lambda 内不要操作同一个 Map不要返回 null十、参考与延伸阅读JDK 源码ConcurrentHashMap.computeIfAbsent()JDK 8u60Caffeine 官方文档https://github.com/ben-manes/caffeineJava 8synchronized锁升级机制偏向锁 → 轻量级锁 → 重量级锁CompletableFuture超时 APIJava 9completeOnTimeout/ Java 8 GuavaFutures.withTimeout
)
