JAVA CompletableFuture并发链中异常传播失败的原因与解决方案

JAVA CompletableFuture 并发链中异常传播失败的原因与解决方案

大家好，今天我们来聊聊 Java CompletableFuture 并发链中异常传播的问题。CompletableFuture 作为 Java 并发编程中的利器，极大地简化了异步任务的处理。然而，在实际应用中，我们经常会遇到并发链中异常没有正确传播的情况，导致程序出现意料之外的行为。本文将深入分析异常传播失败的常见原因，并提供相应的解决方案，帮助大家更好地驾驭 CompletableFuture。

一、CompletableFuture 异常处理机制概述

在深入探讨异常传播失败的原因之前，我们先来回顾一下 CompletableFuture 的异常处理机制。CompletableFuture 提供了多种处理异常的方法，主要包括：

exceptionally(Function<Throwable, ? extends T> fn): 当 CompletableFuture 正常完成时，该方法不会被调用。只有当 CompletableFuture 抛出异常时，才会调用 fn 函数，并将异常作为参数传递给它。 fn 函数返回的值将作为 CompletableFuture 的结果。

handle(BiFunction<? super T, Throwable, ? extends U> fn): 无论 CompletableFuture 是正常完成还是抛出异常，该方法都会被调用。如果 CompletableFuture 正常完成，则将结果作为第一个参数传递给 fn 函数，第二个参数为 null。如果 CompletableFuture 抛出异常，则将结果设置为 null，并将异常作为第二个参数传递给 fn 函数。 fn 函数返回的值将作为 CompletableFuture 的结果。

whenComplete(BiConsumer<? super T, Throwable> action): 与 handle 方法类似，无论 CompletableFuture 是正常完成还是抛出异常，该方法都会被调用。但是，与 handle 不同的是， whenComplete 方法不会修改 CompletableFuture 的结果，它主要用于执行一些副作用操作，例如记录日志。

这些方法允许我们在 CompletableFuture 链中的不同阶段捕获和处理异常，确保程序的健壮性。

二、异常传播失败的常见原因

尽管 CompletableFuture 提供了丰富的异常处理机制，但在实际应用中，我们仍然可能遇到异常传播失败的情况。以下是一些常见的原因：

中间环节未处理异常

这是最常见的原因。如果在 CompletableFuture 链的中间环节没有处理异常，并且后续的 CompletableFuture 没有依赖于前一个 CompletableFuture 的结果，那么异常可能会被忽略，导致程序继续执行，最终产生错误的结果。

示例代码：

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> {
    System.out.println("Task 1 running...");
    if (true) {
        throw new RuntimeException("Task 1 failed");
    }
    return "Result from Task 1";
}).thenApply(result -> {
    System.out.println("Task 2 running...");
    return result.toUpperCase();
});
 
try {
    System.out.println(cf.get()); // 阻塞等待结果
} catch (Exception e) {
    System.err.println("Exception caught: " + e.getMessage());
}

在这个例子中， supplyAsync 抛出了一个 RuntimeException，但是 thenApply 并没有处理这个异常。如果 cf.get() 被调用，将会抛出 ExecutionException，包含了原始的 RuntimeException。但是如果没有调用 get() 或者 join()等方法，并且没有其他依赖于 cf 的 CompletableFuture，这个异常可能会被忽略，程序可能会继续执行其他任务，而没有意识到错误。

解决方案：

在每个可能抛出异常的 CompletableFuture 之后，添加 exceptionally、 handle 或 whenComplete 方法来处理异常。

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> {
    System.out.println("Task 1 running...");
    if (true) {
        throw new RuntimeException("Task 1 failed");
    }
    return "Result from Task 1";
}).thenApply(result -> {
    System.out.println("Task 2 running...");
    return result.toUpperCase();
}).exceptionally(ex -> {
    System.err.println("Exception occurred: " + ex.getMessage());
    return "Fallback Result"; // 返回一个默认值
});
 
try {
    System.out.println(cf.get());
} catch (Exception e) {
    System.err.println("Exception caught: " + e.getMessage());
}

异常被包装在 CompletionException 或 ExecutionException 中

当使用 get()、 join() 或 getNow() 等方法获取 CompletableFuture 的结果时，如果 CompletableFuture 抛出异常，那么原始的异常会被包装在 CompletionException 或 ExecutionException 中。这使得异常处理变得更加复杂，因为需要先解包才能获取原始的异常。

示例代码：

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> {
    throw new IllegalArgumentException("Invalid argument");
});
 
try {
    cf.get();
} catch (InterruptedException | ExecutionException e) {
    System.err.println("Caught: " + e.getClass().getName());
    System.err.println("Cause: " + e.getCause().getClass().getName());
    System.err.println("Message: " + e.getCause().getMessage());
}

在这个例子中， supplyAsync 抛出了一个 IllegalArgumentException。当调用 cf.get() 时，会抛出 ExecutionException，而 IllegalArgumentException 变成了 ExecutionException 的 cause。

