Java Spring中如何多线程执行耗时API任务并等待全部完成？

通用逻辑思路 #

完全贴合需求的执行步骤如下：

• 单线程执行Operation A，完成对整个物品列表的预处理
• 将Operation A输出的列表拆分为独立任务，每个任务对应单个物品的Operation B API调用，提交至多线程池执行
• 阻塞主线程，等待所有Operation B任务全部执行完成，收集所有返回结果
• 单线程执行Operation C，传入Operation B的全部结果做后续处理

主流语言实现示例 #

Java 实现 #

用JDK自带工具类即可实现，无需引入第三方依赖：

// 1. 单线程执行Operation A
List<Item> processedList = operationA(originalItemList);
int taskNum = processedList.size();

// 2. 初始化线程池（生产环境建议自定义ThreadPoolExecutor，不要直接用Executors默认实现）
// 最大线程数按下游API承载能力设置，避免把下游打崩
ExecutorService executor = Executors.newFixedThreadPool(Math.min(taskNum, 10));
CountDownLatch countDownLatch = new CountDownLatch(taskNum);
List<BOperateResult> bResults = Collections.synchronizedList(new ArrayList<>());

// 3. 批量提交Operation B任务
for (Item item : processedList) {
    executor.submit(() -> {
        try {
            BOperateResult result = operationB(item);
            bResults.add(result);
        } finally {
            countDownLatch.countDown();
        }
    });
}

// 4. 等待所有B任务执行完成，设置超时时间避免永久阻塞
countDownLatch.await(30, TimeUnit.SECONDS);
executor.shutdown();

// 5. 单线程执行Operation C
operationC(bResults);

如果不想手动维护CountDownLatch，用invokeAll写法更简洁：

List<Callable<BOperateResult>> tasks = processedList.stream()
    .map(item -> (Callable<BOperateResult>) () -> operationB(item))
    .collect(Collectors.toList());
// 等待所有任务执行完成后统一返回Future列表
List<Future<BOperateResult>> futures = executor.invokeAll(tasks, 30, TimeUnit.SECONDS);
// 遍历Future列表获取结果即可

Python 实现 #

用标准库concurrent.futures即可实现，代码非常精简：

from concurrent.futures import ThreadPoolExecutor, as_completed

# 1. 单线程执行Operation A
processed_list = operation_a(original_item_list)

# 2. 多线程执行所有Operation B
b_results = []
max_workers = min(len(processed_list), 10)
with ThreadPoolExecutor(max_workers=max_workers) as executor:
    futures = [executor.submit(operation_b, item) for item in processed_list]
    # 等待所有任务完成后收集结果
    for future in as_completed(futures):
        b_results.append(future.result())

# 3. 单线程执行Operation C
operation_c(b_results)

注意事项 #

• 线程池大小不要设置过大，要匹配下游Operation B接口的限流阈值，避免请求量过高触发下游限流或者直接打挂服务
• 必须给等待Operation B的步骤加上超时时间，避免下游接口故障导致当前请求永久阻塞
• 如果业务允许Operation B部分失败，可以根据需求加重试逻辑，或者统一收集失败结果在Operation C中做兼容处理

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