异步调用接口批量获取数据的函数性能异常排查

问题背景 #

业务需求：调用第三方接口获取数据，接口限制时间间隔≤24小时，超时时需拆分时间段多次调用。使用@Async实现异步调用后，响应时间随时间段数量线性上升，未达到预期并发效果。

代码实现 #

public List<Response> makeAsyncCalls(Collection<Pair<LocalDateTime, LocalDateTime>> intervals) {

    DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss");
    List<CompletableFuture<Response>> responses = new ArrayList<>();
    intervals.stream()
        .map((Pair<LocalDateTime, LocalDateTime> interval) -> Pair.of(interval.getLeft().format(formatter),
            interval.getRight().format(formatter))).forEach(timeInterval -> {
            try {
                responses.add(
                    fetchDataFromEndPoint(timeInterval));
            } catch (Exception e) {
                // 抛出自定义异常
            }
        });
    CompletableFuture<Void> completedFuture = CompletableFuture.allOf(responses.toArray(new CompletableFuture[0]));
    if (completedFuture.isCompletedExceptionally()) {
        // 处理异常
    }
    return responses.stream().map(CompletableFuture::join).collect(toList());
}

@Async
public CompletableFuture<Response> fetchDataFromEndPoint(Pair<String, String> interval)
            throws Exception {
            // 调用第三方接口获取响应
            Response response = service.getResponse(interval.getLeft(), interval.getRight());
            return CompletableFuture.completedFuture(response);
        }

响应时间数据 #

问题根源 #

1. @Async方法内部为同步执行：fetchDataFromEndPoint中先同步调用service.getResponse()阻塞线程，再返回已完成的CompletableFuture。@Async仅将方法提交到线程池，但核心HTTP请求是同步阻塞的，若线程池核心数不足，请求会排队串行执行。
2. 默认线程池配置不合理：Spring默认@Async线程池（SimpleAsyncTaskExecutor）无复用机制，或默认ThreadPoolTaskExecutor核心线程数过少（默认1），无法支撑并发请求。
3. HTTP客户端无连接池：若service.getResponse()未使用连接池，每次请求新建TCP连接，开销大且无法复用，进一步拉长响应时间。

修复方案 #

1. 让HTTP请求真正异步 #

替换同步HTTP客户端为异步实现（如Spring WebClient），直接返回异步结果，避免线程阻塞：

@Async("asyncTaskExecutor")
public CompletableFuture<Response> fetchDataFromEndPoint(Pair<String, String> interval) {
    return webClient.get()
            .uri(uriBuilder -> uriBuilder
                    .path("/data-api")
                    .queryParam("start", interval.getLeft())
                    .queryParam("end", interval.getRight())
                    .build())
            .retrieve()
            .bodyToMono(Response.class)
            .toFuture();
}

2. 自定义异步线程池 #

配置合理的线程池参数，满足并发需求：

@Configuration
@EnableAsync
public class AsyncConfig {

    @Bean(name = "asyncTaskExecutor")
    public Executor asyncTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10); // 根据业务并发量调整
        executor.setMaxPoolSize(20);
        executor.setQueueCapacity(50);
        executor.setThreadNamePrefix("DataFetch-");
        executor.initialize();
        return executor;
    }
}

3. 配置HTTP连接池 #

以WebClient为例，启用连接池复用TCP连接：

@Bean
public WebClient webClient() {
    HttpClient httpClient = HttpClient.create()
            .poolConfig(pool -> pool
                    .maxConnections(20)
                    .pendingAcquireTimeout(Duration.ofSeconds(10)));
    ClientHttpConnector connector = new ReactorClientHttpConnector(httpClient);
    return WebClient.builder()
            .clientConnector(connector)
            .baseUrl("https://third-party-api-domain.com")
            .build();
}

4. 优化异常处理 #

完善全局异常捕获，避免遗漏失败请求：

CompletableFuture<Void> completedFuture = CompletableFuture.allOf(responses.toArray(new CompletableFuture[0]))
        .exceptionally(ex -> {
            throw new DataFetchException("批量请求失败", ex);
        });
completedFuture.join(); // 等待所有任务完成

预期效果 #

调整后，多时间段请求将真正并发执行，响应时间趋近于单个请求耗时（如5个请求耗时约9秒），大幅提升效率。

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