解决方案：

在捕获 CompletionException 或 ExecutionException 时，使用 getCause() 方法获取原始的异常，并根据原始异常的类型进行处理。

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> {
    throw new IllegalArgumentException("Invalid argument");
});
 
try {
    cf.get();
} catch (InterruptedException | ExecutionException e) {
    Throwable cause = e.getCause();
    if (cause instanceof IllegalArgumentException) {
        System.err.println("Caught IllegalArgumentException: " + cause.getMessage());
    } else {
        System.err.println("Caught other exception: " + e.getMessage());
    }
}

使用了错误的异常处理方法

CompletableFuture 提供了多种异常处理方法，如 exceptionally、 handle 和 whenComplete。选择错误的异常处理方法可能导致异常无法正确传播。例如，如果在 thenApply 方法中使用了 exceptionally 方法，那么只有当 thenApply 方法抛出异常时， exceptionally 方法才会被调用。如果 thenApply 之前的 CompletableFuture 抛出了异常，那么 exceptionally 方法不会被调用。

示例代码：

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> {
    throw new RuntimeException("Task 1 failed");
}).thenApply(result -> {
    return result.toUpperCase();
}).exceptionally(ex -> {
    System.err.println("Exception in thenApply: " + ex.getMessage());
    return "Fallback Result";
});
 
try {
    System.out.println(cf.get());
} catch (Exception e) {
    System.err.println("Exception caught: " + e.getMessage());
}

在这个例子中， supplyAsync 抛出了一个 RuntimeException。 exceptionally 方法被添加到 thenApply 之后，所以它只会在 thenApply 抛出异常时才会被调用。由于异常发生在 supplyAsync 中， exceptionally 方法不会被调用，最终会抛出 ExecutionException。

解决方案：

根据实际情况选择合适的异常处理方法。如果需要在 CompletableFuture 链中的任何阶段捕获异常，可以使用 handle 或 whenComplete 方法。如果只需要处理特定 CompletableFuture 抛出的异常，可以使用 exceptionally 方法。

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> {
    throw new RuntimeException("Task 1 failed");
}).thenApply(result -> {
    return result.toUpperCase();
}).handle((result, ex) -> {
    if (ex != null) {
        System.err.println("Exception occurred: " + ex.getMessage());
        return "Fallback Result";
    } else {
        return result;
    }
});
 
try {
    System.out.println(cf.get());
} catch (Exception e) {
    System.err.println("Exception caught: " + e.getMessage());
}

Lambda 表达式中的异常处理不当

在使用 Lambda 表达式时，需要特别注意异常处理。如果在 Lambda 表达式中抛出受检异常（checked exception），那么需要显式地捕获并处理它，或者将其转换为非受检异常（unchecked exception）抛出。否则，编译器会报错。

示例代码：

// 编译错误
/*CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> {
    try {
        throw new IOException("File not found");
    } catch (IOException e) {
        // 缺少处理
    }
    return "Result";
});*/
 
CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> {
    try {
        throw new IOException("File not found");
    } catch (IOException e) {
        throw new RuntimeException(e); // 将受检异常转换为非受检异常
    }
    return "Result";
});
 
try {
    System.out.println(cf.get());
} catch (Exception e) {
    System.err.println("Exception caught: " + e.getMessage());
    if (e.getCause() != null){
        System.err.println("Cause: " + e.getCause().getMessage());
    }
 
}

在这个例子中，Lambda 表达式中抛出了一个 IOException。由于 IOException 是一个受检异常，因此需要在 Lambda 表达式中显式地捕获并处理它，或者将其转换为非受检异常抛出。

解决方案：

在 Lambda 表达式中，要么捕获并处理受检异常，要么将其转换为非受检异常抛出。

忘记调用 get() 或 join() 方法

如果 CompletableFuture 链中的最后一个 CompletableFuture 的结果没有被获取，那么异常可能不会被抛出。这是因为 CompletableFuture 的异常处理是基于事件驱动的，只有当结果被请求时，异常才会被传播。

示例代码：

CompletableFuture.supplyAsync(() -> {
    throw new RuntimeException("Task failed");
}).thenAccept(result -> {
    System.out.println("Result: " + result);
});
 
// 没有调用 get() 或 join() 方法

在这个例子中， supplyAsync 抛出了一个 RuntimeException，但是由于没有调用 get() 或 join() 方法，因此异常不会被传播。程序可能会继续执行，而没有意识到错误。

解决方案：

确保调用 get() 或 join() 方法来获取 CompletableFuture 的结果，以便触发异常处理。

CompletableFuture<Void> cf = CompletableFuture.supplyAsync(() -> {
    throw new RuntimeException("Task failed");
}).thenAccept(result -> {
    System.out.println("Result: " + result);
});
 
try {
    cf.join(); // 或者 cf.get()
} catch (Exception e) {
    System.err.println("Exception caught: " + e.getMessage());
}

