前言

在项目开发中，经常会遇到一个问题：在一个后端接口里，往往会进行多项耗时任务（相互之间独立，没有依赖）的操作，如：

• 需要从不同的外部接口获取不同的数据，做融合；
• 请求外部接口数据的同时，还需要读取数据库；
• 等等

如果在一个请求的主线程里，串行做这些任务操作，会导致响应时间的线性叠加，极有可能导致不符合要求，如图1：

图1

那么，对这些耗时任务进行并行操作，从而使得：响应时间 约等于 耗时最大的任务处理时间，这样可以大大降低系统的响应时间，如图2：

图2

Future

Future类型，其实就是一个未来任务的返回对象，或者说是子线程的返回对象（通过线程池方式分配子线程）

 ExecutorService executor = Executors.newFixedThreadPool(4); 
// 定义任务:
Callable<String> task = new Task();
// 提交任务并获得Future:
Future<String> future = executor.submit(task);
// 从Future获取异步执行返回的结果:
String result = future.get(); // 可能阻塞
复制代码  

可以看到，通过线程池的方式创建子线程后， executor.submit() 返回的是一个Future对象，通过 future.get() 方法来获得该子任务的运行结果。需要注意的是，这个操作是阻塞的，也就是说，如果这个子任务没有运行结束，主线程会一直block在改行，直到子任务完成。

一个 Future<V> 接口表示一个未来可能会返回的结果，它定义的方法有：

• get()：获取结果（可能会等待）
• get(long timeout, TimeUnit unit)：获取结果，但只等待指定的时间；
• cancel(boolean mayInterruptIfRunning)：取消当前任务；
• isDone()：判断任务是否已完成。

CompletableFuture

当需要判断图2中的所有task是否完成时，如果采用Future，则需要:

• 调用 future.get() 获取运行结果，
• 或者 轮询 future.isDone() 方法直到返回 true 。

无论哪种方法，都是在主线程里调用，且会阻塞主线程。

以上痛点，从Java 8开始引入了 CompletableFuture 方法。主要新增的功能有：

• thenAccept(): 当task正常完成后， 回调 调用 .thenAccept() 方法
• Exception ally(): 当task出现异常时，回调调用 .exceptionally() 方法
• anyOf(): 当所有的task中，只要有一个task完成，则主线程继续往下走，可以使用 .anyOf() 方法
• allOf(): 所有的task均完成后，则主线程继续往下走
• supplyAsync(): 异步执行，有返回值
• runAsync(): 异步执行，无返回值

针对图2，需要所有task都完成后，再执行后续操作，就可以用 allOf() 方法:

 CompletableFuture.allOf(task1, task2, ..., taskn).join();  

注意：CompletableFuture的命名规则：

• xxx()：表示该方法将继续在已有的线程中执行；
• xxxAsync()：表示将异步在线程池中执行，即可以异步执行。

基于CompletableFuture+线程池的实现

线程池配置类

 @ Configuration 
@Slf4j
@EnableAsync
public class ExecutorConfig {
    @Bean
    public Executor asyncExecutor() {
        log.info("start async executor");
        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
//        配置核心线程数
        threadPoolTaskExecutor.setCorePoolSize(ThreadPoolConstant.CORE_POOL_SIZE);
//        配置最大线程数
        threadPoolTaskExecutor.setMaxPoolSize(ThreadPoolConstant.MAX_POOL_SIZE);
//        配置队列大小
        threadPoolTaskExecutor.setQueueCapacity(ThreadPoolConstant.QUEUE_CAPACITY);
//        配置 线程池 中线程的名称前缀
        threadPoolTaskExecutor.setThreadNamePrefix(ThreadPoolConstant.THREAD_NAME_PREFIX);
//   HelloWorldServiceImpl     rejection-policy: 当pool已经达到max size时，如何处理新任务：
//        CallerRunsPolicy: 不在新线程中执行任务，而是由调用者所在的线程来执行；
//        AbortPolicy: 拒绝执行新任务，并抛出RejectedExecutionException异常；
//        DiscardPolicy：丢弃当前将要加入队列的任务；
//        DiscardOldestPolicy：丢弃任务队列中最旧的任务；
        threadPoolTaskExecutor.setRejectedExecutionHandler(
                new ThreadPoolExecutor.CallerRunsPolicy()
        );
        threadPoolTaskExecutor.initialize();
        return threadPoolTaskExecutor;
    }
}  

异步服务与服务实现

 public interface AsyncService {
    @Async("asyncExecutor")
    CompletableFuture<String> getResponseFromCp(QueryTrainInfoDetailReqDTOWithType queryTrainInfoDetailReqDTOWithType, int queryType);
}  

 @Service
public class AsyncServiceImpl implements AsyncService {
    @Autowired
    CustomProps customProps;
    @Autowired
    RestTemplate restTemplate;
    @ Override 
    public CompletableFuture<String> getResponseFromCp(QueryTrainInfoDetailReqDTOWithType queryTrainInfoDetailReqDTOWithType, int queryType) {
        return CompletableFuture
                .completedFuture(
                        FactoryUtil
                                .createFactory(customProps, null, restTemplate)
                                .obtainData(queryTrainInfoDetailReqDTOWithType.setQueryType(queryType), String.class)
                );
    }
}  

业务代码中调用异步服务接口

 ...
    @Autowired
    AsyncService asyncService;

    @Override
    public ReturnData qTrainInfoDetail(QueryTrainInfoDetailReqDTO queryTrainInfoDetailReqDTO) {
        QueryTrainInfoDetailReqDTOWithType queryTrainInfoDetailReqDTOWithType = new QueryTrainInfoDetailReqDTOWithType().setQueryTrainInfoDetailReqDTO(queryTrainInfoDetailReqDTO);
        CompletableFuture<String> fromCpFirstReq = asyncService.getResponseFromCp(queryTrainInfoDetailReqDTOWithType, 1);
        CompletableFuture<String> fromCpSecondReq = asyncService.getResponseFromCp(queryTrainInfoDetailReqDTOWithType, 2);
        CompletableFuture.allOf(fromCpFirstReq, fromCpSecondReq).join(); //阻塞直到当第一次请求和第二次请求都完成
    }
...