如何从与其关联的线程中发出 CompletableFutures 已完成或已取消的信号？

Question

我已经阅读了很多关于 CompletableFutures 的文章，他们关注 CompletableFuture 无法像 Future 那样访问底层计算代码这一事实。那么，如何从与 completableFuture 关联的任何任务中发出 complete() 或 cancel()（或检查 isCancelled()，如果您想从外部中断计算）？

编辑：让我感到困惑的一件事是，CF 被用于替代替代品，因为它的可组合性或手动可设置性，在我看来，如果实现是非常正交的，例如：

CompletableFuture.runAsync(() -> { 
                              if(myCondition) CF_REF?.complete();
                              else            CF_REF?.exceptionally();
                           }).thenApply(() -> {
                                        if (myOtherCondition) CF_REF_2?.complete();
 (...)

以一种不能同时“可完成”+“可组合”的方式设计它有什么意义？

我想找到一种方法来使用它们，就好像 CF 使用类似于假设的 CompletableCallable 的接口作为输入，但不知道如何使用。 Function<CompletableFuture,T> 之类的东西，而不是 Callables/Runnables，所以我们可以将其用作：

CompletableFuture CF_REF_BIS = CompletableFuture.runAsync((CF_REF) -> { 
                                  if(myCondition) CF_REF.complete();
                                  else CF_REF.exceptionally();
                               });

CF_REF_BIS.thenApply(...)

当要计算代码时，内部执行机制将返回 CF_REF_BIS 引用作为在未来计算中实际作为 CF_REF 传递的引用，而无需跨范围访问它。

这意味着我们可以重用去匿名化的计算代码 只需创建 new CompeltableCallable() 然后在任何地方提交 n 次，任何线程都可以访问 CompletableFuture 信号量

例如：

CompletableFuture CF_REF = CompletableFuture.runAsync(myCompletableCallable)
             .thenApply(myCompletableCallable) //again
             .thenApply(anotherCompletableCallable); 

有没有办法满足这种需求？我是不是忘记了什么？我的方法有什么根本的不一致吗？ Java CompletableFuture 是否还在半途而废的便捷设计中？

Answer 1

看看javadoc：

CompletableFuture#complete()

如果尚未完成，则设置get() 和相关返回的值 给定值的方法。

CompletableFuture#cancel()

如果尚未完成，请使用 CompletableFuture 完成 CancellationException。依赖CompletableFutures 没有 已经完成也将异常完成，带有 CompletionException 由这个 CancellationException 引起。

[...]

参数：

ma​​yInterruptIfRunning - 此值在此实现中无效，因为中断不用于控制处理。

CompletableFuture 对象与任何可能对其具有可访问引用的线程完全无关。换句话说，这些对象并不意味着向线程发出信号。这种类型的未来基本上是一个结果的持有者，可以选择注册听众。

whenXyz 和 thenAbc 的所有行为都注册到 CompletableFuture 对象。何时执行该行为取决于许多事情：未来哪个线程完成（成功或异常），是否使用*Async 方法注册了延续，等等。这在the javadoc 中列出。

---

例如，您创建一个CompletableFuture 并将其交给一些线程。据推测，这些线程之一将complete 它。这对其他线程没有影响。他们仍然继续做他们正在做或试图做的事情。

public static void main(String[] args) throws Exception {
    CompletableFuture<String> promise = new CompletableFuture<>();
    ExecutorService executorService = Executors.newFixedThreadPool(3);
    Runnable action = () -> {
        if (promise.complete("done")) {
            System.out.println("completed by " + Thread.currentThread());
        } else {
            System.out.println("somebody got there first");
        }
    };
    executorService.submit(action);
    executorService.submit(action);
    executorService.submit(action);

    executorService.shutdown();
    executorService.awaitTermination(100, TimeUnit.MILLISECONDS);
    System.out.println(promise.get());
}