发布于 由admin

JAVA CompletableFuture链式任务卡死与线程池阻塞分析

JAVA CompletableFuture链式任务卡死与线程池阻塞分析

大家好，今天我们来聊聊Java CompletableFuture链式任务中可能遇到的卡死和线程池阻塞问题。CompletableFuture作为Java并发编程的重要工具，极大地简化了异步编程的复杂性，但如果不正确地使用，很容易掉入陷阱。

CompletableFuture的基本概念与链式操作

首先，我们简单回顾一下CompletableFuture的核心概念。CompletableFuture代表一个异步计算的结果，它允许我们在计算完成时执行回调函数，或者将多个CompletableFuture串联起来形成一个链式任务。

常见的链式操作包括：

  • thenApply(): 对结果进行转换。
  • thenAccept(): 消费结果，不返回任何值。
  • thenRun(): 执行一个Runnable，不依赖结果。
  • thenCompose(): 将结果传递给另一个CompletableFuture，实现异步任务的组合。
  • thenCombine(): 合并两个CompletableFuture的结果。
  • exceptionally(): 处理异常情况。

这些方法可以灵活地组合，构建复杂的异步流程。例如：

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello")
        .thenApply(s -> s + " World")
        .thenApply(String::toUpperCase);

future.thenAccept(System.out::println); // 输出 HELLO WORLD

这段代码创建了一个CompletableFuture，先异步执行"Hello"字符串的生成，然后将结果追加" World"，再转换成大写，最后打印到控制台。

卡死与阻塞的常见场景

尽管CompletableFuture提供了强大的异步编程能力，但以下场景可能会导致卡死或线程池阻塞：

  1. 死锁： 多个CompletableFuture相互等待对方完成，导致所有任务都无法继续执行。
  2. 任务堆积： 提交到线程池的任务数量超过线程池的处理能力，导致任务队列积压，最终阻塞线程池。
  3. 长时间运行的任务： 某个CompletableFuture的任务执行时间过长，占用了线程池中的线程，导致其他任务无法执行。
  4. 错误处理不当： 异常未被正确处理，导致CompletableFuture链中断，后续任务无法执行。
  5. 嵌套的CompletableFuture未正确处理： 在thenApply或其他链式操作中创建新的CompletableFuture，但没有正确地处理它的完成，导致外部的CompletableFuture无法完成。
  6. 默认线程池的资源限制： 使用CompletableFuture.supplyAsync等方法，如果没有指定Executor，会使用ForkJoinPool.commonPool()，这个线程池的线程数量受到CPU核心数的限制，在高并发场景下容易成为瓶颈。

接下来，我们分别分析这些场景，并提供相应的解决方案。

死锁案例与解决方案

死锁是指两个或多个线程相互等待对方释放资源，导致所有线程都无法继续执行。在CompletableFuture中，死锁通常发生在多个CompletableFuture相互依赖，形成循环等待的场景。

public class DeadlockExample {

    public static void main(String[] args) throws Exception {
        CompletableFuture<String> future1 = new CompletableFuture<>();
        CompletableFuture<String> future2 = new CompletableFuture<>();

        future1.thenAcceptAsync(result -> {
            try {
                // 模拟耗时操作
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            future2.complete("Future2 completed based on " + result);
        });

        future2.thenAcceptAsync(result -> {
            try {
                // 模拟耗时操作
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            future1.complete("Future1 completed based on " + result);
        });

        // 触发future1和future2的执行
        future1.complete("Initial value for Future1");
        future2.complete("Initial value for Future2"); // 如果没有这行，更容易死锁

        System.out.println("Waiting for futures to complete...");
        future1.get(); // 阻塞等待future1完成
        future2.get(); // 阻塞等待future2完成
        System.out.println("Futures completed.");
    }
}

在这个例子中，future1等待future2完成，而future2又等待future1完成，形成了一个循环依赖，导致死锁。程序会一直阻塞，无法输出"Futures completed."。

解决方案：

  • 避免循环依赖： 重新设计异步流程，消除CompletableFuture之间的循环依赖关系。
  • 设置超时时间： 在get()方法中设置超时时间，防止无限期等待。例如：future1.get(1, TimeUnit.SECONDS)。
  • 使用独立的Executor： 为不同的CompletableFuture使用不同的Executor，避免线程之间的竞争。

修改后的代码（避免循环依赖）：

public class NoDeadlockExample {

    public static void main(String[] args) throws Exception {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "Initial value for Future1");
        CompletableFuture<String> future2 = future1.thenApply(result -> {
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Future2 completed based on " + result;
        });

        System.out.println("Waiting for futures to complete...");
        System.out.println(future2.get());
        System.out.println("Futures completed.");
    }
}

在这个修改后的例子中，future2依赖于future1，但future1不再依赖于future2，消除了循环依赖，避免了死锁。

任务堆积与线程池阻塞

当提交到线程池的任务数量超过线程池的处理能力时，任务会堆积在任务队列中。如果任务队列是无界的，可能会导致内存溢出。如果任务队列是有界的，当队列满了之后，新的任务会被拒绝，或者阻塞提交线程。

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

public class TaskPileUpExample {

