为什么在异常完成之前调用get()会等待异常执行？

Question

回答这个问题的时候,我注意到了一个奇怪的行为CompletableFuture:如果你有一个CompletableFuture cf和一个调用链cf.exceptionally(),调用cf.get()似乎表现得很奇怪:

• 如果在异常完成之前调用它,它会exceptionally()在返回之前等待块的执行
• 否则,它会立即失败 ExecutionException

我错过了什么或这是一个错误吗？我在Ubuntu 17.04上使用Oracle JDK 1.8.0_131.

以下代码说明了这种现象:

public static void main(String[] args) {
    long start = System.currentTimeMillis();
    final CompletableFuture<Object> future = CompletableFuture.supplyAsync(() -> {
        sleep(1000);
        throw new RuntimeException("First");
    }).thenApply(Function.identity());

    future.exceptionally(e -> {
        sleep(1000);
        logDuration(start, "Exceptionally");
        return null;
    });

    final CompletableFuture<Void> futureA = CompletableFuture.runAsync(() -> {
        try {
            future.get();
        } catch (Exception e) {
        } finally {
            logDuration(start, "A");
        }
    });

    final CompletableFuture<Void> futureB = CompletableFuture.runAsync(() -> {
        sleep(1100);
        try {
            future.get();
        } catch (Exception e) {
        } finally {
            logDuration(start, "B");
        }
    });

    try {
        future.join();
    } catch (Exception e) {
        logDuration(start, "Main");
    }

    futureA.join();
    futureB.join();
}

private static void sleep(final int millis) {
    try {
        Thread.sleep(millis);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}

private static void logDuration(long start, String who) {
    System.out.println(who + " waited for " + (System.currentTimeMillis() - start) + "ms");
}

输出:

B waited for 1347ms
Exceptionally waited for 2230ms
Main waited for 2230ms
A waited for 2230ms

正如你所看到的,futureB在调用之前睡了一会儿get()根本没有阻塞.但是,两者futureA和主线程都在等待exceptionally()完成.

请注意,如果您删除该行为不会发生此问题.thenApply(Function.identity()).

Answer 1

唤醒睡眠线程是一个依赖操作，必须像其他操作一样进行处理，并且没有优先级。另一方面，轮询 a 的线程CompletableFuture在完成后不会进入睡眠状态，不需要被唤醒，因此不需要与其他依赖操作竞争。

用下面的程序

public static void main(String[] args) {
    final CompletableFuture<Object> future = CompletableFuture.supplyAsync(() -> {
        waitAndLog("Supplier", null, 1000);
        throw new RuntimeException("First");
    }).thenApply(Function.identity());
    long start = System.nanoTime();

    CompletableFuture.runAsync(() -> waitAndLog("A", future, 0));

    LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10));

    future.exceptionally(e -> {
        waitAndLog("Exceptionally", null, 1000);
        return null;
    });

    CompletableFuture.runAsync(() -> waitAndLog("B", future, 0));
    CompletableFuture.runAsync(() -> waitAndLog("C", future, 1100));

    waitAndLog("Main", future, 0);
    ForkJoinPool.commonPool().awaitQuiescence(10, TimeUnit.SECONDS);
}
private static void waitAndLog(String msg, CompletableFuture<?> primary, int sleep) {
    long nanoTime = System.nanoTime();
    Object result;
    try {
        if(sleep>0) Thread.sleep(sleep);
        result = primary!=null? primary.get(): null;
    } catch (InterruptedException|ExecutionException ex) {
        result = ex;
    }
    long millis=TimeUnit.NANOSECONDS.toMillis(System.nanoTime()-nanoTime);
    System.out.println(msg+" waited for "+millis+"ms"+(result!=null? ", got "+result: ""));
}

我明白了，

public static void main(String[] args) {
    final CompletableFuture<Object> future = CompletableFuture.supplyAsync(() -> {
        waitAndLog("Supplier", null, 1000);
        throw new RuntimeException("First");
    }).thenApply(Function.identity());
    long start = System.nanoTime();

    CompletableFuture.runAsync(() -> waitAndLog("A", future, 0));

    LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10));

    future.exceptionally(e -> {
        waitAndLog("Exceptionally", null, 1000);
        return null;
    });

    CompletableFuture.runAsync(() -> waitAndLog("B", future, 0));
    CompletableFuture.runAsync(() -> waitAndLog("C", future, 1100));

    waitAndLog("Main", future, 0);
    ForkJoinPool.commonPool().awaitQuiescence(10, TimeUnit.SECONDS);
}
private static void waitAndLog(String msg, CompletableFuture<?> primary, int sleep) {
    long nanoTime = System.nanoTime();
    Object result;
    try {
        if(sleep>0) Thread.sleep(sleep);
        result = primary!=null? primary.get(): null;
    } catch (InterruptedException|ExecutionException ex) {
        result = ex;
    }
    long millis=TimeUnit.NANOSECONDS.toMillis(System.nanoTime()-nanoTime);
    System.out.println(msg+" waited for "+millis+"ms"+(result!=null? ", got "+result: ""));
}

在我的机器上，这表明在这种特定情况下，相关操作首先按照计划的顺序执行A。请注意，我在调度之前插入了额外的等待时间Exceptionally，这将是下一个相关操作。由于B在后台线程中运行，因此它是否能够在Main线程之前调度自身是不确定的。我们可以在其中一个之前插入另一个延迟来执行命令。

由于C轮询一个已经完成的 future，它可以立即继续，因此它的净等待时间接近明确指定的休眠时间。

必须强调的是，这只是特定场景的结果，取决于实现细节。相关操作没有保证的执行顺序。您可能已经注意到，如果没有该.thenApply(Function.identity())步骤，实现将运行不同的代码路径，从而导致相关操作的执行顺序不同。

依赖关系形成一棵树，并且实现必须以有效的方式遍历它，而不会有堆栈溢出的风险，因此它必须以某种方式将其展平，并且依赖关系树的形状的微小变化可能会影响结果的顺序。直观的方式。