面试题：Java 中 CompletableFuture 异步任务回调参数处理之性能优化与原理探究

Question

在高并发场景下使用 CompletableFuture 时，如何优化性能？比如减少线程创建开销、合理利用线程池等。请阐述 CompletableFuture 内部是如何处理异步任务回调参数的，包括线程调度、数据传递等方面的原理，并结合实际应用场景说明如何根据这些原理进行性能调优。

Answer 1

优化 CompletableFuture 性能的方法

1. 减少线程创建开销：使用自定义线程池，避免每次异步任务都创建新线程。例如，创建一个固定大小的线程池 ExecutorService executor = Executors.newFixedThreadPool(10);，然后在使用 CompletableFuture 时通过 CompletableFuture.supplyAsync(() -> { /* 任务代码 */ }, executor) 这样的方式提交任务，复用线程，减少线程创建销毁的开销。
2. 合理利用线程池：根据系统资源和任务类型来配置线程池参数。对于 I/O 密集型任务，线程池大小可以设置得较大，以充分利用等待 I/O 的时间；对于 CPU 密集型任务，线程池大小应接近 CPU 核心数，避免过多线程竞争 CPU 资源。例如，通过 ThreadPoolExecutor 自定义线程池并根据任务特性调整 corePoolSize、maximumPoolSize 等参数。

CompletableFuture 处理异步任务回调参数的原理

1. 线程调度：
   • 默认线程调度：当使用 CompletableFuture 的静态方法如 supplyAsync 或 runAsync 且不传入自定义线程池时，会使用 ForkJoinPool.commonPool() 来调度任务。ForkJoinPool 采用工作窃取算法，线程在处理完自己队列的任务后，可以窃取其他线程队列的任务，提高线程利用率。
   • 自定义线程调度：当传入自定义线程池时，任务由自定义线程池的线程来执行。例如使用 ThreadPoolExecutor 线程池，按照其自身的线程调度策略（如核心线程优先执行、队列满后创建新线程等）来执行任务。
2. 数据传递：
   • 链式调用的数据传递：在 CompletableFuture 的链式调用中，如 CompletableFuture.supplyAsync(() -> "result1").thenApply(s -> s + " appended").thenAccept(System.out::println)，前一个阶段的结果作为参数传递给后续的回调方法。thenApply 方法接收前一个 CompletableFuture 的结果，并返回一个新的 CompletableFuture，其结果是经过 Function 处理后的结果。thenAccept 方法只接收前一个 CompletableFuture 的结果并执行消费操作，不返回新的 CompletableFuture 结果。
   • 组合操作的数据传递：对于组合多个 CompletableFuture 的操作，如 CompletableFuture.allOf 或 CompletableFuture.anyOf，allOf 等待所有 CompletableFuture 完成，anyOf 等待任意一个 CompletableFuture 完成。它们会收集所有或部分 CompletableFuture 的结果（取决于操作），并可以在后续的回调中处理这些结果。例如，CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "result1"); CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "result2"); CompletableFuture<Void> allFuture = CompletableFuture.allOf(future1, future2); allFuture.thenRun(() -> { String result1 = future1.join(); String result2 = future2.join(); System.out.println(result1 + " " + result2); });

结合实际应用场景的性能调优

1. 电商场景 - 商品信息聚合：在电商系统中，获取商品信息可能涉及从多个数据源获取数据，如商品基本信息、库存信息、价格信息等。每个数据源的获取可以是一个异步任务。可以使用自定义线程池，因为不同数据源的获取任务可能是 I/O 密集型（如从数据库读取），设置较大的线程池大小以提高并发度。例如，创建一个线程池 ExecutorService productInfoExecutor = Executors.newFixedThreadPool(20);，然后使用 CompletableFuture 来异步获取各个数据源的信息，如 CompletableFuture<String> basicInfoFuture = CompletableFuture.supplyAsync(() -> getBasicProductInfo(), productInfoExecutor); CompletableFuture<String> stockInfoFuture = CompletableFuture.supplyAsync(() -> getStockInfo(), productInfoExecutor); CompletableFuture<String> priceInfoFuture = CompletableFuture.supplyAsync(() -> getPriceInfo(), productInfoExecutor); CompletableFuture<Void> allFuture = CompletableFuture.allOf(basicInfoFuture, stockInfoFuture, priceInfoFuture); allFuture.thenRun(() -> { String basicInfo = basicInfoFuture.join(); String stockInfo = stockInfoFuture.join(); String priceInfo = priceInfoFuture.join(); // 聚合商品信息 }); 这样可以充分利用线程池，减少线程创建开销，提高性能。
2. 搜索引擎场景 - 多索引查询：在搜索引擎中，可能需要同时查询多个索引来获取相关结果。每个索引查询可以是一个异步任务。由于索引查询可能是 CPU 密集型任务，根据服务器 CPU 核心数设置合适大小的线程池。例如，服务器有 8 个 CPU 核心，创建一个固定大小为 8 的线程池 ExecutorService searchExecutor = Executors.newFixedThreadPool(8);，使用 CompletableFuture 来异步执行各个索引查询任务，如 CompletableFuture<List<SearchResult>> index1Future = CompletableFuture.supplyAsync(() -> searchIndex1(), searchExecutor); CompletableFuture<List<SearchResult>> index2Future = CompletableFuture.supplyAsync(() -> searchIndex2(), searchExecutor); CompletableFuture<Void> allFuture = CompletableFuture.allOf(index1Future, index2Future); allFuture.thenRun(() -> { List<SearchResult> result1 = index1Future.join(); List<SearchResult> result2 = index2Future.join(); // 合并搜索结果 }); 依据 CompletableFuture 的线程调度和数据传递原理，合理利用线程池，提高搜索性能。