线程池配置不当

CompletableFuture 通常使用线程池来执行异步任务。如果线程池的配置不当，例如线程池大小过小，或者使用了错误的拒绝策略，那么可能会导致任务无法执行，或者异常无法正确传播。

示例代码：

ExecutorService executor = Executors.newFixedThreadPool(1); // 线程池大小为 1
 
CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(1000); // 模拟耗时操作
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return "Result from Task 1";
}, executor);
 
CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> {
    throw new RuntimeException("Task 2 failed");
}, executor);
 
try {
    cf1.get();
    cf2.get();
} catch (Exception e) {
    System.err.println("Exception caught: " + e.getMessage());
} finally {
    executor.shutdown();
}

在这个例子中，线程池大小为 1。 cf1 正在执行一个耗时操作，而 cf2 试图立即执行。由于线程池只有一个线程，因此 cf2 必须等待 cf1 完成才能执行。如果 cf1 抛出异常，那么 cf2 将永远无法执行，异常也无法被传播。

解决方案：

根据实际情况配置线程池，确保线程池大小足够大，并且使用了合适的拒绝策略。

ExecutorService executor = Executors.newFixedThreadPool(4); // 线程池大小为 4
 
CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(1000); // 模拟耗时操作
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return "Result from Task 1";
}, executor);
 
CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> {
    throw new RuntimeException("Task 2 failed");
}, executor);
 
try {
    cf1.get();
    cf2.get();
} catch (Exception e) {
    System.err.println("Exception caught: " + e.getMessage());
} finally {
    executor.shutdown();
}

三、最佳实践

为了避免 CompletableFuture 并发链中异常传播失败，以下是一些最佳实践：

四、案例分析

为了更好地理解异常传播失败的原因和解决方案，我们来看一个更复杂的案例。假设我们需要从多个数据源获取数据，并将这些数据合并成一个结果。如果任何一个数据源获取数据失败，那么整个过程都应该失败。

public class DataAggregator {
 
    private final ExecutorService executor;
 
    public DataAggregator(ExecutorService executor) {
        this.executor = executor;
    }
 
    public CompletableFuture<String> aggregateData() {
        CompletableFuture<String> source1 = fetchDataFromSource("Source 1");
        CompletableFuture<String> source2 = fetchDataFromSource("Source 2");
        CompletableFuture<String> source3 = fetchDataFromSource("Source 3");
 
        return CompletableFuture.allOf(source1, source2, source3)
                .thenApply(v -> {
                    try {
                        return source1.join() + ", " + source2.join() + ", " + source3.join();
                    } catch (Exception e) {
                        throw new RuntimeException("Failed to aggregate data", e);
                    }
                })
                .exceptionally(ex -> {
                    System.err.println("Aggregation failed: " + ex.getMessage());
                    return "Aggregation failed";
                });
    }
 
    private CompletableFuture<String> fetchDataFromSource(String sourceName) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("Fetching data from " + sourceName);
            if (sourceName.equals("Source 2")) {
                throw new RuntimeException("Failed to fetch data from " + sourceName);
            }
            return "Data from " + sourceName;
        }, executor);
    }
 
    public static void main(String[] args) throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(3);
        DataAggregator aggregator = new DataAggregator(executor);
 
        CompletableFuture<String> result = aggregator.aggregateData();
 
        try {
            System.out.println("Result: " + result.get());
        } catch (Exception e) {
            System.err.println("Main Exception: " + e.getMessage());
        } finally {
            executor.shutdown();
        }
    }
}

在这个例子中， aggregateData 方法从三个数据源获取数据，并将这些数据合并成一个结果。 fetchDataFromSource 方法模拟从数据源获取数据，其中 "Source 2" 总是会抛出异常。

在这个案例中， CompletableFuture.allOf 确保所有数据源都完成（无论成功或失败）后，才会执行 thenApply 方法。在 thenApply 方法中，我们使用 join() 方法获取每个数据源的结果。如果任何一个数据源抛出异常，那么 join() 方法会抛出一个 CompletionException。我们在 thenApply 方法中捕获 CompletionException，并将其包装在一个 RuntimeException 中抛出。最后，我们使用 exceptionally 方法来处理聚合过程中可能出现的异常。

这个案例展示了如何使用 CompletableFuture 来处理并发任务中的异常，并确保异常能够正确传播。

五、表格总结 CompletableFuture 异常处理方法

六、结论

CompletableFuture 提供了强大的并发编程能力，但也需要谨慎处理异常，避免异常传播失败。理解异常传播失败的常见原因，并采取相应的解决方案，可以帮助我们编写更加健壮和可靠的并发程序。记住，始终处理异常，选择合适的异常处理方法，注意 Lambda 表达式中的异常处理，确保调用 get() 或 join() 方法，以及合理配置线程池，这些都是避免异常传播失败的关键。

异常处理是构建稳定、可靠并发应用的重要组成部分