一旦创建了三个单声道，就并行执行它们，等待所有单声道完成，并以特定的顺序/逻辑收集结果

Question

我是SpringWebFlux的新手，所以请温柔一点。。。我很抱歉，如果我错过了一些明显的东西，但我试图寻找在线的例子，每次我结束与顺序调用。

我有这样的情况：

Mono<Response> mono1 = webclient1.post()...usual_stuffs....bodyToMono(Response.class);
Mono<Response> mono2 = webclient2.post()...usual_stuffs....bodyToMono(Response.class);
Mono<Response> mono3 = webclient3.post()...usual_stuffs....bodyToMono(Response.class);

响应是我的项目中的一个类，但是对于这个例子，我们可以将它们视为单个列表的简单容器。

我想：

• 并行执行它们（一旦我将它们分配给单数1/单数2/单数3，通过调用。
• 当一切都完成后，保存响应，以呼吸1，呼吸2，呼吸3
• 如果res1有结果（列表不是空的）返回res1...
• ...如果res2有结果（列表不是空的）返回res2...
• 返回res3（即使是空的）

我如何做到这一点？（因为我失败过多次）

我的第一次尝试是：

Mono<Response> mono1 = webclient1.post()...usual_stuffs....bodyToMono(Response.class).subscribeOn(Schedulers.parallel());
Mono<Response> mono2 = webclient2.post()...usual_stuffs....bodyToMono(Response.class).subscribeOn(Schedulers.parallel());
Mono<Response> mono3 = webclient3.post()...usual_stuffs....bodyToMono(Response.class).subscribeOn(Schedulers.parallel());
ParallelFlux.from(mono1, mono2, mono3).then().block(); // im not sure if this really execute them in parallel
Response resp1 = mono1.block();
Response resp2 = mono2.block();
Response resp3 = mono3.block();
if (resp1.isNotEmpty()) {
    return resp1;
}
if (resp2.isNotEmpty()) {
    return resp2;
}
return resp3;

这似乎不起作用，ParallelFlux.from（mon1，mon2，mon3）.然后（）. block（）真的并行运行这些monos吗？为什么我需要ParallelFlux？我不能在创建每个单声道时就说“在单独的线程上运行这个单声道”吗？每个. block（）实际上重做调用......就像它在执行单声道一样...为什么？

更新：

通过阅读评论，我将代码更改为：

Mono<Response> mono1 = webclient1.post()...usual_stuffs....bodyToMono(Response.class).subscribeOn(Schedulers.parallel());
Mono<Response> mono2 = webclient2.post()...usual_stuffs....bodyToMono(Response.class).subscribeOn(Schedulers.parallel());
Mono<Response> mono3 = webclient3.post()...usual_stuffs....bodyToMono(Response.class).subscribeOn(Schedulers.parallel());

Tuple3<Response, Response, Response> all = Mono.zip(mono1, mono2, mono3).block();

Response resp1 = all.getT1();
Response resp2 = all.getT2();
Response resp3 = all.getT3();

if (resp1.hasMessages()) {
    return resp1;
}
if (resp2.hasMessages()) {
    return resp2;
}
return resp3;

现在它似乎起作用了。我需要做其他事情还是我同意这个解决方案？我是否也应该更改Mono.zip（mon1，mon2，mon3）. block（）在Mono.zip（mon1，mon2，mon3）。P. S.我现在再次阅读文档，我认为我应该使用Schedulers.elastic（）代替Schedulers.parallel（）。




Answer 1

创建单声道不会自动执行它。您需要一个终端操作符，如订阅或块来触发执行（订阅不是终端操作符。除非您想将执行延迟到另一个线程池，否则您不需要它。默认情况下，它使用默认线程池）。如果您希望多个单声道并行运行，您可以使用zip操作符。

Mono<Response> mono1 = webclient1.post()...usual_stuffs....bodyToMono(Response.class).subscribeOn(Schedulers.parallel());
Mono<Response> mono2 = webclient2.post()...usual_stuffs....bodyToMono(Response.class).subscribeOn(Schedulers.parallel());
Mono<Response> mono3 = webclient3.post()...usual_stuffs....bodyToMono(Response.class).subscribeOn(Schedulers.parallel());

return Mono.zip(mono1, mono2, mono3)
        .map(t -> {
            if (t.getT1().isEmpty()) {
                if (t.getT2().isEmpty()) {
                    return t.getT3();
                } else {
                    return t.getT2();
                }
            } else {
                return t.getT1();
            }
        });

注意：调用它不会执行并给出结果。它会返回一个mono，您可以在其上调用subscribe（）以获得结果。