如何将单声道流转换为通量

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);

    }

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

    @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
    }
]

我是反应式编程新手，将流和单声道结合起来，然后转换为流量的正确方法是什么？

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));
}

注意事项：

>

基本上，您希望getMatchesendpoint返回Flux

此外，不需要使用parallelStream（）。因为您已经在使用reactor，所以一切都将在reactor调度器上并发执行。