等待所有 Mono 响应从下游 API 返回的正确方法是什么

Question

我对 Mono 和 Flux 还很陌生。我正在尝试加入几个下游 API 响应。这是一个传统的阻塞应用程序。我不想收集 Mono 列表，我想要从下游 API 返回的有效负载列表，我从 Mono 获取该列表。然而，返回到控制器的“结果”有时只有一些或没有下游 API 响应。这样做的正确方法是什么？我读过几篇文章如何迭代 Flux 并与 Mono状态混合

你不应该在网络应用程序的任何地方调用 subscribe 。如果这绑定到 HTTP 请求，那么您基本上会触发反应式管道，而无法保证资源或完成情况。调用 subscribe 会触发管道，但不会等到它完成

我应该使用 CompletableFuture 吗？

在我的服务中我尝试过

var result = new ArrayList<List<>>();
List<Mono<X>> monoList = apiCall();
Flux.fromIterable(monoList)
            .flatMap(m -> m.doOnSuccess(
                        x -> {  
                            result.add(x.getData());
                        }
        )).subscribe();

我还在控制器中尝试了以下操作，但该方法返回而不等待订阅完成

var result = new ArrayList<List<X>>();
        Flux.concat(
                this.service.callApis(result, ...)
        ).subscribe();
        return result;

在我的服务中

public Mono<Void> callApis(List<List<x>> result, ..) {
...
return Flux.fromIterable(monoList)
                .flatMap(m -> m.doOnSuccess(
                        x -> {  
                            result.add(x.getData()...);
                        }
                )).then();

Answer 1

Project Reactor 文档（非常好）有一个名为“我需要哪个运算符？”的部分。。您需要从 API 调用中创建一个 Flux ，组合结果，然后返回到同步世界。

就您而言，您的所有下游服务看起来都具有相同的 API，因此它们都返回相同的类型，并且这些响应在您的应用程序中出现的顺序并不重要。另外，我假设apiCall()返回一个List<Mono<Response>>. 你可能想要类似的东西

Flux.fromIterable(apiCall()) // Flux<Mono<Response>>
    .flatMap(mono -> mono) // Flux<Response>
    .map(response -> response.getData()) // Flux<List<X>>
    .collectList() // Mono<List<List<X>>>
    .block(); // List<List<X>>

该fromIterable(...).flatMap(x->x)构造只是将您的转换List<Mono<R>>为Flux<R>.

map()用于提取响应的数据部分。

collectList()创建一个Mono等待Flux完成的结果，并给出包含所有数据列表的单个结果。

block()订阅Mono前一个运算符返回的 s，并阻塞直到完成，这将（在本例中）是当Mono由 返回的所有 sapiCall()完成时。

这里有许多可能的替代方案，最合适的方案取决于您的具体用例。