一、CompletableFuture 核心定义

1. 基础概念

CompletableFuture 是 JDK 8 引入的异步编程工具类，实现了 Future 和 CompletionStage 接口，核心解决传统 Future 的两大痛点：

• 传统 Future 仅能通过 get() 阻塞获取结果，无法异步回调；
• 传统 Future 不支持多个任务的串联、并行组合。

CompletableFuture 提供了链式调用、异步回调、任务编排能力，是 Java 异步编程的核心工具，可轻松实现“异步执行任务 + 结果回调 + 多任务组合”的复杂逻辑。

2. 核心定位

特性	传统 Future	CompletableFuture
结果获取	仅阻塞 get()	支持异步回调（thenAccept/whenComplete）
任务组合	无	支持串联（thenApply）、并行（allOf）
异常处理	需手动捕获	内置 exceptionally/handle 等方法
线程池指定	仅部分场景支持	所有异步方法可指定自定义线程池

二、CompletableFuture 核心使用

1. 核心 API 分类

CompletableFuture 的 API 可分为 5 大类，覆盖异步任务执行、结果处理、任务组合、异常处理等全场景：

（1）异步执行任务（基础）

方法	作用	适用场景
runAsync(Runnable)	执行无返回值异步任务	纯执行操作（如推送、日志）
runAsync(Runnable, Executor)	指定线程池执行无返回值任务	管控并发的异步操作（如大屏推送）
supplyAsync(Supplier)	执行有返回值异步任务	查询数据、计算结果
supplyAsync(Supplier, Executor)	指定线程池执行有返回值任务	管控并发的查询/计算

基础示例：

// 1. 无返回值异步任务（指定自定义线程池）
CompletableFuture.runAsync(() -> {
    // 执行无返回值操作：如大屏WebSocket推送
    largeScreenService.pustMsgByType(vo);
}, WebSocketThreadPool.webSocketExecutor);

// 2. 有返回值异步任务（指定自定义线程池）
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 执行有返回值操作：查询大屏数据
    return largeScreenService.queryScreenData("screen002");
}, WebSocketThreadPool.webSocketExecutor);

（2）结果处理（回调）

方法	作用	入参/返回值
thenApply(Function)	接收上一步结果，返回新结果	入参：上一步结果；返回：新结果
thenAccept(Consumer)	接收上一步结果，无返回值	入参：上一步结果；无返回
thenRun(Runnable)	无参数，无返回值	仅执行操作

示例：

// 查询数据后处理结果，再推送
CompletableFuture.supplyAsync(() -> {
    // 步骤1：查询大屏数据（有返回值）
    return largeScreenService.queryScreenData("screen002");
}, WebSocketThreadPool.webSocketExecutor)
.thenApply(data -> {
    // 步骤2：处理数据（转换为JSON）
    return JSON.toJSONString(data);
})
.thenAccept(message -> {
    // 步骤3：推送数据（使用处理后的结果）
    largeScreenService.pushMessage(message, "screen002");
});

（3）多任务组合

方法	作用	适用场景
allOf(CompletableFuture...)	等待所有任务完成	并行执行多个推送任务
anyOf(CompletableFuture...)	等待任意一个任务完成	多数据源查询（取最快结果）

示例：

// 并行推送多个大屏模块，全部完成后记录日志
CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> {
    largeScreenService.pustMsgByType(vo1);
}, WebSocketThreadPool.webSocketExecutor);

CompletableFuture<Void> future2 = CompletableFuture.runAsync(() -> {
    largeScreenService.pustMsgByType(vo2);
}, WebSocketThreadPool.webSocketExecutor);

// 等待所有任务完成
CompletableFuture.allOf(future1, future2).thenRun(() -> {
    log.info("所有大屏模块推送完成");
});

（4）异常处理

方法	作用	特点
exceptionally(Function)	捕获异常，返回默认值	仅处理异常，正常结果不触发
whenComplete(BiConsumer)	成功/失败都执行	可同时处理结果和异常
handle(BiFunction)	成功/失败都执行，返回新结果	可处理结果+异常，返回默认值

示例：

CompletableFuture.runAsync(() -> {
    // 可能抛出异常的推送逻辑
    largeScreenService.pustMsgByType(vo);
}, WebSocketThreadPool.webSocketExecutor)
// 方式1：仅捕获异常，返回默认值（无返回值任务用whenComplete更合适）
.exceptionally(e -> {
    log.error("推送失败", e);
    return null;
})
// 方式2：成功/失败都执行
.whenComplete((unused, e) -> {
    if (e != null) {
        log.error("推送失败", e);
    } else {
        log.info("推送成功");
    }
});

