本文详解如何在 spring 应用中安全、高效地并行执行多个子 rest 请求，解决串行调用导致的响应延迟问题，并提供线程安全的数据收集方案与最佳实践。
在构建高可用、低延迟的 Spring REST API 时，常会遇到“父请求聚合多个子请求”的典型场景：例如一个订单查询接口需同步调用库存、物流、用户积分等下游服务（10–50 个不等）。若采用传统 for 循环串行调用，总耗时 ≈ 各子请求耗时之和，极易成为性能瓶颈。此时，并行化是关键优化手段——但直接使用 parallelStream().forEach() 会因 Java 的闭包变量限制（要求局部变量 final 或 effectively final）导致无法安全收集结果，如报错：“childResponse needs to be final or effectively final”。
✅ 正确做法：使用线程安全容器 + 明确请求标识
核心原则：每个子请求必须具备唯一、稳定的标识符（如 id、correlationId 或索引），以便结果可追溯、可重组。 基于此，推荐两种生产级方案：
方案一：按顺序索引存储（适合有序且 ID 连续的场景）
// 初始化线程安全的动态数组（预设容量避免扩容竞争）
final List<ChildResponse> childResponses =
Collections.synchronizedList(new ArrayList<>(childRequests.size()));
// 预填充占位符，确保索引稳定（如 childRequest.getId() 返回 0,1,2...）
childRequests.forEach(req -> childResponses.add(null));
// 并行发起请求
childRequests.parallelStream().forEach(request -> {
try {
ChildResponse response = restTemplate.getForObject(
"https://api.example.com/child/" + request.getId(),
ChildResponse.class
);
// 安全写入指定索引（synchronizedList 保证线程安全）
childResponses.set(request.getId(), response);
} catch (Exception e) {
log.error("Failed to fetch child response for id: {}", request.getId(), e);
childResponses.set(request.getId(), ChildResponse.error(request.getId(), e.getMessage()));
}
});
// 后续按原始顺序组装 parent response（保障业务语义一致性）
ParentResponse parentResponse = buildParentResponse(childRequests, childResponses);方案二：按唯一键映射存储（更通用，推荐）
// 使用 ConcurrentHashMap —— 无锁、高性能、天然线程安全
final Map<String, ChildResponse> responseMap = new ConcurrentHashMap<>();
childRequests.parallelStream().forEach(request -> {
try {
String requestId = request.getCorrelationId(); // 如 UUID 或业务主键
ChildResponse response = webClient.get()
.uri("https://api.example.com/child/{id}", requestId)
.retrieve()
.bodyToMono(ChildResponse.class)
.block(); // 注意：此处仅作示例；生产建议用 Mono.zip + Flux.merge
responseMap.put(requestId, response);
} catch (Exception e) {
responseMap.put(request.getCorrelationId(),
ChildResponse.error(request.getCorrelationId(), e.getMessage()));
}
});
// 按原始 childRequests 顺序提取结果，保持业务逻辑可控
List<ChildResponse> orderedResponses = childRequests.stream()
.map(req -> responseMap.getOrDefault(req.getCorrelationId(), null))
.collect(Collectors.toList());⚠️ 关键注意事项与进阶建议
-
避免 parallelStream() 的隐式线程池陷阱：默认使用 ForkJoinPool.commonPool()，其大小受限于 CPU 核心数（通常为 Runtime.getRuntime().availableProcessors() - 1），可能成为 I/O 密集型调用的瓶颈。建议显式配置自定义线程池：
final ExecutorService executor = Executors.newFixedThreadPool(20); // 根据下游 QPS 调优 childRequests.parallelStream().forEach(request -> { /* ... */ }); // ❌ 不生效 // ✅ 正确方式：使用 CompletableFuture + 自定义 Executor List<CompletableFuture<ChildResponse>> futures = childRequests.stream() .map(req -> CompletableFuture.supplyAsync(() -> callApi(req), executor)) .collect(Collectors.toList()); CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); -
优先选用响应式编程（WebClient + Project Reactor）：相比阻塞式 RestTemplate，WebClient 的非阻塞特性可实现更高并发吞吐。结合 Flux.merge 可优雅实现并行+错误隔离：
Flux<ChildResponse> responses = Flux.fromIterable(childRequests) .flatMap(req -> webClient.get() .uri("/child/{id}", req.getId()) .retrieve() .bodyToMono(ChildResponse.class) .onErrorResume(e -> Mono.just(ChildResponse.error(req.getId(), e.getMessage()))) .subscribeOn(Schedulers.boundedElastic()), // 控制并发数 10 // maxConcurrency ); List<ChildResponse> result = responses.collectList().block(); 务必添加超时与熔断：并行调用放大了下游故障风险。使用 WebClient 的 timeout() + Resilience4j 的 CircuitBreaker 是标配。
结果重组需尊重原始顺序：即使调用并行，业务上往往要求子响应与子请求严格一一对应（如前端表格渲染）。切勿依赖 parallelStream() 的执行顺序——始终通过 request.getId() 查找，或使用 Collectors.toMap(..., LinkedHashMap::new) 保持插入序。
综上，并行 REST 调用不是简单加 .parallel()，而是围绕标识唯一性、容器线程安全性、执行资源可控性、结果可追溯性四要素系统设计。从 ConcurrentHashMap 到 WebClient + Reactor，每一步升级都让系统更健壮、更可伸缩。
