使用 Spring Reactive 等待所有 Mono 完成
在响应式编程中，经常会遇到需要并发调用多个服务，并在所有服务都返回结果后，对这些结果进行聚合处理的场景。Spring WebFlux 提供了强大的响应式编程能力，可以优雅地解决这类问题。本文将介绍如何使用 Spring Reactive 等待所有 Mono 完成，并执行后续的逻辑。
假设我们需要从多个 Swagger API 获取服务定义，并将这些定义添加到 definitionContext 中。每个 API 的调用都返回一个 Mono
首先，我们定义一个临时类 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);
}); 代码解释：
- Flux.fromStream(swaggerProperties.getUrls().entrySet().stream()): 将 swaggerProperties.getUrls() 的 entrySet 转换为 Stream，然后使用 Flux.fromStream 将其转换为 Flux。
- flatMap((e) -> ...): 对 Flux 中的每个元素（Map.Entry）执行异步操作。getSwaggerDefinitionForAPI 方法返回一个 Mono
，flatMap 会自动将其转换为 Flux。 - map(swaggerServiceData -> new SwaggerService(swaggerServiceData, e.getKey())): 将 SwaggerServiceData 和服务名称 e.getKey() 封装到 SwaggerService 对象中。
- filter(SwaggerService::hasData): 过滤掉 SwaggerServiceData 为 null 的情况。
- map(swaggerService -> ...): 对每个 SwaggerService 对象执行操作，将服务定义添加到 definitionContext 中。
- collectList(): 将所有 SwaggerServiceData 对象收集到一个 List 中，并将其封装到一个 Mono
- > 中。
- map(this::getAllServicesApiSwagger): 调用 getAllServicesApiSwagger 方法，将 List
转换为 Optional 。 - filter(Optional::isPresent): 过滤掉 Optional 为空的情况。
- map(Optional::get): 从 Optional 中获取 SwaggerServiceData 对象。
- 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 等操作符，可以实现响应式地处理并发请求，并对收集到的数据进行聚合处理。同时，本文也提供了处理错误日志的示例，确保代码的健壮性。这种方式可以显著提高程序的性能和可伸缩性。
