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

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

大家好，今天我们来深入探讨 Java CompletableFuture 的阻塞问题以及 ForkJoinPool 的使用陷阱。CompletableFuture 作为 Java 并发编程的重要组成部分，极大地简化了异步编程，但使用不当很容易造成阻塞，影响程序的性能和响应速度。

CompletableFuture 阻塞的常见原因

CompletableFuture 的设计初衷是提供非阻塞的异步编程模型，但以下情况会导致阻塞：

join() 和 get() 方法的直接调用：

join() 和 get() 方法都会阻塞当前线程，直到 CompletableFuture 完成并返回结果。这是最常见的阻塞原因。

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

// 阻塞当前线程直到 future 完成
String result = future.join(); // 或者 future.get();
System.out.println(result);

在这个例子中，future.join() 会阻塞当前线程 2 秒钟，直到 CompletableFuture 完成并返回 "Hello, World!"。

使用 thenApply() 等同步操作链，且没有合适的 Executor：

thenApply()、thenAccept()、thenRun() 等方法默认在执行 CompletableFuture 的线程池中执行后续操作。如果前一个 CompletableFuture 的执行耗时较长，且线程池大小有限，那么后续操作可能会因为等待线程而阻塞。

ExecutorService executor = Executors.newFixedThreadPool(2); // 模拟线程池大小有限

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

CompletableFuture<String> future2 = future.thenApply(s -> {
    try {
        Thread.sleep(1000); // 模拟耗时操作
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    return s.toUpperCase();
});

System.out.println("继续执行其他操作...");

// 使用 join/get 仍然会阻塞
System.out.println(future2.join());

executor.shutdown();

如果 executor 的线程池大小为 2，而第一个 CompletableFuture 已经占用了一个线程 2 秒，第二个 CompletableFuture 需要等待第一个 CompletableFuture 完成并释放线程才能执行。虽然主线程可以继续执行其他操作，但是最终 future2.join() 依然会阻塞。

不恰当的异常处理：

如果 CompletableFuture 抛出异常，且没有正确处理，那么调用 join() 或 get() 会抛出 ExecutionException 或 CompletionException。如果不处理这些异常，程序可能会崩溃。

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
    if (Math.random() > 0.5) {
        throw new RuntimeException("Something went wrong!");
    }
    return 42;
});

try {
    int result = future.join();
    System.out.println(result);
} catch (CompletionException e) {
    System.err.println("Exception occurred: " + e.getMessage());
}

良好的异常处理可以避免程序崩溃，但如果仅仅是简单地捕获并忽略异常，可能会导致程序逻辑错误。

死锁：

在复杂的异步操作中，如果不小心，可能会导致死锁。例如，一个 CompletableFuture 等待另一个 CompletableFuture 完成，而后者又等待前者的完成。

CompletableFuture<String> future1 = new CompletableFuture<>();
CompletableFuture<String> future2 = new CompletableFuture<>();

future1.thenCombine(future2, (s1, s2) -> s1 + s2).join(); // 隐含的等待 future2 完成

future2.thenCombine(future1, (s1, s2) -> s1 + s2).join(); // 隐含的等待 future1 完成

// 永远无法完成，导致死锁

这个例子中，future1 和 future2 互相等待对方完成，导致死锁。

如何避免 CompletableFuture 阻塞

使用异步回调：

避免直接调用 join() 和 get()，使用 thenApply()、thenAccept()、thenRun() 等方法注册回调函数，在 CompletableFuture 完成时异步执行回调函数。

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: " + result); // 异步处理结果
});

System.out.println("继续执行其他操作...");

在这个例子中，thenAccept() 方法注册了一个回调函数，该函数在 future 完成时异步执行，不会阻塞当前线程。

使用 Executor 指定执行线程池：

为 thenApply()、thenAccept()、thenRun() 等方法指定 Executor，可以避免在默认线程池中执行后续操作，从而避免阻塞。

ExecutorService executor = Executors.newFixedThreadPool(4);

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

