如何基于Spring RestTemplate实现多URL并行调用并合并结果

嘿，这个需求我之前帮同事处理过，并行调用确实能把接口聚合的性能拉满，结合Spring生态有几种靠谱的方案，我给你拆解清楚：

这是我最推荐的方式，因为你能完全掌控每个异步任务的生命周期，包括异常处理和线程资源。

步骤1：自定义线程池（必做！） #

别用JDK默认的ForkJoinPool，它是全局共享的，容易和其他业务抢资源。我们自己配置一个线程池：

@Configuration
public class AsyncThreadPoolConfig {
    @Bean("restAsyncExecutor")
    public ThreadPoolTaskExecutor restAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5); // 根据你的并发量调整，比如URL数量的1/2或者1/3
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(20);
        executor.setThreadNamePrefix("Rest-Async-Worker-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); // 拒绝策略，避免任务丢失
        executor.initialize();
        return executor;
    }
}

步骤2：包装RestTemplate调用为异步任务 #

把每个URL的请求包装成CompletableFuture，然后批量执行：

@Service
public class ParallelDataService {
    private final RestTemplate restTemplate;
    private final ThreadPoolTaskExecutor restAsyncExecutor;

    // 构造注入（Spring 4.3+支持）
    public ParallelDataService(RestTemplate restTemplate, 
                               @Qualifier("restAsyncExecutor") ThreadPoolTaskExecutor restAsyncExecutor) {
        this.restTemplate = restTemplate;
        this.restAsyncExecutor = restAsyncExecutor;
    }

    public MergedResponse mergeParallelResults(List<String> targetUrls) {
        // 把每个URL调用转为异步任务
        List<CompletableFuture<SingleResponse>> futureTasks = targetUrls.stream()
                .map(url -> CompletableFuture.supplyAsync(() -> {
                    // 这里就是原来的串行调用逻辑
                    return restTemplate.getForObject(url, SingleResponse.class);
                }, restAsyncExecutor)
                .exceptionally(ex -> {
                    // 单个任务失败的处理：记录日志+返回默认值，不影响整体流程
                    log.error("请求URL: {} 失败", url, ex);
                    return new SingleResponse(); // 根据你的业务定义默认返回对象
                }))
                .collect(Collectors.toList());

        // 等待所有任务完成，收集结果
        List<SingleResponse> allResults = futureTasks.stream()
                .map(CompletableFuture::join) // join()不会抛出检查型异常，比get()省心
                .collect(Collectors.toList());

        // 合并结果到最终对象
        return assembleMergedResult(allResults);
    }

    // 你的结果合并逻辑，根据业务需求自定义
    private MergedResponse assembleMergedResult(List<SingleResponse> responses) {
        MergedResponse merged = new MergedResponse();
        merged.setTotalItems(responses.stream().mapToInt(SingleResponse::getItemCount).sum());
        merged.setAllItems(responses.stream()
                .flatMap(resp -> resp.getItems().stream())
                .collect(Collectors.toList()));
        return merged;
    }
}

关键注意点 #

• 一定要给supplyAsync()指定自定义线程池，不然会用默认的ForkJoinPool，风险很高。
• 用exceptionally()处理单个任务的异常，避免一个URL失败导致整个聚合任务崩溃。

如果不想手动写CompletableFuture，Spring的@Async注解能帮你快速实现异步调用。

步骤1：开启异步支持+配置线程池 #

@Configuration
@EnableAsync // 必须加这个注解开启异步功能
public class AsyncConfig {
    @Bean("asyncRestExecutor")
    public ThreadPoolTaskExecutor asyncRestExecutor() {
        // 和方案一的线程池配置一样
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(20);
        executor.setThreadNamePrefix("Async-Rest-");
        executor.initialize();
        return executor;
    }
}

步骤2：写异步调用方法 #

@Service
public class AsyncRestClient {
    private final RestTemplate restTemplate;

    public AsyncRestClient(RestTemplate restTemplate) {
        this.restTemplate = restTemplate;
    }

    @Async("asyncRestExecutor") // 指定用我们自定义的线程池
    public Future<SingleResponse> fetchSingleResponse(String url) {
        try {
            SingleResponse response = restTemplate.getForObject(url, SingleResponse.class);
            return new AsyncResult<>(response);
        } catch (Exception ex) {
            log.error("请求URL {}失败", url, ex);
            return new AsyncResult<>(new SingleResponse());
        }
    }
}

步骤3：批量调用并合并结果 #

@Service
public class ParallelDataService {
    private final AsyncRestClient asyncRestClient;

    public ParallelDataService(AsyncRestClient asyncRestClient) {
        this.asyncRestClient = asyncRestClient;
    }

    public MergedResponse mergeParallelResults(List<String> targetUrls) {
        List<Future<SingleResponse>> futureList = new ArrayList<>();
        // 提交所有异步任务
        for (String url : targetUrls) {
            futureList.add(asyncRestClient.fetchSingleResponse(url));
        }

        // 收集结果
        List<SingleResponse> allResults = new ArrayList<>();
        for (Future<SingleResponse> future : futureList) {
            try {
                allResults.add(future.get()); // get()会抛出检查型异常，需要捕获
            } catch (InterruptedException | ExecutionException e) {
                log.error("获取异步结果失败", e);
                allResults.add(new SingleResponse());
            }
        }

        return assembleMergedResult(allResults);
    }

    // 同样的合并方法
    private MergedResponse assembleMergedResult(List<SingleResponse> responses) {
        MergedResponse merged = new MergedResponse();
        merged.setTotalItems(responses.stream().mapToInt(SingleResponse::getItemCount).sum());
        merged.setAllItems(responses.stream()
                .flatMap(resp -> resp.getItems().stream())
                .collect(Collectors.toList()));
        return merged;
    }
}

如果你的Spring版本是5+，我强烈推荐用WebClient代替RestTemplate——它是Spring官方推荐的非阻塞HTTP客户端，配合响应式编程能实现更高效的并行调用，而且代码更简洁：

@Service
public class ReactiveParallelService {
    private final WebClient webClient;

    public ReactiveParallelService(WebClient.Builder webClientBuilder) {
        this.webClient = webClientBuilder.build();
    }

    public MergedResponse mergeParallelResults(List<String> targetUrls) {
        // 用Flux实现并行调用，默认就是并行的
        Flux<SingleResponse> responseFlux = Flux.fromIterable(targetUrls)
                .flatMap(url -> webClient.get()
                        .uri(url)
                        .retrieve()
                        .bodyToMono(SingleResponse.class)
                        .onErrorReturn(new SingleResponse())); // 异常处理

        // 如果是响应式场景，可以不用block()，直接返回Mono<MergedResponse>
        List<SingleResponse> allResults = responseFlux.collectList().block();
        return assembleMergedResult(allResults);
    }

    // 同样的合并方法
    private MergedResponse assembleMergedResult(List<SingleResponse> responses) {
        MergedResponse merged = new MergedResponse();
        merged.setTotalItems(responses.stream().mapToInt(SingleResponse::getItemCount).sum());
        merged.setAllItems(responses.stream()
                .flatMap(resp -> resp.getItems().stream())
                .collect(Collectors.toList()));
        return merged;
    }
}

• 如果你要兼容现有RestTemplate代码，优先选CompletableFuture方案，灵活度最高，可控性强。
• 简单场景可以用**@Async注解**，代码更简洁，不需要手动处理CompletableFuture。
• 新项目或者性能要求高的场景，直接上WebClient，非阻塞性能比RestTemplate高很多。

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