将顺序 Monos 转换为 Flux答案

【问题标题】：Convert sequential Monos to Flux将顺序 Monos 转换为 Flux
【发布时间】：2021-07-11 04:51:36
【问题描述】：

我有一个 Web 服务，我想在其中检索直到根节点的树的元素。我有一个 Webflux 接口，它在每次调用时返回一个 Mono：

public interface WebService {
    Mono<Node> fetchNode(String nodeId);
}

public class Node {
    public String id;
    public String parentId; // null, if parent node
}

假设有一棵树

    1
  2   3
  4   5

我想创建以下方法：

public interface ParentNodeResolver {

    Flux<Node> getNodeChain(String nodeId);
}

这将在getNodeChain(5) 上为我提供一个 Flux，其中包含 5、3 和 1 的节点，然后完成。

不幸的是，我不太明白如何按顺序组合 Monos，但又不会阻塞它们。使用Flux.generate()，我想我需要阻止每个单声道以检查它是否有下一个元素。我发现的其他方法似乎只结合了固定数量的 Monos，但不是以这种递归方式。

这是一个示例代码，它可以模拟网络请求，但会有一些延迟。

public class MonoChaining {
    ExecutorService executorService = Executors.newFixedThreadPool(5);

    @Test
    void name() {
        var nodeChain = generateNodeChainFlux("5")
                .collectList()
                .block();
        assertThat(nodeChain).isNotEmpty();
    }

    private Flux<Node> generateNodeChainFlux(String nodeId) {
        //TODO
        return Flux.empty();


    }

    public Mono<Node> getSingleNode(String nodeId) {
        var future =
                CompletableFuture.supplyAsync(() -> {
                    try {
                        Thread.sleep(2000); // Simulate delay
                        if ("5".equals(nodeId)) {
                            return new Node("5", "3");
                        } else if ("3".equals(nodeId)) {
                            return new Node("3", "1");
                        } else if ("1".equals(nodeId)) {
                            return new Node("1", null);
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return null;
                }, executorService);

        return Mono.fromFuture(future);
    }


    public static class Node {
        public String id;
        public String parentId;

        public Node(String id, String parentId) {
            this.id = id;
            this.parentId = parentId;
        }
    }
}

有没有办法找回这个？谢谢！

【问题讨论】：

你不阻塞，你获取一个，然后使用flatMap读取值，然后在flatMap中采取行动。顺便说一句，您可以使用 Mono.fromCallable 代替 CompletableFuture.fromAsync。

标签： java spring-webflux project-reactor

【解决方案1】：

使用递归flatMap 获取父节点和concat 将当前节点附加到结果flux 可能有效。试试下面的代码：

public Flux<Node> getNodeChain(String nodeId) {
    return fetchNode(nodeId).flatMapMany(node -> {
        if (node.parent != null) {
            Flux<Node> nodeChain = getNodeChain(node.parent);
            return Flux.concat(Flux.just(node), nodeChain);
        }
        return Flux.just(node);
    });
}

这里我使用flatMapMany 将Mono 转换为Flux。

【讨论】：

【解决方案2】：

您要查找的运算符是Mono#expand。它用于递归扩展序列。阅读更多here。

在你的情况下：

private Flux<Node> generateNodeChainFlux(String nodeId) {
        return getSingleNode(nodeId).expand(node -> getSingleNode(node.parentId));
    }

【讨论】：