Spring Webflux：如何等待Flux.fromIterable全部执行完成后返回响应

问题 #

我有一个运行正常的响应式服务，现在需要给GetCasesListResponse数组中的每个元素，从另一服务并行获取额外数据并添加到对应元素中。我尝试用Flux.fromIterable实现，但getMoreComments方法会在Flux.fromIterable的操作完成前就返回Mono.just(casesListResponse)，导致返回的响应里没有附加数据。请问怎么才能等所有Flux.fromIterable的操作完成后再返回响应？

原有正常代码 #

public Mono<GetCasesListResponse> getCasesList(@RequestBody GetCasesListRequest request) {
        return casesService.getCasesList(mapToGetCasesListServiceRequest(request))
                .map(GetCasesListMapper::mapGetCasesListResponse)
                .onErrorResume(exceptionHandler::fallbackMethod);
}

casesService 代码 #

public Mono<GetCasesListServiceResponse> getCasesList(GetCasesListServiceRequest request) {
        return webClient.post()
                .uri("/getCasesList")
                .bodyValue(request)
                .retrieve()
                .bodyToMono(GetCasesListResponse.class);
}

修改后的代码 #

public Mono<GetCasesListResponse> getCasesList(@RequestBody GetCasesListRequest request) {
        return casesService.getCasesList(mapToGetCasesListServiceRequest(request))
                .map(GetCasesListMapper::mapGetCasesListResponse)
                .flatMap(this::getMoreComments)
                .onErrorResume(exceptionHandler::fallbackMethod);
}

private Mono<GetCasesListResponse> getMoreComments(GetCasesListResponse casesListResponse) {
        collectExtIds(casesListResponse).forEach((id, extIdList) -> fromIterable(extIdList).parallel()
                .flatMap(extId -> stpService.getMoreComments(mapToGetMoreCommentsRequest(extId)))
                .subscribe(moreCommentsResponse -> addMoreCommentsToGetCasesListResponse(casesListResponse, id, moreCommentsResponse)));

        return Mono.just(casesListResponse);
}

stpService 的 getMoreComments 方法 #

public Mono<GetCommentResponse> getMoreComments(GetCommentRequest request) {
        return webClient.post()
                .uri("/getCaseComments")
                .bodyValue(request)
                .retrieve()
                .bodyToMono(GetCommentResponse.class);
}

解决方案 #

核心问题是你在getMoreComments里用了subscribe()触发异步请求，但没有把这些异步操作的完成信号和返回的Mono绑定，导致方法直接返回了未处理完的casesListResponse。

要解决这个问题，需要把所有并行获取额外数据的流组合成一个统一的响应式链，等所有操作完成后再返回处理好的响应。修改后的getMoreComments方法如下：

private Mono<GetCasesListResponse> getMoreComments(GetCasesListResponse casesListResponse) {
    // 把所有需要执行的异步操作收集成一个Flux
    Flux<Void> allOperations = Flux.fromIterable(collectExtIds(casesListResponse).entrySet())
            .flatMap(entry -> {
                String id = entry.getKey();
                List<String> extIdList = entry.getValue();
                // 并行处理每个extId的请求
                return Flux.fromIterable(extIdList)
                        .parallel()
                        .runOn(Schedulers.parallel()) // 显式指定并行调度器，优化性能
                        .flatMap(extId -> stpService.getMoreComments(mapToGetMoreCommentsRequest(extId)))
                        .doOnNext(moreCommentsResponse -> 
                            addMoreCommentsToGetCasesListResponse(casesListResponse, id, moreCommentsResponse)
                        )
                        .sequential(); // 并行流转串行，统一后续处理逻辑
            });

    // 等待所有异步操作完成后，返回处理好的响应
    return allOperations.then(Mono.just(casesListResponse));
}

关键修改说明 #

1. 替换subscribe()为doOnNext()：subscribe()会直接触发流但脱离响应式链的编排，doOnNext()可以在处理元素时执行附加逻辑，同时保持流的链式结构。
2. 组合所有异步流：用Flux.fromIterable遍历collectExtIds的结果，把每个条目对应的并行请求流合并成一个大的Flux<Void>（我们只关心操作完成，不关心返回值）。
3. 用then()等待全部完成：allOperations.then()会等待整个Flux的所有元素处理完毕，再发出后续的Mono.just(casesListResponse)，确保返回时所有额外数据都已添加完成。
4. 显式指定并行调度器：parallel().runOn(Schedulers.parallel())可以明确指定并行操作的线程池，避免默认调度器可能带来的性能瓶颈。

内容的提问来源于stack exchange，提问作者Dmitriy88