java - 在调用 CompletableFuture.completeExceptionally(Throwable) 之前捕获 throwable 与异常

Question

我知道网络上充斥着不要捕获 Throwable 的建议，但是它适用于使用 CompleteableFuture 吗？例如，在

    ExecutorService es = Executors.newFixedThreadPool(2);
    CompletableFuture<String> cf = new CompletableFuture<>();
    es.submit(() -> {
        try {
            cf.complete(getValue());
        } catch (Exception e) { // should I catch Throwable instead?
            cf.completeExceptionally(e);
        }
    });

我应该捕获 Throwable 还是 Exception？如果我捕捉到 Exception 并抛出 Error，程序很可能会陷入死锁。

Answer 1

你是对的，不捕获 Throwable 会承担 CompletableFuture 永远无法完成的风险。因此，这种风险将证明捕获 Error 是合理的。您通常不会使用的所有 Throwable。

但由于您基本上是在重新发明 supplyAsync，让我们比较一下两者的行为:

public class CF {
    public static void main(String[] args) {
        ExecutorService es = Executors.newFixedThreadPool(2);

        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> getValue(), es);
        try {
            cf1.join();
        }
        catch(CompletionException ex) {
            ex.printStackTrace();
        }

        System.err.println();

        CompletableFuture<String> cf2 = new CompletableFuture<>();
        es.submit(() -> {
            try {
                cf2.complete(getValue());
            } catch(Throwable t) {
                cf2.completeExceptionally(t);
            }
        });

        try {
            cf2.join();
        }
        catch(CompletionException ex) {
            ex.printStackTrace();
        }
    }

    private static String getValue() {
        throw new OutOfMemoryError();
    }
}

java.util.concurrent.CompletionException: java.lang.OutOfMemoryError
    at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:314)
    at java.base/java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:319)
    at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1702)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: java.lang.OutOfMemoryError
    at CF.getValue(CF.java:39)
    at CF.lambda$main$0(CF.java:11)
    at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1700)
    ... 3 more

java.util.concurrent.CompletionException: java.lang.OutOfMemoryError
    at java.base/java.util.concurrent.CompletableFuture.reportJoin(CompletableFuture.java:412)
    at java.base/java.util.concurrent.CompletableFuture.join(CompletableFuture.java:2044)
    at CF.main(CF.java:31)
Caused by: java.lang.OutOfMemoryError
    at CF.getValue(CF.java:39)
    at CF.lambda$main$1(CF.java:24)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:834)

所以我们看到，捕获错误是CompletableFuture的标准行为。只有堆栈跟踪不同。让我们看看是否可以提高收敛性:

public class CF {
    public static void main(String[] args) {
        ExecutorService es = Executors.newFixedThreadPool(2);

        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> getValue(), es);
        try {
            cf1.join();
        }
        catch(CompletionException ex) {
            ex.printStackTrace();
        }

        System.err.println();

        CompletableFuture<String> cf2 = new CompletableFuture<>();
        es.submit(() -> {
            try {
                cf2.complete(getValue());
            } catch(Throwable t) {
                cf2.completeExceptionally(new CompletionException(t));
            }
        });

        try {
            cf2.join();
        }
        catch(CompletionException ex) {
            ex.printStackTrace();
        }
    }

    private static String getValue() {
        throw new OutOfMemoryError();
    }
}

java.util.concurrent.CompletionException: java.lang.OutOfMemoryError
    at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:314)
    at java.base/java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:319)
    at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1702)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: java.lang.OutOfMemoryError
    at CF.getValue(CF.java:39)
    at CF.lambda$main$0(CF.java:11)
    at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1700)
    ... 3 more

java.util.concurrent.CompletionException: java.lang.OutOfMemoryError
    at CF.lambda$main$1(CF.java:26)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: java.lang.OutOfMemoryError
    at CF.getValue(CF.java:39)
    at CF.lambda$main$1(CF.java:24)
    ... 5 more

那很安静。

当然，如果 getValue() 不抛出检查异常，就没有理由不使用 CompletableFuture.supplyAsync(() -> getValue(), es);。如果我们有理由手动实现完成，比如必须处理已检查的异常，那么我们可以改进一些事情。如果我们不需要 submit 返回的 Future，我们可以使用 execute 来代替，以避免创建不需要的 FutureTask 。此外，CompletableFuture 用 AsynchronousCompletionTask 标记它的完成任务。有助于监控和调试，遵循约定很有用:

CompletableFuture<String> cf2 = new CompletableFuture<>();
es.execute((Runnable & CompletableFuture.AsynchronousCompletionTask)() -> {
    try {
        cf2.complete(getValue());
    } catch(Throwable t) {
        cf2.completeExceptionally(new CompletionException(t));
    }
});

虽然对实际行为没有直接影响。