使用反应性生菜流水线Redis命令

Question

我使用spring boot webflux以非阻塞方式连接和查询Redis。我已经用LettuceConnectionFactory配置了reactivedistemplate。spring文档指出，将管道与reactivedistemplate一起使用的唯一方法是使用execute(

所以我的问题是，在使用Spring ReactiveRedisTemplate时，是否可以对命令进行管道连接？

我还注意到使用reactivedistemplate。使用具有多个Redis命令的RedisCallback执行，执行速度比单独调用命令慢。

带有ReactiveRedisTemboard的管道示例代码：

reactiveRedisTemplate.execute(connection -> keys.flatMap(key -> 
                                connection.hashCommands()
                                .hGetAll(ByteBuffer.wrap(key.getBytes()))))
                    .map(Map.Entry::getValue)
                    .map(ByteUtils::getBytes)
                    .map(b -> {
                        try {
                        return mapper.readValue(b, Value.class);
                        } catch (IOException e1) {
                        return null;
                        }
                    })
                    .collectList();

无管道代码：

keys.flatMap(key -> reactiveRedisTemplate.opsForHash().entries(key))
            .map(Map.Entry::getValue)
            .cast(Value.class)
            .collectList();

谢谢

Answer 1

我认为不可能使用RedisReactiveTemplate或莴苣的反应api。实际上，当你构建你的反应链时，它的某些部分将被延迟评估。

getAsyncValue(a).flatMap(value -> doSomething(value)).subscribe()

例如，在此示例中，仅当getAsyncValue返回值时才会触发。

现在，如果我们获取您的RedisCallback示例并假设我们在连接对象中有一个flushAll方法。你在哪里/什么时候调用它？

tpl.execute(connection -> {
                    Flux<Map.Entry<ByteBuffer, ByteBuffer>> results = keys.flatMap(key ->
                            connection.hashCommands()
                                    .hGetAll(ByteBuffer.wrap(key.getBytes())));
                    connection.fluxAll();
                    return results;
                })

像这样，命令将被刷新到服务器，因为不会触发hashCommands。

现在让我们看看我们拥有的所有信号回调：

• doOnNext
• 错误
• 邓肯塞尔
• doFirst
• doOnSubscribe
• doOnRequest
• doOnTerminate
• DOAFTERMINATE
• 最后，Do
• doOnComplete公司

doOnError或doOnCancel帮不了我们。但我们可以考虑使用doFinally、doOnTerminate、doafterminate：

tpl.execute(connection -> keys.flatMap(key -> connection.hashCommands()
                                        .hGetAll(ByteBuffer.wrap(key.getBytes())))
                                        .doFinally(s -> connection.flushAll()))

但htGetAll在命令刷新到服务器之前不会完成，所以doFinnaly不会被调用，所以我们无法刷新。。。。

我能想到的唯一解决方法是直接使用莴苣的异步api。文档中有一个关于如何执行此操作的示例。

您的代码可能看起来像（未测试）：

// client is a RedisClient from lettuce

StatefulRedisConnection<String, String> connection = client.connect();
RedisAsyncCommands<String, String> command = connection.async();
command.setAutoFlushCommands(false);
keys.map(command::hgetall)
                .collectList()
                .doOnNext(f -> command.flushCommands())
                .flatMapMany(f -> Flux.fromIterable(f).flatMap(Mono::fromCompletionStage))