spring-webflux - 使用 Reactor 和 Caffeine 缓存无法正常工作

Question

我需要在不同的时间间隔检查一些端点，所以我设置了 Caffeine 的缓存构建器。

this.localeWe​​atherCache = newBuilder().build();

this.currentWeatherCache=newBuilder().expireAfterWrite(Duration.ofHours(3)).build();

this.weatherForecastsCache = newBuilder().expireAfterWrite(Duration.ofHours(12)).build();

在我的服务中，我调用了这 3 个端点，最后我使用 Mono.zip() 返回了包含所有详细信息的对象。

在我的测试中，我注意到 climaTempoRepository.findLocaleByCityNameAndState 被执行了两次，并且在 currentWeather 缓存过期后，它再次调用了 locale 端点，weatherForecast 也会发生同样的情况，它会再次调用 locale。

为什么会失败？它不应该使用缓存吗？还是我做的方式不对？

非常感谢任何帮助或指点! :)

public Mono<Weather> weatherForecastByLocation(Location location) {

    Mono<ClimaTempoLocale> locale =
            CacheMono.lookup(key ->
                    Mono.justOrEmpty(localeWeatherCache.getIfPresent(key))
                            .map(Signal::next), location)
                    .onCacheMissResume(() -> climaTempoRepository.findLocaleByCityNameAndState(location.city(), location.state()))
                    .andWriteWith((key, signal) -> Mono.fromRunnable(() ->
                            Optional.ofNullable(signal.get())
                                    .ifPresent(value -> localeWeatherCache.put(key, value))));

    Mono<CurrentWeather> currentWeather =
            CacheMono.lookup(key ->
                    Mono.justOrEmpty(currentWeatherCache.getIfPresent(key))
                            .map(Signal::next), location)
                    .onCacheMissResume(() -> locale.flatMap(climaTempoRepository::findCurrentWeatherByLocale)
                            .subscribeOn(Schedulers.elastic()))
                    .andWriteWith((key, signal) -> Mono.fromRunnable(() ->
                            Optional.ofNullable(signal.get())
                                    .ifPresent(value -> currentWeatherCache.put(key, value))));

    Mono<WeatherForecasts> weatherForecasts =
            CacheMono.lookup(key ->
                    Mono.justOrEmpty(weatherForecastsCache.getIfPresent(key))
                            .map(Signal::next), location)
                    .onCacheMissResume(() -> locale.flatMap(climaTempoRepository::findDailyForecastByLocale)
                            .subscribeOn(Schedulers.elastic()))
                    .andWriteWith((key, signal) -> Mono.fromRunnable(() ->
                            Optional.ofNullable(signal.get())
                                    .ifPresent(value -> weatherForecastsCache.put(key, value))));

    return Mono.zip(currentWeather,
            weatherForecasts,
            (current, forecasts) ->
                    Weather.buildWith(builder -> {
                        builder.location = location;
                        builder.currentWeather = current;
                        builder.weatherForecasts = forecasts;
                    }));

}

Answer 1

AsyncLoadingCache 可以根据键计算值并返回结果的 CompletableFuture。这可以翻译成 Mono 它是 fromFuture 方法。这将确保对于给定的键只有一个执行在进行中，同时不会因为将 future 存储在缓存中而阻塞。

AsyncLoadingCache<Location, ClimaTempoLocale> localeWeatherCache = 
    Caffeine.newBuilder().buildAsync(location -> 
        climaTempoRepository.findLocaleByCityNameAndState(location.city(), location.state()));

AsyncLoadingCache<ClimaTempoLocale, CurrentWeather> currentWeatherCache =
    Caffeine.newBuilder().buildAsync(climaTempoRepository::findCurrentWeatherByLocale);

AsyncLoadingCache<ClimaTempoLocale, WeatherForecasts> weatherForecastsCache =
    Caffeine.newBuilder().buildAsync(climaTempoRepository::findDailyForecastByLocale);

public Mono<Weather> weatherForecastByLocation(Location location) {
  var locale = Mono.fromFuture(localeWeatherCache.get(location));
  var currentWeather = Mono.fromFuture(locale.map(localeWeatherCache::get));
  var weatherForecasts = Mono.fromFuture(locale.map(weatherForecastsCache::get));

  return Mono.zip(currentWeather, weatherForecasts, (current, forecasts) ->
      Weather.buildWith(builder -> {
          builder.location = location;
          builder.currentWeather = current;
          builder.weatherForecasts = forecasts;
      }));
}