java - 有没有办法在 Java Spring 中将不可预测数量的 Monos 合并到列表中？

Question

我有一个 Mono > 方法，它接受 URN 列表的单个参数。

public Mono <List <MyObject>> loadMyObjectsFromUrnAsync (Iterable <String> urns) {...}

根据URN的数量，我必须发出许多异步HTTP请求，每个请求都由一个Mono表示。由于我希望程序能够在运行此方法并执行所有请求时移动到程序的其他部分，因此如何压缩各种 Monos 并返回单个 Mono 的结果？

... {
   List <Mono <MyObject>> myList = new Arraylist <Mono <MyObject>> ();
   while (urns.hasNext()) {
      // create a Mono HTTP request with the current iterated URN 
      myList.add(request); }
   return mono.zip(myList);
   }...

上面的代码显然不起作用，它只是希望让我的问题更容易理解。我在网上找不到任何有关如何压缩 Mono 列表而不必指定每个 Mono 的文档

Answer 1

上面的问题可能没有表达清楚，所以我就重新表述一下，然后描述我的解决方案。希望这对某人有帮助!

还有一个重要提示:该解决方案是在 Java Spring 中使用 WebClient 来执行对远程服务器的异步客户端请求。

Depending on the number of URNs, I have to make a number of asynchronous HTTP requests, each of which is represented by a Mono.

假设有一个函数，它接受 HTTP 端点的 ArrayList 并返回 MyObject 类型的 ArrayList。我希望此函数获取这些端点并使用 Monos 执行相应数量的 HTTP 请求，这些请求将填充该数量的 MyObject POJO(普通旧 Java 对象)，但是，由于 Monos 的数量将根据端点的数量而有所不同，因此我无法显式定义每个 Mono。我使用 Monos 来允许我并行或同时执行这些请求，而不是一个接一个地顺序执行，并且我希望在所有 Monos 完成并可以生成 MyObject 实例时收到通知。

As I want the program to be able to move onto other parts of the program while running this method and executing all the requests, how do I go about zipping the various Monos and returning the result for the singular merged stream when it completes?

组织一组不可预测的 Mono 的解决方案是存储 Monos 的 Iterable 并将其合并到 Flux 中，然后缓冲和阻止以生成 Java 对象!

(为了充分理解这个解决方案，了解 Monos 和 Fluxes 是什么很重要)

public Mono<MyObject> getOneObject(String endpoint) {
    return webClient.get()
        .uri(endpoint)
        .retrieve()
        .bodyToMono(MyObject.class)
}

上面的这个函数根据一个端点检索一个对象。现在，让我们根据给定的端点列表创建一个 Monos 的 Iterable。

public Iterable<Mono<MyObject>> getMultipleObjects(ArrayList<String> endpoints) {
    ArrayList<Mono<MyObject>> monos = new ArrayList<Mono<MyObject>> ();
    for (String endpoint: endpoints) {
         monos.add(getOneObject(endpoint));
    }
    return monos;
}

上面的这个函数返回一个 ArrayList(在 return 语句中转换为 Iterable)，它是通过迭代端点列表来填充的。现在，让我们将单声道的可迭代合并为一个单一的通量!

public Flux<MyObject> mergeMultipleMonos(Iterable<Mono<MyObject>> multipleMonos) {
    return Flux.merge(multipleMonos);
}

现在，您所要做的就是缓冲并阻止通量以生成对象的 ArrayList!

public ArrayList<MyObject> listOfObjects(Flux<MyObject> fluxOfObjects) {
    return fluxOfObjects.buffer().blockLast();
}

Flux.buffer() 会将所有传入值收集到一个列表中，而 Flux.blockLast() 将订阅此通量并无限期阻塞，直到上游发出其最后一个值的信号或完成。

这就是如何将不可预测数量的 Mono(在本例中代表异步客户端 HTTP Web 请求)合并到一个单一流(又名 Flux)中，该流将在每个请求返回时填充所有值!