Spring WebFlux并行执行HTTP请求并反序列化响应：并行未生效问题排查求助

看起来你明明用了parallel()但请求没并行，我帮你排查下核心问题和优化点：

核心问题所在 #

1. WebClient重复创建，严重拖慢并行效率
   你在performGetRequest方法里每次都新建WebClient实例，而WebClient是设计成线程安全、可复用的，每次创建都会初始化底层的HTTP连接池、客户端配置等资源，这个开销会直接抵消并行的优势，甚至让请求看起来像是串行执行。

2. 流操作存在冗余步骤
   你最后那段.collectList().flatMapMany(Flux::fromIterable).collectList()完全是多余的——第一次collectList已经把所有结果收集成List了，后面的操作只是把它拆成Flux再重新收集，平白增加了不必要的性能开销。

3. 调度器选择可能不合适
   Schedulers.boundedElastic()是为阻塞IO操作设计的，而WebClient的请求本身是非阻塞异步的，用这个调度器会带来额外的线程切换成本，反而影响并行效率。

优化后的代码示例 #

第一步：复用WebClient实例 #

把WebClient改成类级别的成员变量，在构造方法里初始化：

private final WebClient webClient;

// 构造方法中初始化WebClient，传入基础API地址
public YourServiceClass(String apiBaseUrl) {
    this.webClient = WebClient.create(apiBaseUrl);
}

public Flux<JsonNode> performGetRequest(String url) {
    // 用uri方法拼接路径，比手动字符串拼接更规范、更安全
    return webClient.get()
            .uri(url)
            .retrieve()
            .bodyToFlux(JsonNode.class);
}

第二步：简化并优化并行流操作 #

public List<CustomModel> fetch(List<String> urls) {
    // 可以手动指定并行度，比如限制为10（根据目标API的并发限制调整，避免触发限流）
    int parallelism = Math.min(urls.size(), 10);
    
    return Flux.fromIterable(urls)
            .parallel(parallelism)
            // 非阻塞异步操作用Schedulers.parallel()更合适
            .runOn(Schedulers.parallel())
            .flatMap(this::performGetRequest)
            // 如果customDeserialize是耗时操作，也可以给它单独指定调度器并行处理
            .flatMap(jsonNode -> 
                Flux.fromIterable(customDeserialize(jsonNode))
                    .subscribeOn(Schedulers.parallel())
            )
            .sequential()
            .collectList()
            .block();
}

额外注意事项 #

• 并行度不要设置过大：虽然我们想并行，但目标API可能有并发请求限制，如果并行数太高，容易触发429限流或者被封禁，建议根据API文档设置合理的并行度（比如10-20之间）。
• 检查customDeserialize是否阻塞：如果你的反序列化逻辑涉及复杂计算或者阻塞操作，一定要用subscribeOn把它放到单独的调度器里，避免阻塞WebClient的异步线程。

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