反应式应用程序中的 Spring Cache

Question

我使用的是 Spring Cache 默认实现（ConcurrentHashMap）：

@Bean
CacheManager cacheManager() {
    ConcurrentMapCacheManager cacheManager = new ConcurrentMapCacheManager("mycache");
    cacheManager.setAllowNullValues(false);
    return cacheManager;
}

Reactor addons 用于获取物品。缓存未命中项写入缓存：

public Mono<T> get(ID key) {
    return CacheMono.lookup(k -> findValue(k).map(Signal::next), key)
                    .onCacheMissResume(Mono.defer(valueRetriever(key)))
                    .andWriteWith((k, signal) -> Mono.fromRunnable(() -> {
                        if (!signal.isOnError()) {
                            writeValue(k, signal.get());
                        }
                    }));
}

private void writeValue(ID key, T value) {
    if (value != null) {
        cache.put(key, value);
    }
}

cache.put(key, value); 是否认为是阻塞调用？应该在不同的调度器上发布吗？

Answer 1

是的。

如果你有机会，我建议你改用带有Lettuce 的 ReactiveRedisConnection。

例如，您可以按如下方式设置您的 bean：

@Bean
public ReactiveRedisConnectionFactory reactiveRedisConnectionFactory() {
    final RedisStandaloneConfiguration redisConfig 
    = new RedisStandaloneConfiguration(SpringBootApp.redisHost, SpringBootApp.redisPort);
    redisConfig.setPassword(SpringBootApp.redisPassword);
    redisConfig.setDatabase(0);
    final LettuceConnectionFactory connFac = new LettuceConnectionFactory(redisConfig);
    return connFac;
}

@Bean
public ReactiveRedisConnection reactiveRedisConnection(final ReactiveRedisConnectionFactory redisConnectionFactory) {
    return redisConnectionFactory.getReactiveConnection();
}

@Bean
public ReactiveRedisOperations<String, Object> redisOperations(ReactiveRedisConnectionFactory factory) {
    final JdkSerializationRedisSerializer serializer = new JdkSerializationRedisSerializer();
    final RedisSerializationContext.RedisSerializationContextBuilder<String, Object> builder = RedisSerializationContext
            .newSerializationContext(new StringRedisSerializer());
    final RedisSerializationContext<String, Object> context = builder.value(serializer).build();
    return new ReactiveRedisTemplate<>(factory, context);
}

然后您可以使用ReactiveRedisOperations 以非阻塞方式从缓存中存储/检索您的数据。你可以这样做：

   Mono.just("my_cache_key::" + uniqueId)
                .flatMap(key ->  reactiveRedisOps.opsForValue().get(key)
                        .cast(MyObj.class)
                        .switchIfEmpty(
                          getMyObjNonBlocking()
                           .flatMap(myObj -> ops.opsForValue().set(key, myObj, 
                                            Duration.ofMinutes(ttl))
                                            .thenReturn(myObj))
                        ));