（5）结果获取（慎用阻塞方法）

方法	作用	注意事项
get()	阻塞获取结果	会抛出检查异常，易导致线程阻塞
get(long, TimeUnit)	超时阻塞获取结果	避免无限阻塞
join()	阻塞获取结果	抛出运行时异常，无需手动捕获
getNow(T)	非阻塞获取结果	未完成则返回默认值

示例：

// 非阻塞获取结果（推荐）
String result = future.getNow("默认数据");

// 超时阻塞获取结果（避免无限阻塞）
try {
    String result = future.get(3, TimeUnit.SECONDS);
} catch (TimeoutException e) {
    log.error("获取结果超时", e);
}

2. 线程池使用原则

• 必须指定自定义线程池：runAsync/supplyAsync 若不指定线程池，会使用 JDK 全局的 ForkJoinPool.commonPool()，该线程池被所有异步任务共享，易被其他任务阻塞；
• 线程池按需定义：不同业务（如推送、入库、查询）建议使用独立线程池，避免“一个任务阻塞导致全链路卡顿”；
• 核心参数适配场景：
  • 低并发场景（如日常推送）：核心线程数 2-5，最大线程数 5-10；
  • 高并发场景（如批量入库）：核心线程数 10-20，最大线程数 50，队列大小 500+。

三、CompletableFuture 典型使用场景

1. 异步执行无返回值任务（核心场景）

适用：纯执行操作（如 WebSocket 推送、日志记录、第三方接口调用）
示例：大屏 WebSocket 异步推送

// 自定义线程池
public static final ExecutorService webSocketExecutor = new ThreadPoolExecutor(
        2, 5, 60L, TimeUnit.SECONDS,
        new LinkedBlockingQueue<>(200),
        new ThreadFactoryBuilder().setNameFormat("websocket-pool-%d").build(),
        new ThreadPoolExecutor.AbortPolicy());

// 异步推送
CompletableFuture.runAsync(() -> {
    try {
        LargeScreenService service = SpringContextUtil.getBean(LargeScreenService.class);
        service.pustMsgByType(PushMessageVO.builder()
                .type("RENT_TYPE_9")
                .largeScreenType("SCREEN_002")
                .build());
    } catch (Exception e) {
        log.error("推送失败", e);
    }
}, webSocketExecutor);

2. 异步查询 + 结果处理

适用：查询数据后立即处理结果（如查询大屏数据→转换格式→推送）
示例：

CompletableFuture.supplyAsync(() -> {
    // 异步查询大屏数据
    return largeScreenService.queryScreenData("screen004");
}, webSocketExecutor)
.thenApply(data -> {
    // 转换数据格式
    return JSON.toJSONString(data);
})
.thenAccept(message -> {
    // 推送处理后的数据
    largeScreenService.pushMessage(message, "screen004");
})
.exceptionally(e -> {
    log.error("查询+推送失败", e);
    return null;
});

3. 多任务并行执行

适用：同时执行多个独立任务（如批量推送多个大屏模块、多数据源查询）
示例：

// 定义多个推送任务
List<PushMessageVO> voList = Arrays.asList(
    PushMessageVO.builder().type("RENT_TYPE_1").largeScreenType("SCREEN_002").build(),
    PushMessageVO.builder().type("RENT_TYPE_2").largeScreenType("SCREEN_002").build(),
    PushMessageVO.builder().type("OVERALL_TYPE_4").largeScreenType("SCREEN_004").build()
);

// 并行执行所有任务
List<CompletableFuture<Void>> futures = voList.stream()
    .map(vo -> CompletableFuture.runAsync(() -> {
        LargeScreenService service = SpringContextUtil.getBean(LargeScreenService.class);
        service.pustMsgByType(vo);
    }, webSocketExecutor))
    .collect(Collectors.toList());

// 等待所有任务完成
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
    .whenComplete((unused, e) -> {
        if (e != null) {
            log.error("部分推送失败", e);
        } else {
            log.info("全部推送完成");
        }
    });

4. 任务串联执行

适用：按顺序执行依赖任务（如“查询数据→校验数据→入库→推送结果”）
示例：

