class RedisReactiveCache<E> extends AbstractReactiveCache<E> {

    private ReactiveRedisOperations<Object, Object> operations;

    private String redisKey;

    private ReactiveCache<E> localCache;

    private String topicName;

    public RedisReactiveCache(String redisKey, ReactiveRedisOperations<Object, Object> operations, ReactiveCache<E> localCache) {
        this.operations = operations;
        this.localCache = localCache;
        this.redisKey = redisKey;
        operations
                .listenToChannel(topicName = ("_cache_changed:" + redisKey))
                .map(ReactiveSubscription.Message::getMessage)
                .cast(String.class)
                .subscribe(s -> {
                    if (s.equals("___all")) {
                        localCache.clear().subscribe();
                        return;
                    }
                    //清空本地缓存
                    localCache.evict(s).subscribe();
                });
    }

    @Override
    protected Mono<Object> getNow(Object key) {
        return (Mono) localCache.getMono(key, () -> (Mono) operations.opsForHash().get(redisKey, key));
    }

    @Override
    public Mono<Void> putNow(Object key, Object value) {<FILL_FUNCTION_BODY>}


    protected <T> Mono<T> handleError(Throwable error) {
        log.error(error.getMessage(), error);
        return Mono.empty();
    }

    @Override
    public Mono<Void> evictAll(Iterable<?> key) {
        return operations
                .opsForHash()
                .remove(redisKey, StreamSupport.stream(key.spliterator(), false).toArray())
                .then(localCache.evictAll(key))
                .flatMap(nil -> Flux
                        .fromIterable(key)
                        .flatMap(k -> operations.convertAndSend(topicName, key))
                        .then())
                .onErrorResume(err -> this.handleError(err));
    }

    @Override
    public Flux<E> getAll(Object... keys) {
        if (keys == null || keys.length == 0) {
            return operations
                    .opsForHash()
                    .values(redisKey)
                    .map(r -> (E) r);
        }
        return operations
                .opsForHash()
                .multiGet(redisKey, Arrays.asList(keys))
                .flatMapIterable(Function.identity())
                .map(r -> (E) r)
                .onErrorResume(err -> this.handleError(err));
    }


    @Override
    public Mono<Void> evict(Object key) {
        return operations
                .opsForHash()
                .remove(redisKey, key)
                .then(localCache.evict(key))
                .then(operations.convertAndSend(topicName, key))
                .onErrorResume(err -> this.handleError(err))
                .then();
    }

    @Override
    public Mono<Void> clear() {
        return operations
                .opsForHash()
                .delete(redisKey)
                .then(localCache.clear())
                .then(operations.convertAndSend(topicName, "___all"))
                .onErrorResume(err -> this.handleError(err))
                .then();
    }
}

        return operations
                .opsForHash()
                .put(redisKey, key, value)
                .then(localCache.evict(key))
                .then(operations.convertAndSend(topicName, key))
                .then();
    
<methods>public non-sealed void <init>() ,public abstract Mono<java.lang.Void> clear() ,public abstract Mono<java.lang.Void> evict(java.lang.Object) ,public abstract Mono<java.lang.Void> evictAll(Iterable<?>) ,public transient Flux<E> getAll(java.lang.Object[]) ,public final Flux<E> getFlux(java.lang.Object) ,public final Flux<E> getFlux(java.lang.Object, Supplier<Flux<E>>) ,public final Mono<E> getMono(java.lang.Object) ,public final Mono<E> getMono(java.lang.Object, Supplier<Mono<E>>) ,public final Mono<java.lang.Void> put(java.lang.Object, Publisher<E>) ,public abstract Mono<java.lang.Void> putNow(java.lang.Object, java.lang.Object) <variables>private final Map<java.lang.Object,org.hswebframework.web.cache.supports.AbstractReactiveCache.CacheLoader> cacheLoading,static Sinks.EmitFailureHandler emitFailureHandler