Java异步编程：CompletableFuture使用详解

Java中实现异步编程的核心在于避免主线程阻塞，提高响应速度和吞吐量，主要通过CompletableFuture实现。1. CompletableFuture提供supplyAsync()和runAsync()创建异步任务，前者用于有返回值的任务，后者用于无返回值的任务；2. 异常处理可通过exceptionally()返回默认值或handle()统一处理结果与异常；3. 组合多个异步任务可使用thenApply()转换结果、thenCompose()链式依赖任务、thenCombine()合并两个独立任务的结果；4. 线程池选择上，默认使用ForkJoinPool.commonPool()适合CPU密集型任务，I/O密集型或需控制并发时应自定义线程池并通过参数传入；5. 避免阻塞应避免调用get()方法，改用thenAccept()或thenApplyAsync()将后续操作放入异步线程执行，确保非阻塞特性。

Java中实现异步编程，核心在于避免主线程阻塞，提高程序响应速度和吞吐量。CompletableFuture是Java 8引入的强大的异步编程工具，它提供了灵活的方式来处理异步任务的结果，组合多个异步操作，以及处理异常。

CompletableFuture的使用方法

CompletableFuture提供了多种静态方法来创建异步任务，例如supplyAsync()、runAsync()。supplyAsync()用于执行有返回值的任务，而runAsync()用于执行没有返回值的任务。

// 执行一个有返回值的异步任务
CompletableFuture future = CompletableFuture.supplyAsync(() -> {
    // 模拟耗时操作
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new RuntimeException(e);
    }
    return "Hello, Async!";
});

// 执行一个没有返回值的异步任务
CompletableFuture futureVoid = CompletableFuture.runAsync(() -> {
    // 模拟耗时操作
    try {
        Thread.sleep(500);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new RuntimeException(e);
    }
    System.out.println("Async task completed!");
});

// 获取异步任务的结果 (会阻塞直到结果可用)
try {
    String result = future.get();
    System.out.println(result);
} catch (Exception e) {
    e.printStackTrace();
}

关键在于，supplyAsync和runAsync默认使用ForkJoinPool.commonPool()线程池。如果需要自定义线程池，可以传入Executor参数。 这点需要注意，避免所有异步任务都挤在同一个公共线程池里。

如何处理异步任务的异常？

CompletableFuture提供了exceptionally()、handle()等方法来处理异常。exceptionally()允许你提供一个函数，在发生异常时返回一个默认值。handle()则允许你同时处理正常结果和异常情况。

CompletableFuture futureWithException = CompletableFuture.supplyAsync(() -> {
    // 模拟可能抛出异常的操作
    if (Math.random() > 0.5) {
        throw new RuntimeException("Something went wrong!");
    }
    return "Success!";
}).exceptionally(ex -> {
    System.err.println("Exception occurred: " + ex.getMessage());
    return "Default Value"; // 返回默认值
});

try {
    String result = futureWithException.get();
    System.out.println("Result: " + result);
} catch (Exception e) {
    e.printStackTrace();
}

handle方法更强大，它接收一个BiFunction，可以根据结果和异常进行更复杂的处理。 个人觉得，使用handle可以避免多层嵌套的try-catch，代码更简洁。

CompletableFuture如何组合多个异步任务？

CompletableFuture提供了thenApply()、thenCompose()、thenCombine()等方法来组合多个异步任务。thenApply()用于对异步任务的结果进行转换。thenCompose()用于将一个异步任务的结果传递给另一个异步任务。thenCombine()用于合并两个独立的异步任务的结果。

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

// 使用 thenCombine 合并两个 Future 的结果
CompletableFuture combinedFuture = future1.thenCombine(future2, (s1, s2) -> s1 + ", " + s2);

// 使用 thenApply 对结果进行转换
CompletableFuture transformedFuture = combinedFuture.thenApply(String::toUpperCase);

try {
    String result = transformedFuture.get();
    System.out.println(result); // 输出: HELLO, WORLD
} catch (Exception e) {
    e.printStackTrace();
}

thenCompose适用于一个任务的完成依赖于另一个任务的结果的场景。 例如，从数据库查询用户信息，然后根据用户信息查询订单信息。

CompletableFuture中的线程池选择有什么讲究？

选择合适的线程池至关重要。 默认的ForkJoinPool.commonPool()适用于CPU密集型任务，但如果你的任务是I/O密集型，或者需要控制并发度，那么自定义线程池是更好的选择。

// 创建一个固定大小的线程池
ExecutorService executor = Executors.newFixedThreadPool(10);

CompletableFuture futureWithExecutor = CompletableFuture.supplyAsync(() -> {
    // 模拟耗时操作
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new RuntimeException(e);
    }
    return "Hello from custom executor!";
}, executor);

try {
    String result = futureWithExecutor.get();
    System.out.println(result);
} catch (Exception e) {
    e.printStackTrace();
} finally {
    executor.shutdown(); // 关闭线程池
}

需要注意的是，使用自定义线程池后，一定要记得关闭线程池，释放资源。 否则，可能会导致程序无法正常退出。

如何避免CompletableFuture的阻塞？

虽然get()方法可以获取异步任务的结果，但它会阻塞当前线程。 为了避免阻塞，可以使用thenAccept()、thenApplyAsync()等方法。 thenAccept()在异步任务完成后执行一个Consumer，而thenApplyAsync()则在另一个异步任务中执行转换操作。

CompletableFuture futureNonBlocking = CompletableFuture.supplyAsync(() -> {
    // 模拟耗时操作
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new RuntimeException(e);
    }
    return "Non-blocking result!";
});

// 使用 thenAccept 在异步任务完成后执行一个 Consumer
futureNonBlocking.thenAccept(result -> {
    System.out.println("Result (non-blocking): " + result);
});

// 使用 thenApplyAsync 在另一个异步任务中执行转换操作
CompletableFuture transformedFutureNonBlocking = futureNonBlocking.thenApplyAsync(String::toUpperCase);

transformedFutureNonBlocking.thenAccept(result -> {
    System.out.println("Transformed result (non-blocking): " + result);
});

// 注意：这里不要调用 get()，否则就阻塞了

避免阻塞的关键在于，不要使用get()方法，而是使用thenAccept、thenApplyAsync等方法，将后续操作也放到异步线程中执行。 这样才能真正实现非阻塞的异步编程。