JAVA 如何避免 CompletableFuture 阻塞？forkjoinpool 使用陷阱 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

JAVA CompletableFuture 避免阻塞与 ForkJoinPool 使用陷阱

大家好，今天我们来深入探讨 Java 中 CompletableFuture 的一个关键问题：如何避免阻塞，以及在使用 ForkJoinPool 时可能遇到的陷阱。CompletableFuture 作为 Java 8 引入的异步编程利器，能够极大地提升程序的并发性和响应速度。但如果使用不当，很容易造成阻塞，反而降低了性能。同时，ForkJoinPool 作为 CompletableFuture 默认使用的线程池，其特性也需要深入理解，否则容易掉入陷阱。

CompletableFuture 的异步特性与阻塞风险

CompletableFuture 的核心思想是将一个耗时操作异步化，让调用线程不必等待结果，从而释放资源去做其他事情。它提供了多种异步执行任务的方法，例如：

supplyAsync(Supplier<U> supplier): 异步执行一个有返回值的任务。
runAsync(Runnable runnable): 异步执行一个没有返回值的任务。
thenApply(Function<T,U> fn): 对前一个任务的结果进行转换。
thenAccept(Consumer<T> consumer): 对前一个任务的结果进行消费。
thenCompose(Function<T, CompletionStage<U>> fn): 将前一个任务的结果传递给另一个 CompletableFuture。

这些方法都有同步和异步两种版本。异步版本通常会使用一个 Executor 来执行任务，如果没有指定 Executor，则默认使用 ForkJoinPool.commonPool()。

阻塞风险主要出现在以下几个场景：

同步等待结果： 最直接的阻塞方式就是调用 CompletableFuture.get() 方法。get() 方法会阻塞当前线程，直到 CompletableFuture 完成并返回结果。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(2000); // 模拟耗时操作
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    return "Hello, World!";
});

try {
    String result = future.get(); // 阻塞当前线程
    System.out.println(result);
} catch (InterruptedException | ExecutionException e) {
    e.printStackTrace();
}

上面的代码中，future.get() 会阻塞当前线程 2 秒钟，直到 CompletableFuture 完成。

错误地使用同步版本的组合方法： 例如 thenApply() 的同步版本，虽然名字和异步版本一样，但实际上是在执行 thenApply() 的线程中执行转换操作，如果这个转换操作耗时，同样会造成阻塞。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Input");

CompletableFuture<String> transformedFuture = future.thenApply(input -> {
    try {
        Thread.sleep(1000); // 模拟耗时转换
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    return input + " - Transformed";
});

try {
    System.out.println(transformedFuture.get());
} catch (InterruptedException | ExecutionException e) {
    e.printStackTrace();
}

在这个例子中，虽然 supplyAsync() 是异步的，但是 thenApply() 的同步版本使得转换操作在 ForkJoinPool 的线程中执行，阻塞了该线程。

过度依赖 ForkJoinPool.commonPool()： ForkJoinPool.commonPool() 是一个全局共享的线程池，如果大量的 CompletableFuture 任务都使用它，并且这些任务都比较耗时，就可能导致线程池资源耗尽，从而阻塞后续的任务。
在 ForkJoinPool 中执行阻塞操作： ForkJoinPool 的设计目标是执行 CPU 密集型任务，如果在一个 ForkJoinPool 线程中执行了 IO 密集型或阻塞操作，会导致该线程长时间处于等待状态，无法执行其他任务，从而降低线程池的效率。

避免阻塞的策略

为了避免 CompletableFuture 阻塞，我们需要采取以下策略：

避免直接调用 get() 方法： 尽量避免使用 get() 方法，而是使用 thenApply(), thenAccept(), thenCompose() 等方法来异步处理结果。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    return "Hello, World!";
});

future.thenAccept(result -> System.out.println(result)); // 异步处理结果

// 主线程可以继续执行其他任务
System.out.println("Main thread continues...");

在这个例子中，我们使用 thenAccept() 来异步处理 CompletableFuture 的结果，避免了主线程的阻塞。

使用异步版本的组合方法： 确保使用 thenApplyAsync(), thenAcceptAsync(), thenComposeAsync() 等异步版本的组合方法，指定一个 Executor 来执行后续操作。

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

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Input");

