Reactive FlatMap并发发送HTTP请求无法扩容问题排查

1. flatMap是否是实现该需求的最佳方式？ #

flatMap是实现高并发HTTP请求的合适选择之一，它能便捷控制并行请求数量，自动处理异步结果的聚合。如果需要更精细化的流量控制（比如动态调整并发数、自定义背压策略），可以结合Semaphore手动控制请求发射，或者使用Flux.create编写自定义请求逻辑，但flatMap仍是最简洁高效的方案，无需额外复杂编码。

2. 调整flatMap并发参数无法提升请求量的原因 #

核心问题出在Jetty HttpClient的连接池配置上：

• 你自定义的ClientConnector和HttpClientTransportDynamic未配置足够的连接池大小，Jetty客户端默认的单目标主机最大连接数通常为16，这直接限制了同时能建立的HTTP连接数。
• 即使flatMap设置了2048的并发数，实际能发送的请求被连接池限制在16个，而第三方服务每个请求固定耗时1秒，因此每秒请求量稳定在16次，和连接池大小完全匹配。

步骤1：正确配置Jetty HttpClient连接池 #

修改WebClient的Bean配置，显式设置连接池的最大连接数，同时可结合虚拟线程提升并发能力（Java 19+）：

@Bean
public WebClient webClient(WebClient.Builder builder) {
    HttpClient httpClient = HttpClient.newHttpClient()
            .connectionPool(new ConnectionPool.DirectConnectionPool()
                    .maxConnections(2048) // 全局最大连接数
                    .maxConnectionsPerDestination(2048)) // 目标主机的最大连接数，需和flatMap并发数匹配
            .responseTimeout(Duration.ofSeconds(10)) // 根据服务处理时间调整超时
            .executor(Executors.newVirtualThreadPerTaskExecutor()); // 使用虚拟线程减少线程开销

    return builder
            .baseUrl("http://some-host:8080/api/path")
            .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
            .clientConnector(new JettyClientHttpConnector(httpClient))
            .build();
}

步骤2：优化Flux输入逻辑 #

Flux.interval会受下游背压限制，当请求处理速度跟不上发射速度时，会自动降低发射频率。如果需要持续高吞吐的输入，可改用Flux.range或Flux.generate：

// 生成无背压限制的请求输入流
Flux<String> inputFlux = Flux.range(0, Integer.MAX_VALUE)
        .map(i -> "transform to request payload number " + i);

// 设置flatMap并发数与连接池最大连接数一致
Flux<String> resultFlux = inputFlux.flatMap(oneInput -> 
        webClient.post()
                .bodyValue(oneInput)
                .retrieve()
                .bodyToMono(String.class),
        2048 // 并发数需和连接池配置匹配
);

return resultFlux.map(oneResult -> doSomething(oneResult));

额外注意事项 #

• 若使用Java 19以下版本，可改用线程池（如Executors.newFixedThreadPool(2048)）替代虚拟线程，但需注意线程开销。
• 可通过metrics监控连接池使用情况，确认连接数是否达到配置上限，进一步调优参数。

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