序
本文主要研究一下cache2k这款新型缓存
示例
代码语言：javascript复制Cache<String,String> cache = new Cache2kBuilder<String, String>() {}
.eternal(true)
.expireAfterWrite(5, TimeUnit.MINUTES) // expire/refresh after 5 minutes
.setupWith(UniversalResiliencePolicy::enable, b -> b // enable resilience policy
.resilienceDuration(30, TimeUnit.SECONDS) // cope with at most 30 seconds
// outage before propagating
// exceptions
)
.refreshAhead(true) // keep fresh when expiring
.loader(k -> expensiveOperation(k)) // auto populating function
.build();
常见问题的解决方案
空值问题
JCache规范不支持null，所以cache2k默认也不支持，不过可以通过permitNullValues(true)来开启，这样子缓存就可以存储null值
cache stampede问题
又称作cache miss storm，指的是高并发场景缓存同时失效导致大面积回源，cache2k采用的是block的请求方式，避免对同一个key并发回源
org/cache2k/core/HeapCache.java
代码语言：javascript复制protected Entry<K, V> getEntryInternal(K key, int hc, int val) {
if (loader == null) {
return peekEntryInternal(key, hc, val);
}
Entry<K, V> e;
for (;;) {
e = lookupOrNewEntry(key, hc, val);
if (e.hasFreshData(clock)) {
return e;
}
synchronized (e) {
e.waitForProcessing();
if (e.hasFreshData(clock)) {
return e;
}
if (e.isGone()) {
metrics.heapHitButNoRead();
metrics.goneSpin();
continue;
}
e.startProcessing(Entry.ProcessingState.LOAD, null);
break;
}
}
boolean finished = false;
try {
load(e);
finished = true;
} finally {
e.ensureAbort(finished);
}
if (e.getValueOrException() == null && isRejectNullValues()) {
return null;
}
return e;
}
同步回源造成的接口稳定性问题
cache2k提供了refreshAhead参数，在新数据没有拉取成功之前，过期数据仍然可以访问，避免请求到来时发现数据过期触发同步回源造成接口延时增大问题。不过具体底层还依赖prefetchExecutor，如果refresh的时候没有足够的线程可以使用则会立马过期，等待下次get出发同步回源
org/cache2k/core/HeapCache.java
代码语言：javascript复制public void timerEventRefresh(Entry<K, V> e, Object task) {
metrics.timerEvent();
synchronized (e) {
if (e.getTask() != task) { return; }
try {
refreshExecutor.execute(createFireAndForgetAction(e, Operations.SINGLETON.refresh));
} catch (RejectedExecutionException ex) {
metrics.refreshRejected();
expireOrScheduleFinalExpireEvent(e);
}
}
}
默认的executor如下，采用的是SynchronousQueue队列，可以通过builder自己去设置refreshExecutor
代码语言：javascript复制 Executor provideDefaultLoaderExecutor(int threadCount) {
int corePoolThreadSize = 0;
return new ThreadPoolExecutor(corePoolThreadSize, threadCount,
21, TimeUnit.SECONDS,
new SynchronousQueue<>(),
threadFactoryProvider.newThreadFactory(getThreadNamePrefix()),
new ThreadPoolExecutor.AbortPolicy());
}
回源故障问题
针对回源的下游出现故障的问题，cache2k提供了ResiliencePolicy策略，其实现类为UniversalResiliencePolicy 当load方法抛出异常且cache里头还有数据的时候，异常不会抛给client，用当前的数据返回，这里有个resilienceDuration时间，如果超过这个时间load方法还继续抛出异常则异常会抛给client。如果没有单独设置resilienceDuration，则默认取的是expiryAfterWrite时间
org/cache2k/core/HeapCache.java
代码语言：javascript复制private Object loadGotException(Entry<K, V> e, long t0, long t, Throwable wrappedException) {
ExceptionWrapper<K, V> exceptionWrapper =
new ExceptionWrapper(keyObjFromEntry(e), wrappedException, t0, e, exceptionPropagator);
long expiry = 0;
long refreshTime = 0;
boolean suppressException = false;
RefreshAheadPolicy.Context<Object> refreshCtx;
try {
if (e.isValidOrExpiredAndNoException()) {
expiry = timing.suppressExceptionUntil(e, exceptionWrapper);
}
if (expiry > t0) {
suppressException = true;
} else {
expiry = timing.cacheExceptionUntil(e, exceptionWrapper);
}
refreshCtx = getContext(e, t0, t, true, true, false, expiry);
refreshTime = timing.calculateRefreshTime(refreshCtx);
} catch (Exception ex) {
return resiliencePolicyException(e, t0, t, new ResiliencePolicyException(ex), null);
}
exceptionWrapper = new ExceptionWrapper<>(exceptionWrapper, Math.abs(expiry));
Object wrappedValue = exceptionWrapper;
if (expiry != 0) {
wrappedValue = timing.wrapLoadValueForRefresh(refreshCtx, e, exceptionWrapper);
}
Object loadResult;
synchronized (e) {
insertUpdateStats(e, (V) wrappedValue, t0, t, true, expiry, suppressException);
if (suppressException) {
e.setSuppressedLoadExceptionInformation(exceptionWrapper);
loadResult = e.getValueOrException();
} else {
if (isRecordModificationTime()) {
e.setModificationTime(t0);
}
e.setValueOrWrapper(exceptionWrapper);
loadResult = exceptionWrapper;
}
finishLoadOrEviction(e, expiry, refreshTime);
}
return loadResult;
}
这里timing.suppressExceptionUntil是委托给了ResiliencePolicy#suppressExceptionUntil
cache2k-addon/src/main/java/org/cache2k/addon/UniversalResiliencePolicy.java
代码语言：javascript复制public long suppressExceptionUntil(K key,
LoadExceptionInfo<K, V> loadExceptionInfo,
CacheEntry<K, V> cachedEntry) {
if (resilienceDuration == 0 || resilienceDuration == Long.MAX_VALUE) {
return resilienceDuration;
}
long maxSuppressUntil = loadExceptionInfo.getSinceTime() resilienceDuration;
long deltaMs = calculateRetryDelta(loadExceptionInfo);
return Math.min(loadExceptionInfo.getLoadTime() deltaMs, maxSuppressUntil);
}
UniversalResiliencePolicy还提供了异常重试的功能，重试间隔为retryInterval，如果没有配置则为resilienceDuration的5%，采取的是指数退避的模式，factor为1.5
小结
cache2k提供了Guava Cache及Caffeine没有的ResiliencePolicy，针对C端高并发场景提供了容错的功能，值得借鉴一下。
个人公众号「码匠的流水账」(geek_luandun)，欢迎关注
doc
- cache2k
- cache2k User Guide
- Introduction to cache2k
- caffeine
- guava cache
