本文介绍了在使用 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<SwaggerServiceData> allServicesApi = new ArrayList<>();
swaggerProperties.getUrls().forEach((serviceName, serviceSwaggerUrl) -> {
log.debug("Attempting service definition refresh for Service : {} ", serviceName);
Mono<SwaggerServiceData> 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<SwaggerServiceData> swaggerAllServicesData = getAllServicesApiSwagger(allServicesApi);
if (swaggerAllServicesData.isPresent()) {
String allApiContent = getJSON(swaggerAllServicesData.get());
definitionContext.addServiceDefinition("All", allApiContent);
}
}
private Mono<SwaggerServiceData> getSwaggerDefinitionForAPI(String serviceName, String url) {
log.debug("Accessing the SwaggerDefinition JSON for Service : {} : URL : {} ", serviceName,
url);
Mono<SwaggerServiceData> 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<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);
});代码解释：
错误处理
上述代码没有处理 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);
});代码解释：
提高代码可读性
为了提高代码的可读性，可以将 lambda 表达式封装成单独的方法：
private Mono<SwaggerService> processSwaggerService(Map.Entry<String, String> entry) {
Mono<SwaggerServiceData> 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);
});注意事项：
总结：
使用 Flux 和 Mono 提供的组合操作，可以方便地实现并发调用 API 并等待所有结果返回。collectList() 操作符是关键，它可以确保在所有上游的 Mono 完成后才执行后续操作。通过添加错误处理和提高代码可读性，可以使代码更加健壮和易于维护。
以上就是Spring Reactive：等待所有 Mono 完成的详细内容，更多请关注php中文网其它相关文章！
每个人都需要一台速度更快、更稳定的 PC。随着时间的推移，垃圾文件、旧注册表数据和不必要的后台进程会占用资源并降低性能。幸运的是，许多工具可以让 Windows 保持平稳运行。
广告Copyright 2014-2025 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号
685
收藏