    public static void main(String[] args) throws InterruptedException {
        // 创建一个固定大小的线程池，大小为2
        ExecutorService executor = Executors.newFixedThreadPool(2);

        // 提交大量任务，超过线程池的处理能力
        for (int i = 0; i < 10; i++) {
            final int taskNumber = i;
            executor.submit(() -> {
                try {
                    System.out.println("Task " + taskNumber + " started by " + Thread.currentThread().getName());
                    // 模拟耗时操作
                    Thread.sleep(2000);
                    System.out.println("Task " + taskNumber + " completed by " + Thread.currentThread().getName());
                    return "Result of task " + taskNumber;
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    return "Task " + taskNumber + " interrupted";
                }
            });
        }

        // 关闭线程池，等待所有任务完成
        executor.shutdown();
        executor.awaitTermination(10, TimeUnit.SECONDS);

        System.out.println("All tasks submitted.");
    }
}

在这个例子中，我们创建了一个大小为2的固定线程池，并提交了10个任务。由于线程池的处理能力有限，任务会堆积在任务队列中，导致后面的任务需要等待前面的任务完成才能开始执行。

解决方案：

  • 合理配置线程池大小： 根据任务的类型（CPU密集型或IO密集型）和并发量，选择合适的线程池大小。
  • 使用有界队列： 为线程池配置有界队列，防止任务无限堆积。可以使用ThreadPoolExecutor的构造函数来指定队列类型和大小。
  • 使用RejectedExecutionHandler： 当任务队列已满时，RejectedExecutionHandler可以处理被拒绝的任务。常见的策略包括：
    • AbortPolicy：直接抛出RejectedExecutionException。
    • CallerRunsPolicy：由提交任务的线程来执行被拒绝的任务。
    • DiscardPolicy：直接丢弃被拒绝的任务。
    • DiscardOldestPolicy：丢弃队列中最老的任务，然后尝试提交新任务。
  • 使用异步流控： 在高并发场景下，可以使用异步流控技术，例如令牌桶算法或漏桶算法，控制任务的提交速率，防止任务堆积。

修改后的代码（使用有界队列和RejectedExecutionHandler）：

import java.util.concurrent.*;

public class TaskPileUpSolution {

    public static void main(String[] args) throws InterruptedException {
        // 创建一个固定大小的线程池，大小为2，使用有界队列，队列大小为5
        ExecutorService executor = new ThreadPoolExecutor(2, 2, 0L, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(5), new ThreadPoolExecutor.CallerRunsPolicy());

        // 提交大量任务，超过线程池的处理能力
        for (int i = 0; i < 10; i++) {
            final int taskNumber = i;
            executor.submit(() -> {
                try {
                    System.out.println("Task " + taskNumber + " started by " + Thread.currentThread().getName());
                    // 模拟耗时操作
                    Thread.sleep(2000);
                    System.out.println("Task " + taskNumber + " completed by " + Thread.currentThread().getName());
                    return "Result of task " + taskNumber;
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    return "Task " + taskNumber + " interrupted";
                }
            });
        }

        // 关闭线程池，等待所有任务完成
        executor.shutdown();
        executor.awaitTermination(10, TimeUnit.SECONDS);

        System.out.println("All tasks submitted.");
    }
}

在这个修改后的例子中，我们使用了ArrayBlockingQueue作为有界队列，队列大小为5。当任务队列已满时，CallerRunsPolicy会由提交任务的线程来执行被拒绝的任务，从而避免任务被直接丢弃。

长时间运行的任务

如果某个CompletableFuture的任务执行时间过长，会占用线程池中的线程，导致其他任务无法执行，最终阻塞线程池。

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

public class LongRunningTaskExample {

