java - 如何在 Flux 中使用 zipWith 迭代元组？

Question

我想遍历提供 referenceIds 的基本列表。然后在下一步中，我想进行多个服务调用并将结果聚合到一个元组中。

然后作为最后一步，我想遍历我所有的 referenceIds，并通过事件流直接返回每个生成的 JSONObject。

@GetMapping(value = "/test", produces = TEXT_EVENT_STREAM_VALUE)
public Flux<JSONObject> test(Integer pages) {
    pages = 1;
    return Flux.range(1, pages)
            .map(pageNumber -> Arrays.asList(1 * pageNumber, 2 * pageNumber, 3 * pageNumber)) //referenceIds for testing
            .flatMap(numbers -> Flux.fromIterable(numbers).zipWith(Mono.zip(a(numbers), b(numbers)))) //results based on reference ids
            .map(tuple -> {
                Integer number = tuple.getT1();
                Tuple2<List<String>, List<String>> lookup = tuple.getT2();

                JSONObject json = new JSONObject();
                json.put("number", number);
                json.put("someMore", <fromLookup>);
                return json;
            });
}

对于这个例子，方法a() 和b() 的返回类型无关紧要。重要的部分是:

如果我只是返回 Flux.fromIterable(numbers); 一切正常。但是当使用 .zipWith() 聚合时，我只收到数字列表的第一个元素。所有其他人都丢失了。为什么？

旁注:我需要使用 .zipWith() 来并行执行这些方法调用(一些运行时间较长的事务)。

Answer 1

All others are lost. Why?

来自 Flux 类文档:

Zip this {@link Flux} with another {@link Publisher} source, that is to say wait for both to emit one element and combine these elements once into a {@link Tuple2}. The operator will continue doing so until any of the sources completes.

您可以执行 a() 和 b() 调用，压缩结果，然后将 numbers 列表展开为 flux 并像这样添加结果:

   .flatMap(numbers -> Mono.zip(a(numbers), b(numbers))
                .flatMapMany(tuple -> Flux.fromIterable(numbers).map(i -> Tuples.of(i,tuple))))