聊聊JetCache的CachePenetrationProtect

序

本文主要研究一下JetCache的CachePenetrationProtect

CachePenetrationProtect

com/alicp/jetcache/anno/CachePenetrationProtect.java

@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD, ElementType.FIELD})
public @interface CachePenetrationProtect {
 boolean value() default true;
 int timeout() default CacheConsts.UNDEFINED_INT;
 TimeUnit timeUnit() default TimeUnit.SECONDS;
}

它定义value、timeout、timeUnit属性

computeIfAbsentImpl

com/alicp/jetcache/AbstractCache.java

 static <K, V> V computeIfAbsentImpl(K key, Function<K, V> loader, boolean cacheNullWhenLoaderReturnNull,
 long expireAfterWrite, TimeUnit timeUnit, Cache<K, V> cache) {
 AbstractCache<K, V> abstractCache = CacheUtil.getAbstractCache(cache);
 CacheLoader<K, V> newLoader = CacheUtil.createProxyLoader(cache, loader, abstractCache::notify);
 CacheGetResult<V> r;
 if (cache instanceof RefreshCache) {
 RefreshCache<K, V> refreshCache = ((RefreshCache<K, V>) cache);
 r = refreshCache.GET(key);
 refreshCache.addOrUpdateRefreshTask(key, newLoader);
 } else {
 r = cache.GET(key);
 }
 if (r.isSuccess()) {
 return r.getValue();
 } else {
 Consumer<V> cacheUpdater = (loadedValue) -> {
 if(needUpdate(loadedValue, cacheNullWhenLoaderReturnNull, newLoader)) {
 if (timeUnit != null) {
 cache.PUT(key, loadedValue, expireAfterWrite, timeUnit).waitForResult();
 } else {
 cache.PUT(key, loadedValue).waitForResult();
 }
 }
 };
 V loadedValue;
 if (cache.config().isCachePenetrationProtect()) {
 loadedValue = synchronizedLoad(cache.config(), abstractCache, key, newLoader, cacheUpdater);
 } else {
 loadedValue = newLoader.apply(key);
 cacheUpdater.accept(loadedValue);
 }
 return loadedValue;
 }
 }

AbstractCache的computeIfAbsentImpl方法,在cache.config().isCachePenetrationProtect()执行的是synchronizedLoad

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

synchronizedLoad会判断如果loaderThread不是当前线程,则通过ll.signal.await进行等待,等待timeout或者InterruptedException时则执行加载newLoader.apply(key)

小结

JetCache提供了@CachePenetrationProtect注解,支持多线程并发去回源的时候,控制在指定超时时间内整个JVM中只有1个去回源,过了超时时间若未有结果则自行加载,若已有结果则返回loader线程加载的结果。

doc

• MethodCache