Spring Webflux：返回响应前等待所有Flux<T>完成

你当前代码的核心问题是手动调用subscribe()并同步返回集合，完全违背了Reactor异步非阻塞的设计逻辑：

• forEach里的subscribe()会触发异步插入，但不会等待这些操作完成
• 外层的map是同步操作，会立刻返回空的buf（或仅填充少量已完成的插入结果），导致控制器提前响应，后续未完成的插入操作被直接终止

重构Service层代码，用Reactor原生操作符正确编排异步流，确保所有插入操作完成后再返回结果：

修正后的Service代码 #

public Mono<List<Foo>> addFoos() {
    return webClient.getSomeStuff()
            // 将获取到的结果集合转为Flux，实现逐个异步处理
            .flatMapMany(results -> Flux.fromIterable(
                    results.stream()
                            .map(/* ...transform to domain model */)
                            .toList()
            ))
            // 并行处理每个Foo的插入（如需串行执行，替换为concatMap）
            .flatMap(foo -> fooDao.createFoo(foo))
            // 等待所有插入操作完成，收集结果为List后返回
            .collectList();
}

DAO代码优化（可选但更符合语义） #

插入单条数据应该返回单个结果，建议把createFoo的返回类型从Flux<Foo>改为Mono<Foo>：

public Mono<Foo> createFoo(final Foo foo) {
    var query = "<INSERT_STUFF_STATEMENT>";

    var connection = Mono.from(publisher);
    return connection.flatMap(c -> c.createStatement(query)
            .bind("STUFF", foo.getStuff())
            .execute())
            // 取插入操作的第一个结果（单条插入仅会有一个结果）
            .next()
            .map(result -> Foo.builder()
                    .stuff(whatever)
                    .build());
}

• flatMapMany：将获取到的集合转为Flux，让每个元素都能进入异步处理流
• flatMap/concatMap：处理单个Foo的插入操作，flatMap默认并行执行，concatMap保证串行执行
• collectList：等待Flux中所有元素处理完成，收集为List后包装成Mono返回，确保控制器在所有插入完成后才响应

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