    public static void main(String[] args) throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(2);

        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println("Long running task started by " + Thread.currentThread().getName());
                TimeUnit.SECONDS.sleep(5); // 模拟长时间运行的任务
                System.out.println("Long running task completed by " + Thread.currentThread().getName());
                return "Result of long running task";
            } catch (InterruptedException e) {
                e.printStackTrace();
                return "Long running task interrupted";
            }
        }, executor);

        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            System.out.println("Short task started by " + Thread.currentThread().getName());
            return "Result of short task";
        }, executor);

        System.out.println(future2.get()); //等待future2完成

        System.out.println(future1.get()); // 等待future1完成

        executor.shutdown();
        executor.awaitTermination(10, TimeUnit.SECONDS);

    }
}

在这个例子中，future1执行一个需要5秒钟才能完成的任务，而future2执行一个很快就能完成的任务。由于线程池的大小只有2，future1会占用一个线程，导致future2需要等待future1完成后才能开始执行。

解决方案：

  • 优化任务执行时间： 尽量缩短任务的执行时间，例如通过优化算法、减少IO操作等。
  • 使用更大的线程池： 增加线程池的大小，提高并发处理能力。
  • 将长时间运行的任务拆分成多个小任务： 将长时间运行的任务拆分成多个小任务，并使用CompletableFuture链式操作将它们串联起来，从而释放线程，让其他任务有机会执行。
  • 使用专门的Executor： 为长时间运行的任务使用专门的Executor，与其他任务隔离。

修改后的代码（使用更大的线程池）：

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

public class LongRunningTaskSolution {

    public static void main(String[] args) throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(4); // 增加线程池大小

        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println("Long running task started by " + Thread.currentThread().getName());
                TimeUnit.SECONDS.sleep(5); // 模拟长时间运行的任务
                System.out.println("Long running task completed by " + Thread.currentThread().getName());
                return "Result of long running task";
            } catch (InterruptedException e) {
                e.printStackTrace();
                return "Long running task interrupted";
            }
        }, executor);

        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            System.out.println("Short task started by " + Thread.currentThread().getName());
            return "Result of short task";
        }, executor);

        System.out.println(future2.get()); //等待future2完成

        System.out.println(future1.get()); // 等待future1完成

        executor.shutdown();
        executor.awaitTermination(10, TimeUnit.SECONDS);

    }
}

在这个修改后的例子中，我们将线程池的大小增加到4，从而提高了并发处理能力，future2不再需要等待future1完成后才能开始执行。

错误处理不当

如果CompletableFuture链中某个任务抛出异常，但没有被正确处理，可能会导致CompletableFuture链中断，后续任务无法执行。

import java.util.concurrent.CompletableFuture;

public class ExceptionHandlingExample {

    public static void main(String[] args) throws Exception {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("Task started");
            throw new RuntimeException("Something went wrong");
        }).thenApply(result -> {
            System.out.println("This will not be printed");
            return result.toUpperCase();
        });

        try {
            System.out.println(future.get());
        } catch (Exception e) {
            System.err.println("Exception caught: " + e.getMessage());
        }
    }
}

在这个例子中，supplyAsync中的任务抛出了RuntimeException，导致thenApply中的任务无法执行。如果没有捕获这个异常，程序可能会崩溃。

解决方案：

  • 使用exceptionally()方法： exceptionally()方法可以处理CompletableFuture中的异常，并返回一个默认值。
  • 使用handle()方法： handle()方法可以处理CompletableFuture中的结果和异常，并返回一个新的CompletableFuture。
  • 在get()方法中捕获异常： 在使用get()方法获取CompletableFuture的结果时，需要捕获可能抛出的异常。

修改后的代码（使用exceptionally()方法）：

import java.util.concurrent.CompletableFuture;

public class ExceptionHandlingSolution {

    public static void main(String[] args) throws Exception {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("Task started");
            throw new RuntimeException("Something went wrong");
        }).thenApply(result -> {
            System.out.println("This will not be printed");
            return result.toUpperCase();
        }).exceptionally(e -> {
            System.err.println("Exception caught: " + e.getMessage());
            return "Default value";
        });

        System.out.println(future.get());
    }
}