CompletableFuture.supplyAsync(() -> {
    // 步骤1：查询数据
    return dataService.queryData("2025-02");
}, executor)
.thenApply(data -> {
    // 步骤2：校验数据
    if (data == null) {
        throw new RuntimeException("数据为空");
    }
    return data;
})
.thenAccept(data -> {
    // 步骤3：入库
    dataService.insert(data);
    // 步骤4：推送结果
    largeScreenService.pushMessage(JSON.toJSONString(data), "screen004");
})
.exceptionally(e -> {
    log.error("任务执行失败", e);
    return null;
});

5. 超时控制

适用：避免异步任务无限阻塞（如第三方接口调用、大数据量计算）
示例：

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 调用第三方支付接口
    return thirdPartyService.pay(orderId);
}, executor);

// 超时控制：3秒未返回则触发超时处理
CompletableFuture<String> timeoutFuture = future.orTimeout(3, TimeUnit.SECONDS)
    .exceptionally(e -> {
        log.error("支付接口调用超时", e);
        return "超时默认值";
    });

// 处理结果
timeoutFuture.thenAccept(result -> {
    log.info("支付结果：{}", result);
});

四、CompletableFuture 核心优点

1. 异步化解耦，提升系统响应速度

• 主线程无需等待异步任务完成，可立即响应新请求，解决“同步执行导致的请求堆积、响应延迟”；
• 例如：大屏推送任务异步执行后，数据更新接口的响应时间从 500ms 降至 50ms。

2. 强大的任务编排能力

• 支持“串联、并行、任意一个完成”等多任务组合方式，轻松实现复杂业务逻辑；
• 相比传统线程池 + CountDownLatch，代码更简洁、可读性更高。

3. 完善的异常处理机制

• 内置 exceptionally/whenComplete/handle 等方法，避免异步任务“静默失败”；
• 可统一处理异常，无需在每个异步任务中手动 try-catch（推荐仍加局部 try-catch 精准定位）。

4. 灵活的线程池管控

• 可指定自定义线程池，精准控制异步任务的并发数，避免系统过载；
• 相比 new Thread() 手动创建线程，线程池可复用线程，降低资源消耗。

5. 非阻塞式回调，避免线程阻塞

• 结果回调方法（thenAccept/thenApply）在任务完成后自动执行，无需手动调用 get() 阻塞线程；
• 结合 orTimeout 可实现超时控制，避免无限阻塞。

五、使用注意事项（避坑指南）

1. 必须指定自定义线程池

• 禁止使用默认的 ForkJoinPool.commonPool()，该线程池是全局共享的，易被其他任务阻塞；
• 例如：若其他业务占用了 commonPool 的所有线程，大屏推送任务会排队等待，导致延迟。

2. 异常处理不可少

• 异步任务的异常不会主动抛出到主线程，必须通过 exceptionally/whenComplete 捕获；
• 例如：推送任务抛出空指针异常，若未捕获，日志中无任何记录，无法定位问题。

3. 避免过度异步化

• 简单、快速的操作（如内存数据处理）无需异步，线程切换的开销可能大于异步收益；
• 建议仅对“耗时操作”（如数据库查询、第三方接口调用、大数据量计算）使用异步。

4. 注意异步线程的 Spring Bean 注入问题

• 异步线程脱离 Spring 容器管理，无法直接 @Resource/@Autowired 注入 Bean；
• 解决方案：通过 SpringContextUtil.getBean() 获取 Bean，或在主线程注入后传递到异步线程。

5. 避免任务嵌套过深

• 异步任务嵌套过深会导致代码可读性差、异常排查困难；
• 建议拆分嵌套任务为独立的 CompletableFuture，通过 allOf/thenCompose 组合。

六、总结

CompletableFuture 是 Java 异步编程的“瑞士军刀”，核心解决了传统 Future 的回调和组合痛点，适配所有“耗时操作、非核心链路、需要控制并发”的场景（如 WebSocket 推送、第三方接口调用、批量数据处理）。

核心使用范式：

CompletableFuture.runAsync(() -> {
    // 耗时操作（加try-catch）
}, 自定义线程池)
.whenComplete((unused, e) -> {
    // 成功/失败处理
});

在实际项目中，结合“自定义线程池 + CompletableFuture”，可大幅提升系统的响应速度和稳定性，是高并发场景下的首选异步编程方案。