摘要：
｛14//只有当第一个请求传入时，create[0]才能为true 15 create[0]=true；17loaderLock.signal=newCountDownLatch（1）；然后成功被修改为true 29ll.success=true；这样当下一个请求传入后，create[0]可以为true 33 loaderMap.remove（redisKey）；
本文代码逻辑思想来自阿里的JetCache框架，这里只是自己的学习与理解，记录下；具体实现可以去查看JetCache源码：git地址：https://github.com/alibaba/jetcache
实际应用中可以接JetCache框架，使用@CachePenetrationProtect注解即可实现
当缓存访问未命中的情况下，对并发进行的加载行为进行保护。 当前版本实现的是单JVM内的保护，即同一个JVM中同一个key只有一个线程去加载，其它线程等待结果。
1 // 本地缓存map 2 static Map<String, LoaderLock> loaderMap = new ConcurrentHashMap<>(); 3 4 static Object cachePenetrationProtectTest() { 5 6 String redisKey = "redisKey"; 7 8 // TODO 查缓存，查到则直接退出，没查到则继续往下执行，查询数据库 9
// 源码中这里加了while(true),如果查询数据库操作抛异常(ll.isSuccess为false)，岂不是一直卡在循环中了？ 10 Object loadedValue; 11 // 用boolean数组，而不是boolean变量，应该没什么特别原因，主要是computeIfAbsent中的变量需要final修饰 12 boolean create[] = new boolean[1]; 13 LoaderLock ll = loaderMap.computeIfAbsent(redisKey, key -> { 14 // 只有第一个请求进来时create[0]才会为true 15 create[0] = true; 16 LoaderLock loaderLock = new LoaderLock(); 17 loaderLock.signal = new CountDownLatch(1); 18 return loaderLock; 19 }); 20 // 是第一个进来的请求 21 if (create[0]) { 22 try { 23 // TODO 执行具体业务代码，这里loadedValue暂时返回null(实际应该是业务代码执行后的返回值) 24 loadedValue = null; 25 // 这里存放执行业务代码后需要返回的值 26 ll.value = loadedValue; 27 28 // 业务代码执行成功没抛异常，则success修改为true 29 ll.success = true; 30 return loadedValue; 31 } finally { 32 // 删除掉map中的lockKey值，使下个请求进来的时候create[0]可以为true 33 loaderMap.remove(redisKey); 34 // 让其他线程结束等待 35 ll.signal.countDown(); 36 } 37 } else { 38 try { 39 // 其他请求在这等待，设置个超时时间，可以做成可配 40 ll.signal.await(5, TimeUnit.SECONDS); 41 if (ll.success) { 42 return ll.value; 43 } else { 44 // TODO 数据库查询异常，这里可以抛出异常，直接返回请求，配合熔断措施处理 45 throw new RuntimeException("queryDB exception"); 46 } 47 } catch (InterruptedException e) { 48 throw new CacheException("loader wait interrupted", e); 49 } 50 } 51 } 52 53 static class LoaderLock { 54 CountDownLatch signal; 55 volatile boolean success; 56 volatile Object value; 57 }
JetCache部分源码(2.6.0版本)，synchronizedLoad方法：
static <K, V> V synchronizedLoad(CacheConfig config, AbstractCache<K,V> abstractCache, K key, Function<K, V> newLoader, Consumer<V> cacheUpdater) { ConcurrentHashMap<Object, LoaderLock> loaderMap = abstractCache.initOrGetLoaderMap(); Object lockKey = buildLoaderLockKey(abstractCache, key); while (true) { boolean create[] = new boolean[1]; LoaderLock ll = loaderMap.computeIfAbsent(lockKey, (unusedKey) -> { create[0] = true; LoaderLock loaderLock = new LoaderLock(); loaderLock.signal = new CountDownLatch(1); loaderLock.loaderThread = Thread.currentThread(); return loaderLock; }); if (create[0] || ll.loaderThread == Thread.currentThread()) { try { V loadedValue = newLoader.apply(key); ll.success = true; ll.value = loadedValue; cacheUpdater.accept(loadedValue); return loadedValue; } finally { if (create[0]) { ll.signal.countDown(); loaderMap.remove(lockKey); } } } else { try { Duration timeout = config.getPenetrationProtectTimeout(); if (timeout == null) { ll.signal.await(); } else { boolean ok = ll.signal.await(timeout.toMillis(), TimeUnit.MILLISECONDS); if(!ok) { logger.info("loader wait timeout:" + timeout); return newLoader.apply(key); } } } catch (InterruptedException e) { logger.warn("loader wait interrupted"); return newLoader.apply(key); } if (ll.success) { return (V) ll.value; } else { continue; } } } } static class LoaderLock { CountDownLatch signal; Thread loaderThread; volatile boolean success; volatile Object value; }