java - 如何取消 Java 8 可完成的 future ？

Question

我正在玩 Java 8 可完成的 future 。我有以下代码:

CountDownLatch waitLatch = new CountDownLatch(1);

CompletableFuture<?> future = CompletableFuture.runAsync(() -> {
    try {
        System.out.println("Wait");
        waitLatch.await(); //cancel should interrupt
        System.out.println("Done");
    } catch (InterruptedException e) {
        System.out.println("Interrupted");
        throw new RuntimeException(e);
    }
});

sleep(10); //give it some time to start (ugly, but works)
future.cancel(true);
System.out.println("Cancel called");

assertTrue(future.isCancelled());

assertTrue(future.isDone());
sleep(100); //give it some time to finish

我使用 runAsync 安排执行等待闩锁的代码。接下来我取消了future，期待里面抛出一个中断的异常。但似乎线程在 await 调用中仍然被阻塞，并且即使 future 被取消(断言通过)也永远不会抛出 InterruptedException 。使用 ExecutorService 的等效代码按预期工作。这是 CompletableFuture 中的错误还是我的示例中的错误？

Answer 1

当您调用 CompletableFuture#cancel 时，您只会停止链的下游部分。上游部分，i。 e.最终将调用 complete(...) 或 completeExceptionally(...) 的东西，不会得到任何不再需要结果的信号。

那些“上游”和“下游”的东西是什么？

让我们考虑以下代码:

CompletableFuture
        .supplyAsync(() -> "hello")               //1
        .thenApply(s -> s + " world!")            //2
        .thenAccept(s -> System.out.println(s));  //3

在这里，数据从上到下流动——从供应商创建，通过功能修改，到被 println 使用。特定步骤上方的部分称为上游，下方的部分称为下游。例如步骤 1 和步骤 2 是步骤 3 的上游。

这是幕后发生的事情。这并不精确，而是对正在发生的事情的一个方便的思维模型。

1. Supplier(第 1 步)正在执行(在 JVM 的通用 ForkJoinPool 内)。
2. 供应商的结果随后由 complete(...) 传递到下一个 CompletableFuture 下游。
3. 收到结果后，CompletableFuture 调用下一步 - 一个函数(第 2 步)接收上一步结果并返回将进一步传递给下游 CompletableFuture 的 complete(...)。
4. 收到第 2 步的结果后，第 3 步 CompletableFuture 调用消费者 System.out.println(s)。消费者完成后，下游CompletableFuture会收到它的值，(Void) null

正如我们所见，此链中的每个 CompletableFuture 都必须知道下游有谁在等待将值传递给它们的 complete(...) (或 completeExceptionally(...))。但是 CompletableFuture 不必知道任何关于它的上游(或上游 - 可能有多个)的信息。

因此，在第 3 步调用 cancel() 不会中止第 1 步和第 2 步，因为没有从第 3 步到第 2 步的链接。

如果您正在使用 CompletableFuture，那么您的步骤应该足够小，因此如果执行几个额外的步骤也不会造成损害。

如果您希望将取消传播到上游，您有两种选择:

• 自己实现 - 创建一个专用的 CompletableFuture(将其命名为 cancelled)，它在每一步之后都会被检查(类似于 step.applyToEither(cancelled, Function .identity()))
• 使用 RxJava 2、ProjectReactor/Flux 或 Akka Streams 等响应式(Reactive)堆栈