multithreading - 在 supplyAsync 阻塞主线程后使用 thenAccept

Question

我正在开发一个与其他网络应用程序通信的网络应用程序。有时，我的系统会向其他系统发送 HTTP 请求作为通知。由于他们的响应对我来说不是必需的，我使用 Java 8 CompletableFuture supplyAsync 发送请求并使用 thenAccept 打印他们的响应，这样我的主线程就不会被阻塞。然而，我发现 CompletableFuture 函数链每次大约需要 100 到 200 毫秒，这让我感到困惑，因为根据我的理解，thenAccept() 应该与 supplyAsync() 在同一个线程中运行。

我用下面的代码模拟了我的过程

public static void run() {
    long start = System.currentTimeMillis();
    log.info("run start -> " + new Timestamp(start));
    CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 42;
    }).thenAccept(res -> log.info("run result -> " + res + ", time -> " + new Timestamp(System.currentTimeMillis())));
    log.info("run duration ->" + (System.currentTimeMillis() - start));
}

public static void runAsync() {
    long start = System.currentTimeMillis();
    log.info("runAsync start -> " + new Timestamp(start));
    CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 42;
    }).thenAcceptAsync(res -> log.info("runAsync result -> " + res + ", time ->" + new Timestamp(System.currentTimeMillis())));
    log.info("runAsync duration ->" + (System.currentTimeMillis() - start));
}

public static void main(String[] args) throws InterruptedException {
    Test.run();
    Test.runAsync();
    Thread.sleep(1000);
}

run() 方法使用 thenAccept() 和 supplyAsync()，而 runAsync() 使用 thenAcceptAsync()。我希望它们都只需要几毫秒。然而，真正的输出是:

10:04:54.632 [main] INFO Test - run start -> 2017-12-08 10:04:54.622
10:04:54.824 [main] INFO Test - run duration ->202
10:04:54.824 [main] INFO Test - runAsync start -> 2017-12-08 10:04:54.824
10:04:54.826 [main] INFO Test - runAsync duration ->2
10:04:55.333 [ForkJoinPool.commonPool-worker-1] INFO Test - run result -> 42, time -> 2017-12-08 10:04:55.333
10:04:55.333 [ForkJoinPool.commonPool-worker-3] INFO Test - runAsync result -> 42, time ->2017-12-08 10:04:55.333

我们可以看到 run() 耗时 202 毫秒，是仅使用 2 毫秒的 runAsync() 持续时间的 100 倍。

我不明白 202 毫秒的开销从何而来，显然它不是 supplyAysnc() 中休眠 500 毫秒的 lambda 函数。

谁能解释为什么 run() 方法会阻塞，我应该始终使用 thenAcceptAsync() 而不是 thenAccept() 吗？

非常感谢。

Answer 1

…because from my understanding thenAccept() should run in the same thread with supplyAsync()'s

你的理解是错误的。

来自 the documentation of CompletableFuture :

• Actions supplied for dependent completions of non-async methods may be performed by the thread that completes the current CompletableFuture, or by any other caller of a completion method.

最明显的结果是，当 future 已经完成时，传递给 thenAccept() 的函数将直接在调用者的线程中求值，因为 future 不可能命令线程完成了。事实上，CompletableFuture 与线程根本没有关联，因为任何人 都可以对其调用complete，而不仅仅是执行的线程您传递给 supplyAsync 的 Supplier。这也是cancel不支持中断的原因。 future 不知道哪个线程可能会尝试完成它。

不那么明显的后果是，即使是上述行为也不能得到保证。短语“或完成方法的任何其他调用者”并不限于注册依赖操作的完成方法的调用者。它也可以是任何其他调用者在同一个 future 上注册一个依赖操作。因此，如果两个线程在同一个 future 上同时调用 thenApply，那么它们中的任何一个都可能最终评估两个函数，甚至更奇怪的是，每个线程最终都可能执行另一个线程的操作。规范并不排除它。

对于您在问题中提供的测试用例，您更有可能测量初始化开销，如this answer 中所述。 .但是对于框架只会初始化一次的 Web 应用程序中的实际问题，您可能会因错误理解thenApply 的行为(或任何非异步 一般的链接方法)。如果您想确保评估不会在调用者的线程中发生，您必须使用 thenApplyAsync。