在另一个通量结束后执行并行通量

Question

顺便说一句，我还在学习WebLUX；我不知道这是不是可能的，或者我有错误的方法，但考虑到这种平行的通量。

Flux<String> enablers = Flux.fromIterable(enablersList)
                .parallel()
                .runOn(Schedulers.elastic())
                .flatMap(element -> service.getAMono(string, entity, element))
                .sequential();

调用具有webclient请求的方法（service.getamono）

webClient.post()
              .uri(url)
              .headers(headers -> headers.addAll(httpHeaders))
              .body(BodyInserters.fromObject(request))
              .retrieve()
              .bodyToMono(entity2.class);

我需要等待enablers Flux的流结束，并处理它内部的所有响应，原因是如果其中一个给我错误或否定的响应，我不会运行这个其他并行Flux的阻塞器

Flux<String> blockers = Flux.fromIterable(blockersList)
                .parallel()
                .runOn(Schedulers.elastic())
                .flatMap(element -> service.callAMono(string, entity, element))
                .sequential();

我想过“zip”方法，但这一个合并了两个响应，并不是我想要的，如果任何人可以帮助我这样做。

更新

enablers. //handle enablers response and if error return a custom Mono<response> with .reduce

如果使能器的句柄没有错误，则继续到.thenmany和其他flux

Answer 1

我在第一个flux中找到了条件any来实现它的方法，如下所示

Flux.fromIterable(enablersList)
                .parallel()
                .runOn(Schedulers.elastic())
                .flatMap(element -> service.getAMono(string, entity, element))
                .sequential()
                .any(element -> *stuff here)//condition
                .flatMap(condition->{
                        if(condition.equals(Boolean.FALSE)){
                           return Flux.fromIterable(blockersList)
                                                   .parallel()
                                                   .runOn(Schedulers.elastic())
                                                   .flatMap(element -> service.callAMono(string, entity, element))
                                                   .sequential()
                                                   .reduce(**stuff here)// handle noError response and return;
                          }
                          return Mono.just(**stuff here);//handle error response and return
                 });

如果有其他的方法来做这件事，我会很高兴你把它贴在这里谢谢，:D