如何在 Project Reactor 中缓存项目并避免 Cache Stampede？答案

【问题标题】：How to cache items in Project Reactor and avoid a Cache Stampede?如何在 Project Reactor 中缓存项目并避免 Cache Stampede？
【发布时间】：2018-08-14 18:15:58
【问题描述】：

如何执行非阻塞的获取/计算查找并避免缓存踩踏。

这是一个不会踩踏，而是阻塞的例子。

public static <KEY, VALUE> Mono<VALUE> lookupAndWrite(
    Map<KEY, Signal<? extends VALUE>> cacheMap, KEY key, Mono<VALUE> mono) {
    return Mono.defer(() -> Mono.just(cacheMap.computeIfAbsent(key, k -> 
        mono.materialize().block())).dematerialize());
}

这是一个不会阻塞，但可以踩踏的例子。

public static <KEY, VALUE> MonoCacheBuilderCacheMiss<KEY, VALUE> lookup(
        Function<KEY, Mono<Signal<? extends VALUE>>> reader, KEY key) {
    return otherSupplier -> writer -> Mono.defer(() ->
            reader.apply(key)
                    .switchIfEmpty(otherSupplier.get()
                            .materialize()
                            .flatMap(signal -> writer.apply(key, signal)
                            )
                    )
                    .dematerialize());
}

有没有不会踩踏或阻塞的方法？只在自己的调度程序上订阅阻塞调用有意义吗？

【问题讨论】：

在这里使用AsyncLoadingCache 有意义吗？然后你不会在计算过程中阻塞，缓存会在未来值完成（成功或错误）时添加回调。（P.S.AsyncCache 将在下一个版本中，但很容易模仿）
@BenManes 我不确定。密钥没有包含足够的信息来知道如何加载数据（也许这无关紧要）。
您可以避免get(key) 调用并让加载程序抛出异常，以降级为临时AsyncCache。然后你可以用类似cache.get(key, k -> mono.toFuture())的方式编写你的代码
由于 SO 不适合讨论，如果您愿意，我们可以将其带回您的 github issue。
@BenManes 如果您认为这是最好的。我在这里提出它是为了让更多人关注它，因为这似乎是一个非常常见的问题，但尚未提出（对于这个框架）。

标签： java-8 project-reactor caffeine

【解决方案1】：

为了重新表述您的问题，您希望在允许异步执行计算的同时避免踩踏。理想情况下，这将使用 ConcurrentMap<K, Mono<V>> 和 computeIfAbsent 来完成，如果计算失败，则会丢弃该条目。

Caffeine 的AsyncLoadingCache 通过使用CompletableFuture<V> 作为值类型来提供这种类型的行为。你可以将你的阻塞函数重写为

public static <KEY, VALUE> Mono<VALUE> lookupAndWrite(
    AsyncLoadingCache<KEY, VALUE> cache, KEY key, Mono<VALUE> mono) {
  return Mono.defer(() -> Mono.fromFuture(cache.get(key, (k,e) ->
            mono.subscribeOn(Schedulers.fromExecutor(e)).toFuture())));
}

从 2.6.x 版本开始，如果需要，没有更简单的 AsyncCache 来听取反馈，它将在 2.7 版本中发布。这还将包括一个ConcurrentMap<K, CompletableFuture<V>> 视图，它可以让您概括您的方法以不具有特定于提供程序的接口。目前，您可以通过避免加载方法并使用Caffeine.newBuilder().buildAsync(key -> null) 来模拟非加载缓存。

【讨论】：

谢谢先生，有没有办法控制是否缓存了出错的Mono。
不，因为同步 ConcurrentMap 不会缓存异常或空值，异步缓存在未来也会这样做。如果你想要一个负缓存，那么转换成你可以查询的成功结果，例如Mono<Optional<V>>。这种方法似乎造成的混乱最少，保持接口一致，并且不会使用不同的处理策略使缓存复杂化。