Spring Reactive 如何等待所有 Mono 完成

在使用Spring Reactive进行响应式编程时，经常需要并发调用多个服务，并在所有服务返回结果后进行聚合处理。本文深入探讨了如何利用Spring WebFlux的强大功能，优雅地等待所有`Mono`完成，并执行后续逻辑，例如从多个Swagger API获取服务定义并将其添加到`definitionContext`中。通过`Flux.fromStream`、`flatMap`、`collectList`等操作符，实现了并发请求的响应式处理和数据聚合。此外，本文还提供了处理错误日志的实用示例，确保代码的健壮性，以及线程安全和代码可读性方面的注意事项。掌握这些技巧，能够显著提升程序的性能和可伸缩性，有效解决响应式编程中的并发等待问题。

使用 Spring Reactive 等待所有 Mono 完成

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

假设我们需要从多个 Swagger API 获取服务定义，并将这些定义添加到 definitionContext 中。每个 API 的调用都返回一个 Mono。我们需要在所有 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 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，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> 中。
7. map(this::getAllServicesApiSwagger): 调用 getAllServicesApiSwagger 方法，将 List 转换为 Optional。
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 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 等操作符，可以实现响应式地处理并发请求，并对收集到的数据进行聚合处理。同时，本文也提供了处理错误日志的示例，确保代码的健壮性。这种方式可以显著提高程序的性能和可伸缩性。

好了，本文到此结束，带大家了解了《Spring Reactive 如何等待所有 Mono 完成》，希望本文对你有所帮助！关注golang学习网公众号，给大家分享更多文章知识！