彈簧反應。如何等待所有單聲道完成？

我有以下代碼，我通過 webclient 呼叫外部 API 并回傳 Mono。當我收到資料時，我需要執行一些邏輯。畢竟，請求被處理，為所有收集的資料執行一個邏輯。我可以收集所有的 Mono 并將它們放到 Flux 中，然后在最后執行一些邏輯。但是我有 serviceName 檔案，它只能在回圈中訪問，所以我需要在回圈中執行單聲道邏輯，在這里我被卡住了，不知道如何等待所有資料完成并以反應方式完成。

@Scheduled(fixedDelay = 50000)
public void refreshSwaggerConfigurations() {
  log.debug("Starting Service Definition Context refresh");
  List<SwaggerServiceData> allServicesApi = new ArrayList<>();

  swaggerProperties.getUrls().forEach((serviceName, serviceSwaggerUrl) -> {
    log.debug("Attempting service definition refresh for Service : {} ", serviceName);

    Mono<SwaggerServiceData> swaggerData = getSwaggerDefinitionForAPI(serviceName,
        serviceSwaggerUrl);

    swaggerData.subscribe(swaggerServiceData -> {
      if (swaggerServiceData != null) {
        allServicesApi.add(swaggerServiceData);
        String content = getJSON(swaggerServiceData);
        definitionContext.addServiceDefinition(serviceName, content);
      } else {
        log.error("Skipping service id : {} Error : Could not get Swagger definition from API ",
            serviceName);
      }
    });
  });
  //I need to wait here for all monos to complete and after that proceed for All gathered data...
  //Now it's empty And I know why, just don't know how to make it.
  Optional<SwaggerServiceData> swaggerAllServicesData = getAllServicesApiSwagger(allServicesApi);
  if (swaggerAllServicesData.isPresent()) {
    String allApiContent = getJSON(swaggerAllServicesData.get());
    definitionContext.addServiceDefinition("All", allApiContent);
  }
}


private Mono<SwaggerServiceData> getSwaggerDefinitionForAPI(String serviceName, String url) {
  log.debug("Accessing the SwaggerDefinition JSON for Service : {} : URL : {} ", serviceName,
      url);
    Mono<SwaggerServiceData> swaggerServiceDataMono = webClient.get()
        .uri(url)
        .exchangeToMono(clientResponse -> clientResponse.bodyToMono(SwaggerServiceData.class));
    return swaggerServiceDataMono;
}

uj5u.com熱心網友回復：

我會添加一個臨時類來分組資料和服務名稱：

record SwaggerService(SwaggerServiceData swaggerServiceData, String serviceName) {
        boolean hasData() {
            return swaggerServiceData != null;
        }
 }

然后改變你的管道：

Flux.fromStream(swaggerProperties.getUrls().entrySet().stream())
                .flatMap((e) -> {
                    Mono<SwaggerServiceData> swaggerDefinitionForAPI = getSwaggerDefinitionForAPI(e.getKey(),
                            e.getValue());
                    return swaggerDefinitionForAPI.map(swaggerServiceData -> new SwaggerService(swaggerServiceData, e.getKey()));
                })
                .filter(SwaggerService::hasData)
                .map(swaggerService -> {
                    String content = getJSON(swaggerService.swaggerServiceData());
                    definitionContext.addServiceDefinition(swaggerService.serviceName(), content);
                    return swaggerService.swaggerServiceData();
                })
                // here we will collect all datas and they will be emmited as single Mono with list of SwaggerServiceData
                .collectList()
                .map(this::getAllServicesApiSwagger)
                .filter(Optional::isPresent)
                .map(Optional::get)
                .subscribe(e -> {
                    String allApiContent = getJSON(e);
                    definitionContext.addServiceDefinition("All", allApiContent);
                });

當 SwaggerServiceData 為 null 時，這不會處理日志記錄錯誤，但您可以根據需要進一步更改它。我也認為這DefinitionContext是執行緒安全的。

錯誤記錄的解決方案（使用flatMapand Mono.empty()）：

        Flux.fromStream(swaggerProperties.getUrls().entrySet().stream())
                .flatMap((e) -> {
                    Mono<SwaggerServiceData> swaggerDefinitionForAPI = getSwaggerDefinitionForAPI(e.getKey(),
                            e.getValue());
                    return swaggerDefinitionForAPI
                            .flatMap(swaggerServiceData -> {
                                if(swaggerServiceData != null) {
                                    return Mono.just(new SwaggerService(swaggerServiceData, e.getKey()));
                                } else {
                                    log.error("Skipping service id : {} Error : Could not get Swagger definition from API ",
                                            e.getKey());
                                    return Mono.empty();
                                }
                            });
                })
                .map(swaggerService -> {
                    String content = getJSON(swaggerService.swaggerServiceData());
                    definitionContext.addServiceDefinition(swaggerService.serviceName(), content);
                    return swaggerService.swaggerServiceData();
                }).collectList()
                .map(this::getAllServicesApiSwagger)
                .filter(Optional::isPresent)
                .map(Optional::get)
                .subscribe(e -> {
                    String allApiContent = getJSON(e);
                    definitionContext.addServiceDefinition("All", allApiContent);
                });

您還可以將這些 lambdas 包裝成一些有意義的方法以提高可讀性。

uj5u.com熱心網友回復：

我有一個簡單的 Java 程式，它使用 Spring WebClient 發送多個請求。每個都回傳一個單聲道，我正在使用 response.subscribe() 來檢查結果。

但是，除非我添加一個長的 Thread.sleep()，否則我的執行主執行緒會在處理所有請求之前完成。

使用 CompletableFutures，您可以使用： CompletableFuture.allOf(futures).join();

標籤：爪哇春天单核细胞增多症spring-webflux通量

上一篇：我有一個帶有JWT令牌的SpringBoot應用程式，我給出了一個錯誤

下一篇：使用React和Typescript隱藏徽章編號為0