CompletableFuture异步编程【四】

贺公子之数据科学与艺术

发布于 2025-12-18 08:32:16

1360

CompletableFuture异步编程

基本概念

CompletableFuture是Java 8引入的一个强大的异步编程工具，它实现了Future和CompletionStage接口，提供了丰富的API来支持异步操作和函数式编程。相比传统的Future，CompletableFuture具有以下优势：

可以手动完成Future（设置返回值或异常）
支持链式调用和组合多个异步操作
提供了异常处理机制
支持回调函数
可以与Java 8的Stream API和Lambda表达式无缝配合

核心方法

创建CompletableFuture

静态工厂方法：

// 创建一个已完成的CompletableFuture
CompletableFuture<String> completedFuture = CompletableFuture.completedFuture("Hello");

// 创建一个异步执行的CompletableFuture
CompletableFuture<Void> runAsyncFuture = CompletableFuture.runAsync(() -> {
    // 无返回值的异步任务
});

CompletableFuture<String> supplyAsyncFuture = CompletableFuture.supplyAsync(() -> {
    // 有返回值的异步任务
    return "Result";
});

自定义Executor：

Executor executor = Executors.newFixedThreadPool(10);
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 使用自定义线程池执行任务
    return "Custom Executor Result";
}, executor);

结果处理

阻塞获取结果：

// 阻塞等待结果
String result = future.get();

// 带超时的阻塞等待
String result = future.get(1, TimeUnit.SECONDS);

非阻塞回调：

future.thenAccept(result -> {
    // 处理结果
    System.out.println("Result: " + result);
});

future.thenApply(result -> {
    // 转换结果
    return result.length();
});

future.whenComplete((result, exception) -> {
    // 完成时处理（无论成功或失败）
    if (exception != null) {
        exception.printStackTrace();
    } else {
        System.out.println(result);
    }
});

组合多个Future

顺序组合：

CompletableFuture.supplyAsync(() -> "Hello")
    .thenApply(s -> s + " World")
    .thenApply(String::toUpperCase)
    .thenAccept(System.out::println);

并行组合：

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "Hello");
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "World");

future1.thenCombine(future2, (s1, s2) -> s1 + " " + s2)
       .thenAccept(System.out::println);

多任务组合：

CompletableFuture<Void> allFutures = CompletableFuture.allOf(future1, future2, future3);
allFutures.thenRun(() -> {
    // 所有任务完成后执行
});

CompletableFuture<Object> anyFuture = CompletableFuture.anyOf(future1, future2, future3);
anyFuture.thenAccept(result -> {
    // 任一任务完成后执行
});

异常处理

显式处理：

future.exceptionally(ex -> {
    System.out.println("Exception occurred: " + ex.getMessage());
    return "default value";
});

链式处理：

future.handle((result, ex) -> {
    if (ex != null) {
        return "recovery value";
    }
    return result;
});

实际应用场景

并行调用多个服务：

CompletableFuture<User> userFuture = CompletableFuture.supplyAsync(() -> userService.getUser(userId));
CompletableFuture<Order> orderFuture = CompletableFuture.supplyAsync(() -> orderService.getOrder(orderId));

userFuture.thenCombine(orderFuture, (user, order) -> {
    // 合并用户和订单信息
    return new UserOrder(user, order);
}).thenAccept(this::processUserOrder);

异步流水线处理：

CompletableFuture.supplyAsync(() -> fetchDataFromRemote())
    .thenApply(data -> transformData(data))
    .thenApply(transformedData -> saveToDatabase(transformedData))
    .thenRun(() -> sendNotification())
    .exceptionally(ex -> {
        log.error("Pipeline failed", ex);
        return null;
    });

超时处理：

CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(2000);
            return "Result";
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    })
    .completeOnTimeout("Timeout Result", 1, TimeUnit.SECONDS)
    .thenAccept(System.out::println);