future.thenApplyAsync(s -> s.toUpperCase(), executor) // 使用 executor 执行后续操作
      .thenAccept(result -> System.out.println("Result: " + result));

System.out.println("继续执行其他操作...");
executor.shutdown();

在这个例子中，thenApplyAsync() 方法使用 executor 指定的线程池执行后续操作，避免了在默认线程池中阻塞。

使用 orTimeout() 设置超时时间：

可以使用 orTimeout() 方法为 CompletableFuture 设置超时时间。如果 CompletableFuture 在指定时间内没有完成，则抛出 TimeoutException。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(3000); // 模拟耗时操作
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    return "Hello, World!";
}).orTimeout(2, TimeUnit.SECONDS); // 设置超时时间为 2 秒

try {
    String result = future.join();
    System.out.println(result);
} catch (CompletionException e) {
    if (e.getCause() instanceof TimeoutException) {
        System.err.println("Timeout occurred!");
    } else {
        System.err.println("Exception occurred: " + e.getMessage());
    }
}

在这个例子中，如果 future 在 2 秒内没有完成，则会抛出 TimeoutException，避免永久阻塞。

使用 completeOnTimeout() 设置默认值：

可以使用 completeOnTimeout() 方法为 CompletableFuture 设置默认值。如果 CompletableFuture 在指定时间内没有完成，则返回默认值。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(3000); // 模拟耗时操作
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    return "Hello, World!";
}).completeOnTimeout("Default Value", 2, TimeUnit.SECONDS); // 设置超时时间为 2 秒，默认值为 "Default Value"

String result = future.join();
System.out.println(result); // 如果超时，则输出 "Default Value"

在这个例子中，如果 future 在 2 秒内没有完成，则返回 "Default Value"，避免永久阻塞。

避免死锁：

在设计复杂的异步操作时，要仔细分析依赖关系，避免出现循环等待的情况。可以使用线程转储分析工具来检测死锁。

正确处理异常：

使用 exceptionally()、handle() 等方法处理 CompletableFuture 抛出的异常，避免程序崩溃。

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
    if (Math.random() > 0.5) {
        throw new RuntimeException("Something went wrong!");
    }
    return 42;
}).exceptionally(e -> {
    System.err.println("Exception occurred: " + e.getMessage());
    return -1; // 返回默认值
});

int result = future.join();
System.out.println(result);

在这个例子中，exceptionally() 方法处理了 CompletableFuture 抛出的异常，并返回一个默认值，避免程序崩溃。

ForkJoinPool 使用陷阱

ForkJoinPool 是 Java 7 引入的线程池，专门用于执行可以递归分解的任务，即 ForkJoinTask。它使用工作窃取算法来提高 CPU 的利用率，但在使用过程中也存在一些陷阱。

默认 ForkJoinPool 的滥用：

CompletableFuture.supplyAsync() 等方法如果不指定 Executor，默认会使用 ForkJoinPool.commonPool()。这个公共线程池的线程数量等于 Runtime.getRuntime().availableProcessors() - 1。如果所有的异步任务都使用这个公共线程池，可能会导致线程饥饿，影响程序的性能。
```
// 所有任务都使用默认的 ForkJoinPool
CompletableFuture.supplyAsync(() -> {
    // 耗时操作
    return "Task 1";
});

CompletableFuture.supplyAsync(() -> {
    // 耗时操作
    return "Task 2";
});
```
更好的做法是为不同的任务创建不同的线程池，避免互相干扰。
阻塞操作放入 ForkJoinTask：

ForkJoinPool 旨在执行计算密集型的任务，如果将阻塞操作放入 ForkJoinTask，会导致线程空闲，降低 CPU 的利用率。
```
ForkJoinPool pool = new ForkJoinPool();
pool.submit(() -> {
    try {
        Thread.sleep(2000); // 阻塞操作
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    return "Result";
});
```
应该将阻塞操作放在专门处理 I/O 的线程池中，避免占用 ForkJoinPool 的线程。

