如何用CompletableFuture在@Async方法内实现子任务并行执行？

基础实现示例 #

1. 开启异步支持（必要配置）

先确保Spring项目启用异步功能，添加配置类：

import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;

@Configuration
@EnableAsync
public class AsyncConfig {
    // 推荐自定义线程池，避免默认线程池的性能问题
    /*
    @Bean(name = "customAsyncExecutor")
    public Executor asyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(20);
        executor.setThreadNamePrefix("AsyncWorker-");
        executor.initialize();
        return executor;
    }
    */
}

2. 实现Service1的并行子任务

Service1的method1内部需要并行执行多个子任务，核心是用CompletableFuture触发子任务并行：

import org.springframework.stereotype.Service;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;

@Service
public class Service1 {
    // 若使用自定义线程池，注入后指定
    // @Autowired
    // @Qualifier("customAsyncExecutor")
    // private Executor customExecutor;

    // 子任务T1.1
    private String task1_1() {
        // 模拟复杂业务逻辑
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return "T1.1执行结果";
    }

    // 子任务T1.2
    private Integer task1_2() {
        try {
            Thread.sleep(1200);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return 200;
    }

    // 子任务T1.3
    private Boolean task1_3() {
        try {
            Thread.sleep(800);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return true;
    }

    // Method1：内部并行执行子任务
    public CompletableFuture<CombinedResult> method1() {
        // 启动所有子任务并行执行
        CompletableFuture<String> t1_1Future = CompletableFuture.supplyAsync(this::task1_1);
        CompletableFuture<Integer> t1_2Future = CompletableFuture.supplyAsync(this::task1_2);
        CompletableFuture<Boolean> t1_3Future = CompletableFuture.supplyAsync(this::task1_3);
        // 自定义线程池写法：supplyAsync(this::task1_1, customExecutor)

        // 等待所有子任务完成后合并结果
        return CompletableFuture.allOf(t1_1Future, t1_2Future, t1_3Future)
                .thenApply(v -> {
                    String res1 = t1_1Future.join();
                    Integer res2 = t1_2Future.join();
                    Boolean res3 = t1_3Future.join();
                    return new CombinedResult(res1, res2, res3);
                });
    }

    // 封装Method1的合并结果
    public static class CombinedResult {
        private String res1;
        private Integer res2;
        private Boolean res3;

        public CombinedResult(String res1, Integer res2, Boolean res3) {
            this.res1 = res1;
            this.res2 = res2;
            this.res3 = res3;
        }

        // 按需添加getter/setter
    }
}

3. 定义其他Service（Service2、Service3）

import org.springframework.stereotype.Service;
import java.util.concurrent.CompletableFuture;

@Service
public class Service2 {
    public CompletableFuture<String> method2() {
        return CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1500);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return "Method2执行结果";
        });
    }
}

@Service
public class Service3 {
    public CompletableFuture<Long> method3() {
        return CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(900);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return 1000L;
        });
    }
}

4. 主Service的@Async方法

import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import java.util.concurrent.CompletableFuture;

@Service
public class MainService {
    private final Service1 service1;
    private final Service2 service2;
    private final Service3 service3;

    // 构造注入
    public MainService(Service1 service1, Service2 service2, Service3 service3) {
        this.service1 = service1;
        this.service2 = service2;
        this.service3 = service3;
    }

    @Async
    public CompletableFuture<Void> mainAsyncMethod() {
        // 启动三个顶级任务并行执行
        CompletableFuture<Service1.CombinedResult> t1Future = service1.method1();
        CompletableFuture<String> t2Future = service2.method2();
        CompletableFuture<Long> t3Future = service3.method3();

        // 等待所有顶级任务完成，按需处理最终结果
        return CompletableFuture.allOf(t1Future, t2Future, t3Future)
                .thenAccept(v -> {
                    Service1.CombinedResult t1Res = t1Future.join();
                    String t2Res = t2Future.join();
                    Long t3Res = t3Future.join();
                    // 此处可添加结果打印、持久化等逻辑
                    System.out.println("T1合并结果: " + t1Res);
                    System.out.println("T2结果: " + t2Res);
                    System.out.println("T3结果: " + t3Res);
                });
    }
}

核心注意事项 #

• 子任务并行的关键：必须通过CompletableFuture.supplyAsync/runAsync将子任务提交到线程池，直接调用普通方法仍是同步执行。
• 线程池优化：务必自定义线程池，避免使用Spring默认的SimpleAsyncTaskExecutor（每次创建新线程，性能损耗大）。
• 异常处理：实际场景中要为CompletableFuture添加异常处理（如exceptionally或handle方法），防止异常被静默吞掉。
• 返回值规范：Method1必须返回CompletableFuture，确保主方法的顶级任务能等待其内部所有子任务完成。

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