在这个修改后的例子中，我们使用了exceptionally()方法来处理异常，并返回一个默认值。即使supplyAsync中的任务抛出异常，thenApply中的任务无法执行，程序也不会崩溃，而是会输出"Default value"。

嵌套的CompletableFuture未正确处理

在thenApply或其他链式操作中，如果创建了新的CompletableFuture，但没有正确地处理它的完成，会导致外部的CompletableFuture无法完成。

import java.util.concurrent.CompletableFuture;

public class NestedFutureExample {

    public static void main(String[] args) throws Exception {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello")
                .thenApply(s -> {
                    CompletableFuture<String> innerFuture = CompletableFuture.supplyAsync(() -> s + " World");
                    // 这里缺少对innerFuture的处理
                    return "Outer task completed";
                });

        System.out.println(future.get());
    }
}

在这个例子中，thenApply中创建了一个新的CompletableFuture innerFuture，但是没有等待innerFuture完成就直接返回了"Outer task completed"。这导致外部的future在innerFuture完成之前就完成了，结果可能不是我们期望的。

解决方案：

  • 使用thenCompose()方法： thenCompose()方法可以将结果传递给另一个CompletableFuture，并返回一个新的CompletableFuture，该CompletableFuture在内部的CompletableFuture完成后才会完成。
  • 等待内部CompletableFuture完成： 在thenApply中，可以使用innerFuture.join()或innerFuture.get()方法等待内部CompletableFuture完成。

修改后的代码（使用thenCompose()方法）：

import java.util.concurrent.CompletableFuture;

public class NestedFutureSolution {

    public static void main(String[] args) throws Exception {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello")
                .thenCompose(s -> CompletableFuture.supplyAsync(() -> s + " World"));

        System.out.println(future.get());
    }
}

在这个修改后的例子中，我们使用了thenCompose()方法来处理内部的CompletableFuture，确保外部的future在内部的innerFuture完成后才会完成。

默认线程池的资源限制

CompletableFuture.supplyAsync()等方法，如果没有指定Executor，会使用ForkJoinPool.commonPool()，这个线程池的线程数量受到CPU核心数的限制，在高并发场景下容易成为瓶颈。

解决方案：

  • 自定义Executor： 使用ThreadPoolExecutor或ForkJoinPool创建自定义的Executor，并根据实际情况调整线程池大小。
  • 避免阻塞操作： 在ForkJoinPool.commonPool()中执行的任务应该避免阻塞操作，否则会影响其他任务的执行。

总结表格

为了更清晰地总结上述内容，我们使用表格来概括：

问题 描述 解决方案
死锁 多个CompletableFuture相互等待对方完成，导致所有任务都无法继续执行。 避免循环依赖；设置超时时间；使用独立的Executor。
任务堆积 提交到线程池的任务数量超过线程池的处理能力，导致任务队列积压。 合理配置线程池大小；使用有界队列；使用RejectedExecutionHandler；使用异步流控。
长时间运行的任务 某个CompletableFuture的任务执行时间过长，占用了线程池中的线程。 优化任务执行时间；使用更大的线程池；将长时间运行的任务拆分成多个小任务；使用专门的Executor。
错误处理不当 异常未被正确处理，导致CompletableFuture链中断，后续任务无法执行。 使用exceptionally()方法；使用handle()方法；在get()方法中捕获异常。
嵌套的CompletableFuture未正确处理 在thenApply或其他链式操作中创建新的CompletableFuture，但没有正确地处理它的完成，导致外部的CompletableFuture无法完成。 使用thenCompose()方法；等待内部CompletableFuture完成。
默认线程池的资源限制 使用ForkJoinPool.commonPool()，线程数量受到CPU核心数的限制，高并发场景下容易成为瓶颈。 自定义Executor；避免阻塞操作。

避免卡死和阻塞，优化你的异步代码

通过以上分析，我们可以看到，CompletableFuture的卡死和线程池阻塞问题通常是由不正确的线程池配置、任务依赖关系以及错误处理方式引起的。解决这些问题的关键在于理解CompletableFuture的执行机制，合理配置线程池，避免循环依赖，正确处理异常，并选择合适的链式操作方法。希望今天的分享能够帮助大家更好地使用CompletableFuture，编写出更高效、更健壮的异步代码。

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注