Java CompletableFuture多线程调用API慢、结果为空问题求解

第一版代码不并行的原因 #

你第一版代码里两个supplyAsync提交后，立刻就调用了listFuture1.join()，这个方法是阻塞等待第一个任务执行完成才会往下走，等第一个任务全跑完才会执行到listFuture2.join()，相当于两个任务被强制串行执行，自然没有并行效果。
同时这版把callAPI2放在了join之后的主线程执行，两个API2调用也是串行的，进一步拖慢整体速度。

第二版代码返回空集合的原因 #

thenApply绝对不会在前置任务未完成时提前执行，你拿到空集合是两个问题叠加导致的：

• 你在supplyAsync的逻辑里提前初始化了空的ArrayList，如果callAPI1调用抛出异常，catch块里没有任何处理逻辑，方法会直接返回一开始初始化的空list，异常被完全吞掉
• 你在main方法里是同步调用getList(id1)，这个方法内部会直接join等待整条任务链执行完才返回，等于你先等id1对应的全链路（API1+API2）跑完拿到res1，才会启动id2的任务链，本质还是串行执行，完全不符合两个任务流全链路并行的需求。

要实现两个独立任务流全链路并行，每个流内部按顺序先调API1、再用结果调API2，正确写法如下：

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ParallelTaskDemo {
    // 建议自定义业务专属线程池，不要用默认ForkJoinPool.commonPool，避免业务任务被公共池其他任务阻塞
    private static final ExecutorService BUSINESS_POOL = Executors.newFixedThreadPool(4);

    private static CompletableFuture<List<String>> buildTaskFlow(int id) {
        return CompletableFuture
                .supplyAsync(() -> {
                    // 不要提前初始化空集合做无效兜底，异常直接抛出让链路感知，或做明确的业务错误处理
                    try {
                        return callAPI1(id);
                    } catch (Exception e) {
                        throw new RuntimeException("调用API1失败, id=" + id, e);
                    }
                }, BUSINESS_POOL)
                .thenApplyAsync(api1Result -> {
                    try {
                        return callAPI2(api1Result);
                    } catch (Exception e) {
                        throw new RuntimeException("调用API2失败, id=" + id, e);
                    }
                }, BUSINESS_POOL);
    }

    // 模拟API1调用
    private static List<String> callAPI1(int id) throws InterruptedException {
        Thread.sleep(1000);
        return List.of("api1-res-" + id);
    }

    // 模拟API2调用
    private static List<String> callAPI2(List<String> api1Res) throws InterruptedException {
        Thread.sleep(1000);
        List<String> res = new ArrayList<>(api1Res);
        res.add("api2-res");
        return res;
    }

    public static void main(String[] args) {
        int id1 = 1;
        int id2 = 2;
        long start = System.currentTimeMillis();

        // 先把两个并行任务流全部提交，不要提交一个就阻塞等结果
        CompletableFuture<List<String>> flow1 = buildTaskFlow(id1);
        CompletableFuture<List<String>> flow2 = buildTaskFlow(id2);

        // 所有任务提交完成后再统一等待结果
        List<String> res1 = flow1.join();
        List<String> res2 = flow2.join();

        long cost = System.currentTimeMillis() - start;
        System.out.println("res1: " + res1);
        System.out.println("res2: " + res2);
        System.out.println("总耗时: " + cost + "ms"); // 总耗时约2000ms，证明两个流并行执行，每个流内部串行

        BUSINESS_POOL.shutdown();
    }
}

• 提交异步任务后不要立刻调用join()/get()阻塞，要把所有需要并行的任务全部提交后，再统一等待结果，否则会把并行任务强制变成串行
• 异步任务里的异常不要吞，如果你在catch块里不抛出、也不做正确的错误返回，会导致异常被掩盖，返回不符合预期的默认值（比如你写的空集合）
• 尽量不要用CompletableFuture默认的公共ForkJoin线程池，业务场景最好自定义独立线程池，配置合适的线程数，避免和其他共用公共池的任务互相影响
• 如果API2也是IO密集型调用，用thenApplyAsync指定线程池执行，不要用thenApply（后者默认复用前序任务的执行线程，或者如果前序任务已完成会用调用join的主线程执行，没有异步效果）

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