我是响应式(Reactive)世界的新手，正在努力理解如何完成任务。我正在处理一个遗留项目，我必须在其中实现一个接口(interface)，该接口(interface)具有许多方法来从 redis 查询各种对象。有时查询就像按 ID 查询哈希一样简单，因此只需调用 redis 一次即可获取哈希。其他时候，我可能需要先根据一些参数从 Redis 集中查找 ID，然后使用结果 ID 获取哈希值。我在 Spring Boot 应用程序中使用 Reactor 3.1.0.M3 和 Lettuce 5.0.0.RC1。

这两个示例方法的现有代码如下所示:

    public <T extends CatalogInfo> T get(String id, Class<T> clazz) {
        String result = (String)repository.getRedisHashRepository().getHashById(CatalogUtils.root(clazz).getSimpleName(), id);
        if (null != result) {
            return serializer.fromData(result, clazz);
        }
        return null;

    }

    public <T extends CatalogInfo> T get(String attName, String attValue, Class<T> clazz) {

        String attKey = CatalogUtils.buildKey(CatalogUtils.root(clazz), attName, attValue);

        String id = CatalogUtils.getIdFromSet(repository.getRedisSetRepository().getSetMembers(attKey));
        if (id == null) {
            return null;
        }
        return get(id, clazz);
    }

其中有一些实用函数可以帮助我从类名构建要用于 redis 的键，并确保存储在 redis 集中的 ID 是单个值。如您所见，在第二个 get 方法中，我使用集合中的结果调用第一个 get 方法。

使用 Lettuce/Reactor 实现第一种方法非常简单:

public <T extends CatalogInfo> Mono<String> getReactive(String id, Class<T> clazz, Publisher<String>... publisher) {
    Mono<String> mono = Mono.just(id).flatMap(new Function<String, Mono<String>>() {
        @Override
        public Mono<String> apply(String id) {
            return hashCommands.hget(CatalogUtils.root(clazz).getSimpleName(), id);
        }
    });
    return mono;
}

此时，我可以调用 mono.block() 来获取结果值。通过将两个 flatMaps/函数链接在一起，我可以很容易地获得第二个 get 方法的功能:

public <T extends CatalogInfo> Flux<String> getIdFromSetReactive(String attName, String attValue, Class<T> clazz) {
    String attKey = CatalogUtils.buildKey(CatalogUtils.root(clazz), attName, attValue);

    Flux<String> flux = Flux.just(attKey).flatMap(new Function<String, Flux<String>>() {
        @Override
        public Flux<String> apply(String attKey) {
            return commands.smembers(attKey);
        }
    }).flatMap(new Function<String, Publisher<String>>() {
        @Override
        public Publisher<String> apply(String id) {
            return hashCommands.hget(CatalogUtils.root(clazz).getSimpleName(), id);
        }
    });
    return flux;
} 

我有许多不同类型的方法，可能需要多达 8 次调用 redis 才能完成。在我的原始代码中，我可以重复使用每个方法并从另一个方法调用一个方法，但我不知道如何使用 react 器来做到这一点。

我希望能够调用一个构建 Flux 的方法以从 redis 集合中获取 ID(我们称之为 fluxA)，然后调用另一个构建 Flux 的方法以根据 ID(fluxB)查询 redis 哈希), 等

我认为我可能需要将我可能需要的每个函数定义为成员变量，如下所示:

private Function<String, Flux<String>> getIdFromSetFunction = new Function<String, Flux<String>>() {
        @Override
        public Flux<String> apply(String attKey) {
            return commands.smembers(attKey);
        }
    }; 

然后打电话

return Flux.just(attKey).flatMap(getIdFromSetFunction).flatMap(getHash);

唯一的问题是在这些函数中执行的代码需要当前在我的方法调用中可用的类信息。但我不确定这是正确的方法。

如有任何建议，我们将不胜感激!

最佳答案

从概念上讲，您不是在“组合单一对象”，而是“为每个步骤创建一个新对象”。

Mono<String> a = Mono.just("something");
Mono<String> b = a.flatMap( s -> goDoSomethingElseThatReturnsAMono(s));

String result = b.block();

您可以随心所欲地保持这样的链接。 (或者如果您有多个要单独处理的数据项，请使用 Mono.flatMapMany 进入 Flux 世界)。

在 block 被调用之前什么都不会发生，因为它订阅了 b 并阻塞当前线程直到结果可用。

关于redis - 将多个响应式(Reactive) Publisher/Flux/Mono 对象组合成阻塞请求，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/45381480/