文章目录
- 前置
- 自定义线程池
- 使用 CompletableFuture 创建异步任务
前置
来自 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 "默认结果"; // 返回默认值
});