如何将 Mono 流转换为 Flux

Question

我有一个尝试使用 WebClient 返回 Mono 的方法

    @GetMapping("getMatch")
    public Mono<Object> getMatch(@RequestParam Long matchId) {
        return WebClient.create(OpenDotaConstant.BASE_URL).get()
                .uri("/matches/{matchId}", matchId)
                .accept(MediaType.APPLICATION_JSON)
                .retrieve()
                .bodyToMono(Object.class);

    }

它可以返回我期望的结果。然后我尝试创建另一种方法来支持 List 作为参数

    @GetMapping("getMatches")
    public Flux<Object> getMatches(@RequestParam String matchesId) {
        List<Long> matchesList = JSON.parseArray(matchesId, Long.class);

        return Flux.fromStream(matchesList.parallelStream().map(this::getMatch));
    }

但这次返回了一个奇怪的结果。

[
    {
        "scanAvailable": true
    },
    {
        "scanAvailable": true
    }
]

我是反应式编程的新手，结合 Stream 和 Mono，然后转换为 Flux 的正确方法是什么？

Answer 1

也许，您需要的是以下内容：

@GetMapping("getMatches")
public Flux<Object> getMatches(@RequestParam String matchesId) {
    List<Long> matchesList = JSON.parseArray(matchesId, Long.class);
    return Flux.fromStream(matchesList.stream())
               .flatMap(this::getMatch);
}

代替：

@GetMapping("getMatches")
public Flux<Object> getMatches(@RequestParam String matchesId) {
    List<Long> matchesList = JSON.parseArray(matchesId, Long.class);
    return Flux.fromStream(matchesList.parallelStream().map(this::getMatch));
}

笔记：

• 基本上，您期望getMatches端点返回Flux<Object>。然而，正如它所写的 - 它实际上返回Flux<Mono<Object>>，因此你会看到奇怪的输出。为了获得Flux<Object>，我建议首先创建Flux<Long>匹配 ID，然后flatMap调用结果getMatch（返回Mono<Object>），最后给出Flux<Object>。

• 另外，没有必要使用parallelStream(). 因为您已经在使用reactor，所以所有事情都将在reactor调度程序上同时执行。