java CompletableFuture创建异步任务(Completable异步+ExecutorService线程池)

前置

来自 import java.util.concurrent.CompletableFuture;

自定义线程池

推荐根据业务需求配置

ExecutorService pool = new ThreadPoolExecutor(10, // 核心线程数20, // 最大线程数60L, TimeUnit.SECONDS, // 空闲线程存活时间new LinkedBlockingQueue<>(100), // 任务队列new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略
);

如果使用自定义线程池，确保在任务完成后关闭线程池（executorService.shutdown()）

使用 CompletableFuture 创建异步任务

创建异步
有返回值异步任务,避免使用默认的 ForkJoinPool.commonPool()，尤其是在高并发场景下

// 有返回值的异步任务
// 默认使用 ForkJoinPool.commonPool()，也可自定义线程池
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {// 模拟耗时操作System.out.println("异步任务执行中...");try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}return "任务结果";
});// 有返回值的异步任务
// 自定义线程池
ExecutorService customPool = Executors.newFixedThreadPool(4);
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {...}, customPool);// 后续通过.join 获取返回结果 T
future.join();

无返回值的异步任务

CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {System.out.println("无返回值的异步任务执行中...");Thread.sleep(500);
});

处理异步返回结果(可链式处理)

// java 类似 map 映射处理 result 结果
CompletableFuture<String> resultFuture = future.thenApply(result -> {return "转换后的结果：" + result;
});// 映射处理
CompletableFuture<String> composedFuture = future.thenCompose(result -> {// 根据前一个结果继续执行新的异步任务return anotherAsyncTask(result);
});// thenRun()// thenAccept()

allOf 和 anyOf
allOf 全部结束后发消息

// allOf 不是强等待他们结束，而是仅仅是他们结束后给个通知
CompletableFuture<Void> allFutures = CompletableFuture.allOf(future1, future2);
allFutures.thenRun(() -> {System.out.println("所有任务已完成");
});

anyOf 任意结束后发消息

CompletableFuture<Object> anyFuture = CompletableFuture.anyOf(future1, future2);
anyFuture.thenAccept(result -> {System.out.println("任一任务完成，结果为: " + result);
});

等待线程执行完
等待单个线程执行完，使用 join 或者 get 获取 future 返回结果

// 创建一个异步任务
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {try {Thread.sleep(1000); // 模拟耗时操作} catch (InterruptedException e) {e.printStackTrace();}return "任务结果";
});// 主线程等待异步任务完成
String result = future.join(); // 或 future.get();
System.out.println("主线程获取结果: " + result);
System.out.println("主线程继续执行...");

等待多个线程都执行，使用 join 或者 get 获取 allFuture 返回结果

// 创建多个异步任务
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}return "任务1结果";
});
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {try {Thread.sleep(2000);} catch (InterruptedException e) {e.printStackTrace();}return "任务2结果";
});// 等待所有任务完成
CompletableFuture<Void> allFutures = CompletableFuture.allOf(future1, future2);
allFutures.join(); // 或 allFutures.get();// 收集结果
String result1 = future1.join();
String result2 = future2.join();

设置超时时间

// orTimeout() 方法设置超时时间，避免无限等待
future.orTimeout(3, TimeUnit.SECONDS).exceptionally(ex -> {System.err.println("任务超时: " + ex.getMessage());return "默认结果";
});

异常处理

// exceptionally 捕获异常返回默认值
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {if (Math.random() > 0.5) throw new RuntimeException("任务失败");return "成功结果";
}).exceptionally(ex -> {System.err.println("异常已捕获: " + ex.getMessage());return "默认结果"; // 返回默认值
});