在 spring Reactive 编程中，经常会遇到需要并发调用多个外部 API，并将所有 API 返回的结果进行聚合处理的场景。例如，你需要从多个服务获取 Swagger 定义，并将它们合并成一个总的 Swagger 定义。在这种情况下，你需要确保所有 API 调用都完成后，才能执行后续的聚合逻辑。
直接使用 Mono.zip 或 Flux.merge 等操作符虽然可以实现并发调用，但无法方便地获取每个 API 调用的服务名，也难以处理 API 调用失败的情况。本文将介绍一种使用 Flux 和 collectList 操作符来实现并发调用和结果聚合的方法，并提供处理错误日志的示例代码。
使用 Flux 和 collectList 实现并发调用和结果聚合
首先，我们需要创建一个临时类，用于存储 API 返回的数据和服务名：
record SwaggerService(SwaggerServiceData swaggerServiceData, String serviceName) { boolean hasData() { return swaggerServiceData != NULL; } }
然后，我们可以使用 Flux.fromStream 将服务名和 URL 的映射转换为 Flux，并使用 flatmap 操作符来并发调用 API：
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(); }) .collectList() .map(this::getAllServicesApiSwagger) .filter(Optional::isPresent) .map(Optional::get) .subscribe(e -> { String allApiContent = getJSON(e); definitionContext.addServiceDefinition("All", allApiContent); });
这段代码的流程如下：
- Flux.fromStream：将 swaggerProperties.getUrls() 的 entrySet 转换为 Stream，再将 Stream 转换为 Flux。
- flatMap：对于 Flux 中的每个元素（服务名和 URL 的映射），并发调用 getSwaggerDefinitionForAPI 方法获取 Swagger 定义。flatMap 允许并发执行多个 Mono。
- map：将 SwaggerServiceData 和服务名封装到 SwaggerService 对象中。
- filter：过滤掉 SwaggerServiceData 为 null 的情况。
- map：将 SwaggerServiceData 转换为 JSON 字符串，并将其添加到 definitionContext 中。
- collectList：将所有 SwaggerServiceData 收集到一个 List 中，并将其转换为 Mono
- >。这是关键的一步，它会将所有并发的 Mono 的结果收集起来，并在所有 Mono 完成后才发出结果。
- map：调用 getAllServicesApiSwagger 方法，将所有 Swagger 定义合并成一个总的 Swagger 定义。
- filter：过滤掉 Optional 为空的的情况。
- map：从 Optional 中获取 SwaggerServiceData。
- subscribe：订阅 Mono，并在所有 API 调用完成后，将总的 Swagger 定义添加到 definitionContext 中。
处理 API 调用失败的情况
如果 API 调用失败，getSwaggerDefinitionForAPI 方法可能会返回一个空的 Mono。为了处理这种情况，可以使用 flatMap 和 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); });
在这个版本中，如果 swaggerServiceData 为 null，则会记录一条错误日志，并返回 Mono.empty()。Mono.empty() 不会发出任何元素，因此 collectList 不会收集到这个结果，从而避免了因 API 调用失败而导致的问题。
总结
通过使用 Flux 和 collectList 操作符，可以方便地实现并发调用多个 API，并等待所有 API 调用完成后再执行后续的聚合操作。同时，可以使用 flatMap 和 Mono.empty() 来处理 API 调用失败的情况，以确保程序的健壮性。这种方法可以应用于各种需要并发调用多个服务并将结果聚合的场景，例如微服务架构中的数据聚合、批量处理等。为了提高代码的可读性，可以将 Lambda 表达式封装成单独的方法。
