1

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

this.localeWeatherCache = 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;
                    }));

}
4

2 回答 2

3

AAsyncLoadingCache可以根据键计算值并返回CompletableFuture结果的 a。这可以翻译成Mono它的fromFuture方法。这将确保对于给定的键只有一次执行正在进行中，而不会由于将期货存储在缓存中而阻塞。

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;
      }));
}
于 2019-03-15T22:52:48.393 回答
2

如这里https://stackoverflow.com/a/52803247/11209784所示，ClimaTempoLocale可以按如下方式计算：

Cache<Location, ClimaTempoLocale> weatherLocaleCache = Caffeine.newBuilder().build();

private Mono<ClimaTempoLocale> findLocale(Location location) {
    Mono<ClimaTempoLocale> locale;
    ClimaTempoLocale cachedLocale = weatherLocaleCache.getIfPresent(location);
    if (cachedLocale != null) {
        locale = Mono.just(cachedLocale);
    } else {
        locale = climaTempoRepository.findLocaleByCityNameAndState(location.city(), location.state())
                .doOnNext(climaTempoLocale -> weatherLocaleCache.put(location, climaTempoLocale));
    }

    return locale;
}

一个副作用是，当并发调用导致缓存未命中时，可能会连续写入同一个键。

通过这样做，依赖于的调用ClimaTempoLocale可以以相同的方式继续：

Cache<Location, CurrentWeather> currentWeatherCache = Caffeine.newBuilder().expireAfterWrite(Duration.ofHours(3)).build();

Cache<Location, WeatherForecasts> weatherForecastsCache = Caffeine.newBuilder().expireAfterWrite(Duration.ofHours(12)).build();

public Mono<Weather> weatherForecastByLocation(Location location) {
    Mono<ClimaTempoLocale> locale = findLocale(location);

    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;
                    }));
}
于 2019-03-16T14:44:05.293 回答