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_05 x64,Windows 7上.

我在这里错过了什么吗？

Answer 1

问题是你从未要求接收你的电话结果linksCF.thenAccept(..).

您的电话linksCF.get()会等待链中的执行结果.但它只会返回当前linksCF未来的结果.这不包括断言的结果.

linksCF.thenAccept(..)将返回一个新的CompletableFuture实例.在新返回的CompletableFuture实例上获取异常抛出调用get()或检查异常状态isCompletedExceptionally().

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

Answer 2

尽管Gregor Koukkoullis(+1)基本上已经回答了这个问题,但这是我为测试这个而创建的MCVE.

有几种方法可以获得导致内部问题的实际异常.但是,我不明白为什么要求get回归的未来thenAccept应该成为一个问题.有疑问,你也可以使用thenApply身份功能,并使用一个很好的流畅模式,如

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();
        }
    }
}