Spring Reactive如何等待所有Mono执行完成？

核心问题分析 #

你当前代码里的subscribe()是异步非阻塞调用，循环会直接跳过所有异步任务执行后续代码，导致allServicesApi为空。需要用响应式API组合所有异步任务，等待全部完成后再执行汇总逻辑，同时保留每个任务中的serviceName用于单个服务的处理。

修改后的代码 #

@Scheduled(fixedDelay = 50000)
public void refreshSwaggerConfigurations() {
  log.debug("Starting Service Definition Context refresh");

  // 将每个服务的异步任务组合为Flux
  Flux<SwaggerServiceData> serviceDataFlux = Flux.fromIterable(swaggerProperties.getUrls().entrySet())
      .flatMap(entry -> {
        String serviceName = entry.getKey();
        String serviceSwaggerUrl = entry.getValue();
        log.debug("Attempting service definition refresh for Service : {} ", serviceName);

        return getSwaggerDefinitionForAPI(serviceName, serviceSwaggerUrl)
            // 处理HTTP请求异常，避免单个失败导致整个流程终止
            .onErrorResume(e -> {
              log.error("Failed to fetch Swagger definition for service: {}", serviceName, e);
              return Mono.empty();
            })
            // 处理单个服务的业务逻辑（此时可正常访问serviceName）
            .doOnNext(swaggerServiceData -> {
              if (swaggerServiceData != null) {
                String content = getJSON(swaggerServiceData);
                definitionContext.addServiceDefinition(serviceName, content);
              } else {
                log.error("Skipping service id : {} Error : Could not get Swagger definition from API ", serviceName);
              }
            })
            // 过滤无效数据，只保留有效的Swagger服务数据
            .filter(Objects::nonNull);
      });

  // 等待所有异步任务完成，收集有效数据
  List<SwaggerServiceData> allServicesApi = serviceDataFlux
      .collectList()
      // 阻塞等待所有任务执行完毕（@Scheduled为同步方法，必须确保任务完成后再执行后续逻辑）
      .block();

  // 执行汇总逻辑
  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);
  return webClient.get()
      .uri(url)
      .exchangeToMono(clientResponse -> clientResponse.bodyToMono(SwaggerServiceData.class));
}

关键修改点说明 #

• 用Flux组合异步任务：通过Flux.fromIterable遍历所有服务配置，将每个服务的API调用转换为响应式流中的一个任务。
• 保留serviceName上下文：在flatMap的闭包中直接使用serviceName，确保单个服务处理逻辑能访问到该字段。
• 异常处理：添加onErrorResume捕获HTTP请求异常，避免单个服务的调用失败导致整个刷新流程终止。
• 等待所有任务完成：使用collectList().block()阻塞当前线程，直到所有异步API调用和单个服务处理逻辑执行完毕，确保allServicesApi包含所有有效数据。
• 过滤无效数据：通过filter(Objects::nonNull)剔除空的服务数据，避免无效数据进入汇总逻辑。

内容的提问来源于stack exchange，提问作者Daimon Cool