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

Question

我知道网络上有很多关于不要抓住Throwables 的建议，但是在使用CompleteableFutures 时它是否适用？例如，在

    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 永远不会完成的风险。因此，这种风险将证明捕捉Errors 是合理的。所有你通常不会的Throwables。

但既然你基本上是在重新发明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));
    }
});

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