CompletableFuture<String> transformedFuture = future.thenApplyAsync(input -> {
    try {
        Thread.sleep(1000); // 模拟耗时转换
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    return input + " - Transformed";
}, executor); // 指定使用自定义的线程池

transformedFuture.thenAccept(result -> System.out.println(result));

在这个例子中，我们使用 thenApplyAsync() 并指定了一个自定义的线程池，确保转换操作在自定义的线程池中异步执行，避免了阻塞 ForkJoinPool.commonPool()。

使用自定义的 Executor： 为不同的 CompletableFuture 任务使用不同的 Executor，避免所有任务都挤在 ForkJoinPool.commonPool() 中。特别是IO密集型的任务，应该使用固定大小的线程池，并且线程数量要足够大，应对大量IO操作。 CPU密集型的任务可以使用 ForkJoinPool 或者固定大小的线程池，线程数量等于CPU核心数。
```
ExecutorService cpuExecutor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
ExecutorService ioExecutor = Executors.newFixedThreadPool(20);

CompletableFuture<String> cpuFuture = CompletableFuture.supplyAsync(() -> {
    // CPU 密集型任务
    return "CPU Result";
}, cpuExecutor);

CompletableFuture<String> ioFuture = CompletableFuture.supplyAsync(() -> {
    // IO 密集型任务
    return "IO Result";
}, ioExecutor);
```
在这个例子中，我们为 CPU 密集型任务和 IO 密集型任务分别创建了不同的线程池。

避免在 ForkJoinPool 中执行阻塞操作： 如果需要在 CompletableFuture 中执行 IO 操作，应该使用专门的 IO 线程池，或者使用异步 IO API (例如 AsynchronousFileChannel)。

ExecutorService ioExecutor = Executors.newFixedThreadPool(20);

CompletableFuture<String> ioFuture = CompletableFuture.supplyAsync(() -> {
    try {
        // 模拟 IO 操作
        Thread.sleep(1000);
        return "IO Result";
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        return "Error";
    }
}, ioExecutor);

在这个例子中，我们使用了一个专门的 IO 线程池来执行 IO 操作。

设置超时时间： 如果你确实需要使用 get() 方法，那么强烈建议设置超时时间，防止无限期阻塞。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(5000); // 模拟耗时操作
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    return "Hello, World!";
});

try {
    String result = future.get(3, TimeUnit.SECONDS); // 设置超时时间为 3 秒
    System.out.println(result);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
    e.printStackTrace();
}

如果超过 3 秒钟，CompletableFuture 仍然没有完成，get() 方法会抛出 TimeoutException。

ForkJoinPool 使用陷阱

ForkJoinPool 是一个专门用于执行分治任务的线程池，它通过工作窃取算法来提高线程的利用率。但如果不了解其特性，很容易掉入陷阱。

工作窃取原理： ForkJoinPool 中的每个线程都有一个自己的工作队列 (Deque)，当一个线程完成自己的任务后，会尝试从其他线程的工作队列中窃取任务来执行。这个过程称为工作窃取 (Work Stealing)。
不适合执行长时间运行的任务： ForkJoinPool 的设计目标是执行短期的、CPU 密集型的任务，如果在一个 ForkJoinPool 线程中执行了长时间运行的任务，会导致其他线程无法窃取到任务，从而降低线程池的效率。
谨慎使用 ForkJoinPool.commonPool()： ForkJoinPool.commonPool() 是一个全局共享的线程池，其大小受到系统资源的限制。如果大量的 CompletableFuture 任务都使用它，并且这些任务都比较耗时，就可能导致线程池资源耗尽，从而阻塞后续的任务。此外，如果某个任务抛出了未捕获的异常，可能会导致整个 ForkJoinPool.commonPool() 崩溃，影响其他使用该线程池的任务。
避免任务之间的依赖关系： ForkJoinPool 的工作窃取算法依赖于任务之间的独立性。如果任务之间存在依赖关系，会导致某些线程一直处于等待状态，无法执行其他任务，从而降低线程池的效率。

正确处理异常： 在 ForkJoinPool 中执行的任务可能会抛出异常，需要正确处理这些异常，避免影响其他任务的执行。可以使用 CompletableFuture.exceptionally() 方法来处理异常。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    if (true) {
        throw new RuntimeException("Something went wrong!");
    }
    return "Success";
});

future.exceptionally(e -> {
    System.err.println("Exception: " + e.getMessage());
    return "Default Value";
}).thenAccept(result -> System.out.println("Result: " + result));

在这个例子中，我们使用 exceptionally() 方法来处理异常，如果 CompletableFuture 抛出异常，则返回一个默认值。

代码示例：使用自定义线程池避免阻塞

下面是一个完整的代码示例，演示了如何使用自定义线程池来避免 CompletableFuture 阻塞：

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class CompletableFutureExample {

    public static void main(String[] args) throws InterruptedException {
        // 创建一个自定义的线程池
        ExecutorService executor = Executors.newFixedThreadPool(10);

        // 创建一个 CompletableFuture，使用自定义线程池执行异步任务
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000); // 模拟耗时操作
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return "Hello, World!";
        }, executor);

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

        // 主线程可以继续执行其他任务
        System.out.println("Main thread continues...");

        // 关闭线程池
        executor.shutdown();
        executor.awaitTermination(5, TimeUnit.SECONDS);
    }
}

在这个例子中，我们创建了一个自定义的线程池，并将其传递给 supplyAsync() 方法。这样，异步任务就会在自定义的线程池中执行，避免了阻塞 ForkJoinPool.commonPool()。

使用建议总结

尽量避免使用 get() 方法，使用异步组合方法处理结果。
使用 thenApplyAsync(), thenAcceptAsync() 等方法指定 Executor。
为不同的任务类型创建不同的 Executor。
避免在 ForkJoinPool 中执行阻塞操作。
设置超时时间，防止无限期阻塞。
理解 ForkJoinPool 的工作原理，避免滥用 ForkJoinPool.commonPool()。
正确处理异常，避免影响其他任务的执行。
避免任务间的依赖，保证工作窃取的顺利进行。

希望今天的分享能够帮助大家更好地理解和使用 CompletableFuture，避免阻塞，充分发挥其异步编程的优势。谢谢大家！