CompletableFuture吞咽异常？

Question

我一直在玩CompletableFuture，发现了一件奇怪的事情。

String url = "http://google.com";

CompletableFuture<String> contentsCF = readPageCF(url);
CompletableFuture<List<String>> linksCF = contentsCF.thenApply(_4_CompletableFutures::getLinks);

linksCF.thenAccept(list -> {
    assertThat(list, not(empty()));
});

linksCF.get();

如果在我的thenAccept调用中，断言失败，则不会传播异常。当时我尝试了更丑陋的东西：

linksCF.thenAccept(list -> {
    String a = null;
    System.out.println(a.toString());
});

不会发生任何事情，不会传播任何异常。我尝试使用诸如handle等方法以及与CompletableFutures中异常相关的其他方法，但失败了-没有一个方法像预期的那样传播异常。

当我调试CompletableFuture时，它会捕获如下异常：

final void internalComplete(T v, Throwable ex) {
    if (result == null)
        UNSAFE.compareAndSwapObject
            (this, RESULT, null,
             (ex == null) ? (v == null) ? NIL : v :
             new AltResult((ex instanceof CompletionException) ? ex :
                           new CompletionException(ex)));
    postComplete(); // help out even if not triggered
}

没有别的了。

我使用的是JDK 1.8.0\U 05 x64，Windows 7。

我错过什么了吗？

Answer 1

如果在thenAccept调用中，断言失败，则不会传播异常。

您注册的continuation（继续）与linksCF（继续）是一项单独的任务。linksCF任务成功完成；报告时没有错误。它有它的最终价值。linksCF引发的异常只应指示产生linksCF结果的问题；如果使用结果的其他代码片段抛出，这并不表示生成结果失败。

要观察延续中发生的异常，必须观察延续的可完成未来。

正确。但是1）我不应该被迫调用get（）-新结构的要点之一；2）它被包装在ExecutionException中

如果您想使用accept（）将结果传递给多个独立的continuations，该怎么办？如果要抛出其中一个连续体，为什么会影响父连续体或其他连续体？

如果要将链接scf视为链中的节点并观察链中发生的结果（以及任何异常），则应在链中的最后一个链接上调用get（）。

您可以通过使用connect（）而不是get（）来避免选中ExecutionException，这将把错误包装在未选中的CompletionException中（但它仍然被包装）。

Answer 2

虽然Gregor Koukkoullis（1）基本上已经回答了这个问题，但我创建了一个MCVE来测试这个问题。

有几个选项可用于获取导致内部问题的实际异常。然而，我不明白为什么对由accept返回的future调用get应该是一个问题。毫无疑问，您还可以使用identity函数使用然后再应用，并使用一个流畅的模式，如

List<String> list = 
    readPage().
    thenApply(CompletableFutureTest::getLinks).
    thenApply(t -> {
        // check assertion here
        return t;
    }).get();

但也许有一个特别的原因让你想避免这种情况。

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.function.Supplier;

public class CompletableFutureTest
{
    public static void main(String[] args) 
        throws InterruptedException, ExecutionException
    {
        CompletableFuture<String> contentsCF = readPage();
        CompletableFuture<List<String>> linksCF = 
            contentsCF.thenApply(CompletableFutureTest::getLinks);

        CompletableFuture<Void> completionStage = linksCF.thenAccept(list -> 
        {
            String a = null;
            System.out.println(a.toString());
        });        

        // This will NOT cause an exception to be thrown, because
        // the part that was passed to "thenAccept" will NOT be
        // evaluated (it will be executed, but the exception will
        // not show up)
        List<String> result = linksCF.get();
        System.out.println("Got "+result);


        // This will cause the exception to be thrown and
        // wrapped into an ExecutionException. The cause
        // of this ExecutionException can be obtained:
        try
        {
            completionStage.get();
        }
        catch (ExecutionException e)
        {
            System.out.println("Caught "+e);
            Throwable cause = e.getCause();
            System.out.println("cause: "+cause);
        }

        // Alternatively, the exception may be handled by
        // the future directly:
        completionStage.exceptionally(e -> 
        { 
            System.out.println("Future exceptionally finished: "+e);
            return null; 
        });

        try
        {
            completionStage.get();
        }
        catch (Throwable t)
        {
            System.out.println("Already handled by the future "+t);
        }

    }

    private static List<String> getLinks(String s)
    {
        System.out.println("Getting links...");
        List<String> links = new ArrayList<String>();
        for (int i=0; i<10; i++)
        {
            links.add("link"+i);
        }
        dummySleep(1000);
        return links;
    }

    private static CompletableFuture<String> readPage()
    {
        return CompletableFuture.supplyAsync(new Supplier<String>() 
        {
            @Override
            public String get() 
            {
                System.out.println("Getting page...");
                dummySleep(1000);
                return "page";
            }
        });
    }

    private static void dummySleep(int ms)
    {
        try
        {
            Thread.sleep(ms);
        }
        catch (InterruptedException e)
        {
            e.printStackTrace();
            Thread.currentThread().interrupt();
        }
    }
}

Answer 3

问题是您从不请求接收对link sCF. thenAccep（...）的调用结果。

您对linksCF的呼叫。get（）将等待链中的执行结果。但它只会返回linksCF future的结果。这不包括您的断言的结果。

link sCF. thenAccep（...）将返回一个新的CompletableFuture实例。要获取抛出的异常，请调用get（）或在新返回的CompletableFuture实例上使用isCompletedExcep（）检查异常状态。

CompletableFuture<Void> acceptedCF = linksCF.thenAccept(list -> {
    assertThat(list, not(empty()));
});

acceptedCF.exceptionally(th -> {
    // will be executed when there is an exception.
    System.out.println(th);
    return null;
});
acceptedCF.get(); // will throw ExecutionException once results are available

可供替代的

CompletableFuture<List<String>> appliedCF = linksCF.thenApply(list -> {
    assertThat(list, not(empty()));
    return list;
});

appliedCF.exceptionally(th -> {
    // will be executed when there is an exception.
    System.out.println(th);
    return Coolections.emptyList();
});
appliedCF.get(); // will throw ExecutionException once results are available