Spring Reactive：等待所有 Mono 完成

本文介绍了在使用 Spring Reactive WebClient 调用多个外部 API 并返回 Mono 的场景下，如何等待所有 Mono 完成后再执行后续逻辑。通过将每个 Mono 的结果与服务名称关联，并使用 Flux 组合操作，可以实现响应式地处理所有 API 调用结果，并在全部完成后执行聚合操作。本文提供了详细的代码示例，包括错误处理和提高可读性的方法。

在使用 Spring Reactive 进行开发时，经常会遇到需要并发调用多个 API 接口，并且在所有接口都返回结果后才能进行下一步处理的场景。如果使用传统的阻塞式方法，会导致性能瓶颈。Spring Reactive 提供了强大的响应式编程模型，可以优雅地解决这个问题。

以下将介绍如何使用 Flux 和 Mono 来实现并发调用 API 并等待所有结果返回，最后执行聚合操作。

首先，假设我们有一个定时任务，需要从多个服务获取 Swagger 定义，并将它们添加到 definitionContext 中。每个服务都有一个名称和一个 Swagger URL。

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

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

    Mono 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 swaggerAllServicesData = getAllServicesApiSwagger(allServicesApi);
  if (swaggerAllServicesData.isPresent()) {
    String allApiContent = getJSON(swaggerAllServicesData.get());
    definitionContext.addServiceDefinition("All", allApiContent);
  }
}


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

上述代码存在的问题是，使用了 subscribe 方法，导致异步操作无法等待所有 Mono 完成。allServicesApi 列表在所有 API 调用完成之前就被处理，因此无法正确获取所有服务的 Swagger 定义。

解决方案：使用 Flux 组合操作

为了解决这个问题，可以使用 Flux 来组合所有的 Mono，并使用 collectList() 操作符来等待所有 Mono 完成。

首先，创建一个临时类，用于将 SwaggerServiceData 和服务名称关联起来：

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

然后，修改代码如下：

Flux.fromStream(swaggerProperties.getUrls().entrySet().stream())
    .flatMap((e) -> {
        Mono 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();
    })
    .collectList()
    .map(this::getAllServicesApiSwagger)
    .filter(Optional::isPresent)
    .map(Optional::get)
    .subscribe(e -> {
        String allApiContent = getJSON(e);
        definitionContext.addServiceDefinition("All", allApiContent);
    });

代码解释：

迷你天猫商城

迷你天猫商城是一个基于Spring Boot的综合性B2C电商平台，需求设计主要参考天猫商城的购物流程：用户从注册开始，到完成登录，浏览商品，加入购物车，进行下单，确认收货，评价等一系列操作。 作为迷你天猫商城的核心组成部分之一，天猫数据管理后台包含商品管理，订单管理，类别管理，用户管理和交易额统计等模块，实现了对整个商城的一站式管理和维护。所有页面均兼容IE10及以上现代浏览器。部署方式1、项目

下载

1. Flux.fromStream(swaggerProperties.getUrls().entrySet().stream()): 将 swaggerProperties.getUrls() 中的所有条目转换为一个 Flux 流。
2. flatMap((e) -> ...): 对流中的每个条目（服务名称和服务 URL）执行异步操作。flatMap 允许将每个条目转换为一个 Mono，并将所有 Mono 合并到一个 Flux 中。
3. swaggerDefinitionForAPI.map(swaggerServiceData -> new SwaggerService(swaggerServiceData, e.getKey())): 获取 SwaggerServiceData 后，将其与服务名称关联，创建一个 SwaggerService 对象。
4. filter(SwaggerService::hasData): 过滤掉 SwaggerServiceData 为 null 的情况。
5. map(swaggerService -> ...): 将 SwaggerServiceData 转换为 JSON 字符串，并添加到 definitionContext 中。
6. collectList(): 将所有 SwaggerServiceData 收集到一个 List 中，并返回一个包含该 List 的 Mono。 注意：collectList() 会等待所有上游的 Mono 完成。
7. map(this::getAllServicesApiSwagger): 对收集到的 SwaggerServiceData 列表执行聚合操作。
8. filter(Optional::isPresent): 过滤掉 Optional 为空的情况。
9. map(Optional::get): 从 Optional 中获取 SwaggerServiceData。
10. subscribe(e -> ...): 订阅最终的 Mono，并将聚合后的 SwaggerServiceData 添加到 definitionContext 中。

错误处理

上述代码没有处理 SwaggerServiceData 为 null 时的错误日志记录。可以使用 flatMap 和 Mono.empty() 来添加错误处理：

Flux.fromStream(swaggerProperties.getUrls().entrySet().stream())
    .flatMap((e) -> {
        Mono 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);
    });

代码解释：

• 在 swaggerDefinitionForAPI.flatMap 中，如果 swaggerServiceData 为 null，则记录错误日志并返回 Mono.empty()。Mono.empty() 会导致该条目被过滤掉，不会影响后续的聚合操作。

提高代码可读性

为了提高代码的可读性，可以将 lambda 表达式封装成单独的方法：

private Mono processSwaggerService(Map.Entry entry) {
    Mono swaggerDefinitionForAPI = getSwaggerDefinitionForAPI(entry.getKey(),
            entry.getValue());
    return swaggerDefinitionForAPI
            .flatMap(swaggerServiceData -> {
                if(swaggerServiceData != null) {
                    return Mono.just(new SwaggerService(swaggerServiceData, entry.getKey()));
                } else {
                    log.error("Skipping service id : {} Error : Could not get Swagger definition from API ",
                            entry.getKey());
                    return Mono.empty();
                }
            });
}

private SwaggerServiceData processSwaggerData(SwaggerService swaggerService) {
    String content = getJSON(swaggerService.swaggerServiceData());
    definitionContext.addServiceDefinition(swaggerService.serviceName(), content);
    return swaggerService.swaggerServiceData();
}


Flux.fromStream(swaggerProperties.getUrls().entrySet().stream())
    .flatMap(this::processSwaggerService)
    .map(this::processSwaggerData)
    .collectList()
    .map(this::getAllServicesApiSwagger)
    .filter(Optional::isPresent)
    .map(Optional::get)
    .subscribe(e -> {
        String allApiContent = getJSON(e);
        definitionContext.addServiceDefinition("All", allApiContent);
    });

注意事项：

• 确保 definitionContext 是线程安全的，因为多个 Mono 可能会并发地向其添加数据。
• 根据实际情况调整并发度，避免对外部 API 造成过大的压力。可以使用 flatMap 的 concurrency 参数来限制并发度。
• 在生产环境中，需要添加更完善的错误处理机制，例如重试、熔断等。

总结：

使用 Flux 和 Mono 提供的组合操作，可以方便地实现并发调用 API 并等待所有结果返回。collectList() 操作符是关键，它可以确保在所有上游的 Mono 完成后才执行后续操作。通过添加错误处理和提高代码可读性，可以使代码更加健壮和易于维护。