任务分解不合理：

ForkJoinTask 需要将任务分解成更小的子任务，如果分解不合理，会导致过多的任务创建和销毁，增加开销。

// 过度分解，导致开销增加
class MyRecursiveTask extends RecursiveTask<Integer> {
    private int start;
    private int end;

    public MyRecursiveTask(int start, int end) {
        this.start = start;
        this.end = end;
    }

    @Override
    protected Integer compute() {
        if (end - start <= 1) { // 分解粒度太小
            return start + end;
        } else {
            int middle = (start + end) / 2;
            MyRecursiveTask left = new MyRecursiveTask(start, middle);
            MyRecursiveTask right = new MyRecursiveTask(middle + 1, end);
            left.fork();
            right.fork();
            return left.join() + right.join();
        }
    }
}

应该根据任务的特点选择合适的分解粒度，避免过度分解。

ForkJoinPool 的线程数量设置不合理：

ForkJoinPool 的线程数量应该根据 CPU 的核心数和任务的特点来设置。如果线程数量太少，无法充分利用 CPU 的资源；如果线程数量太多，会导致过多的上下文切换，降低性能。
```
// 线程数量设置太少，无法充分利用 CPU 资源
ForkJoinPool pool = new ForkJoinPool(1); // 假设 CPU 有 8 个核心
```
一般来说，ForkJoinPool 的线程数量可以设置为 CPU 的核心数或核心数的两倍。
忘记关闭 ForkJoinPool：

ForkJoinPool 在不再使用时应该关闭，否则会导致资源泄漏。
```
ForkJoinPool pool = new ForkJoinPool();
// ... 使用 pool ...
pool.shutdown(); // 关闭 ForkJoinPool
```
可以使用 try-with-resources 语句来自动关闭 ForkJoinPool。

表格总结

问题	原因	解决方案
`join()` 和 `get()` 的直接调用	阻塞当前线程，直到 `CompletableFuture` 完成。	使用异步回调，例如 `thenApply()`、`thenAccept()`、`thenRun()`。
同步操作链，没有合适的 `Executor`	后续操作在默认线程池中执行，可能因为线程池大小有限而阻塞。	为 `thenApply()`、`thenAccept()`、`thenRun()` 等方法指定 `Executor`。
不恰当的异常处理	如果 `CompletableFuture` 抛出异常，且没有正确处理，会导致程序崩溃或逻辑错误。	使用 `exceptionally()`、`handle()` 等方法处理异常。
死锁	在复杂的异步操作中，出现循环等待的情况。	仔细分析依赖关系，避免循环等待。
默认 `ForkJoinPool` 的滥用	所有异步任务都使用 `ForkJoinPool.commonPool()`，可能导致线程饥饿。	为不同的任务创建不同的线程池。
阻塞操作放入 `ForkJoinTask`	`ForkJoinPool` 旨在执行计算密集型任务，阻塞操作会导致线程空闲。	将阻塞操作放在专门处理 I/O 的线程池中。
任务分解不合理	分解粒度太小会导致过多的任务创建和销毁，增加开销。	根据任务的特点选择合适的分解粒度。
`ForkJoinPool` 的线程数量设置不合理	线程数量太少无法充分利用 CPU 的资源，线程数量太多会导致过多的上下文切换。	根据 CPU 的核心数和任务的特点来设置线程数量。
忘记关闭 `ForkJoinPool`	导致资源泄漏。	在不再使用时关闭 `ForkJoinPool`。

总结陈词

避免 CompletableFuture 阻塞的关键在于使用异步回调，并为后续操作指定合适的 Executor。使用 ForkJoinPool 时需要注意避免滥用默认线程池、避免将阻塞操作放入 ForkJoinTask、合理分解任务、设置合适的线程数量，并在不再使用时关闭线程池。正确使用 CompletableFuture 和 ForkJoinPool 可以提高程序的性能和响应速度。

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

发表回复 取消回复

发表回复取消回复