使用 Spring Reactive 等待所有 Mono 完成

在响应式编程中，经常会遇到需要并发调用多个服务，并在所有服务都返回结果后，对这些结果进行聚合处理的场景。Spring WebFlux 提供了强大的响应式编程能力，可以优雅地解决这类问题。本文将介绍如何使用 Spring Reactive 等待所有 Mono 完成，并执行后续的逻辑。

假设我们需要从多个 Swagger API 获取服务定义，并将这些定义添加到 definitionContext 中。每个 API 的调用都返回一个 Mono<SwaggerServiceData>。我们需要在所有 API 调用完成后，将所有服务的数据聚合起来，生成一个 "All" 的服务定义。

首先，我们定义一个临时类 SwaggerService，用于封装 SwaggerServiceData 和服务名称 serviceName：

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

然后，我们可以使用 Flux.fromStream 将服务 URL 的 entrySet 转换为 Flux。使用 flatMap 并行调用 getSwaggerDefinitionForAPI 方法，并将结果映射为 SwaggerService 对象。

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);
    });

代码解释：

1. Flux.fromStream(swaggerProperties.getUrls().entrySet().stream()): 将 swaggerProperties.getUrls() 的 entrySet 转换为 Stream，然后使用 Flux.fromStream 将其转换为 Flux。
2. flatMap((e) -> ...): 对 Flux 中的每个元素（Map.Entry）执行异步操作。getSwaggerDefinitionForAPI 方法返回一个 Mono<SwaggerServiceData>，flatMap 会自动将其转换为 Flux。
3. map(swaggerServiceData -> new SwaggerService(swaggerServiceData, e.getKey())): 将 SwaggerServiceData 和服务名称 e.getKey() 封装到 SwaggerService 对象中。
4. filter(SwaggerService::hasData): 过滤掉 SwaggerServiceData 为 null 的情况。
5. map(swaggerService -> ...): 对每个 SwaggerService 对象执行操作，将服务定义添加到 definitionContext 中。
6. collectList(): 将所有 SwaggerServiceData 对象收集到一个 List 中，并将其封装到一个 Mono<List<SwaggerServiceData>> 中。
7. map(this::getAllServicesApiSwagger): 调用 getAllServicesApiSwagger 方法，将 List<SwaggerServiceData> 转换为 Optional<SwaggerServiceData>。
8. filter(Optional::isPresent): 过滤掉 Optional 为空的情况。
9. map(Optional::get): 从 Optional 中获取 SwaggerServiceData 对象。
10. subscribe(e -> ...): 订阅 Mono，并在 SwaggerServiceData 对象可用时执行操作，将 "All" 的服务定义添加到 definitionContext 中。

错误处理：

如果 getSwaggerDefinitionForAPI 方法返回的 SwaggerServiceData 为 null，我们需要记录错误日志。可以使用 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);
    });

代码解释：

在 flatMap 中，如果 swaggerServiceData 为 null，则返回 Mono.empty()。Mono.empty() 表示一个空的 Mono，flatMap 会自动将其过滤掉。这样就可以避免在后续的操作中处理 null 值。

注意事项：

• 确保 definitionContext 是线程安全的，因为多个 API 调用是并发执行的。
• 可以将 lambda 表达式封装到有意义的方法中，以提高代码的可读性。

总结：

本文介绍了如何使用 Spring Reactive 等待所有 Mono 完成，并执行后续的逻辑。通过使用 Flux.fromStream、flatMap、collectList 等操作符，可以实现响应式地处理并发请求，并对收集到的数据进行聚合处理。同时，本文也提供了处理错误日志的示例，确保代码的健壮性。这种方式可以显著提高程序的性能和可伸缩性。