使用 spring webflux 替代@Cacheable

Question

我需要缓存来自ReactiveMongoRepository. 数据大约每年更新两次，所以我不关心缓存是否过期。由于我们不能将 @Cacheable与Flux一起使用，我想找到一种直接、简单的方法来将来自 Mongo 的数据存储到 redis，并使用该数据（如果存在），否则存储它并提供原始数据。有没有比这更直接的方法  @GetMapping  public Flux<AvailableInspection> getAvailableInspectionsRedis() {    AtomicInteger ad = new AtomicInteger();    return availableInspectionReactiveRedisOperations.opsForZSet().range("availableInspections", Range.<Long>from(Range.Bound.inclusive(0L)).to(Range.Bound.inclusive(-1L)))        .switchIfEmpty(availableInspectionMongoRepository.findAll().map(e -> {          availableInspectionReactiveRedisOperations.opsForZSet().add("availableInspections", e, ad.getAndIncrement()).block();          return e;        }));  }我正在寻找的是一个选项，它可以让我像 @Cacheable 注释那样缓存数据。我正在寻找能够缓存任何类型的通量的通用解决方案。

Answer 1

我怀疑这个问题是否有现成的解决方案。但是，您可以轻松构建自己的界面来获取通用缓存对象并将它们加载到缓存中：

public interface GetCachedOrLoad<T> {

  Flux<T> getCachedOrLoad(String key, Flux<T> loader, Class<? extends T> clazz);

}

每个需要此功能的类都将通过构造函数注入它并按如下方式使用它：

public class PersistedObjectRepository {

  private final GetCachedOrLoad<PersistedObject> getCachedOrLoad;

  public PersistedObjectRepository(final GetCachedOrLoad<PersistedObject> getCachedOrLoad) {

    this.getCachedOrLoad = getCachedOrLoad;

  }

  public Flux<PersistedObject> queryPersistedObject(final String key) {

    return getCachedOrLoad.getCachedOrLoad(key, queryMongoDB(key), PersistedObject.class);

  }

  private Flux<PersistedObject> queryMongoDB(String key) {

    // use reactivemongo api to retrieve Flux<PersistedObject>

  }

}

然后你需要创建一个对象实现GetCachedOrLoad<T>并使其可用于依赖注入。

public class RedisCache<T> implements GetCachedOrLoad<T> {

  private final Function<String, Flux<String>> getFromCache;

  private final BiConsumer<String, String> loadToCache;

  private final Gson gson;

  public RedisCache(Gson gson, RedisReactiveCommands<String, String> redisCommands) {

    this.getFromCache = key -> redisCommands.lrange(key, 0, -1);

    this.loadToCache = redisCommands::lpush;

    this.gson = gson;

  }

  @Override

  public Flux<T> getCachedOrLoad(final String key, Flux<T> loader, Class<? extends T> clazz) {

    final Flux<T> cacheResults = getFromCache.apply(key)

      .map(json -> gson.fromJson(json, clazz));

    return cacheResults.switchIfEmpty(

      loader.doOnNext(value -> loadToCache.accept(key, gson.toJson(value))));

  }

}

希望这足够通用:)。

附注。这不是生产就绪的实现，需要根据您自己的需求进行调整，例如添加异常处理、自定义 json 序列化等。