在另一个助焊剂结束后执行平行助焊剂答案

【问题标题】：Execute a parallel flux after another flux has ended在另一个助焊剂结束后执行平行助焊剂
【发布时间】：2020-02-07 19:00:14
【问题描述】：

顺便说一句，我还在学习 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);

我需要等待使能通量的流程结束并处理其中的所有响应，原因是如果其中一个给我错误或否定响应，我将不会为阻止程序运行另一个 Parallel 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

如果enablers 的句柄没有错误，则与另一个Flux 一起转到.thenMany

【问题讨论】：

如果我满足您的要求，thenMany 运算符就是您要找的：projectreactor.io/docs/core/release/api/reactor/core/publisher/…
好的，我知道 thenMany 将帮助我在第一个通量结束时调用第二个通量，但是我如何处理来自第一个通量的数据以及如果条件失败返回给客户端一个错误和thenMany 内部的通量不被称为顺便说一句我忘了补充说我必须减少对 Mono 的响应我将更新代码

标签： java spring spring-webflux

【解决方案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

【讨论】：