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

Question

我读了很多关于CompletableFutures的书，他们关注的是一个事实，即CompletableFutures和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的接口作为输入一样，但不知道如何使用。比如函数

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的引用，而无需访问它。

这意味着我们只需创建新的CompletableCallable（），然后在任何线程可以访问CompletableFuture信号量的地方提交n次，就可以重用一个反匿名的计算代码

例如：

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

有没有办法满足这种需求？我忘了什么吗？我的方法有什么根本的不一致吗？Java CompletableFuture仍然是一种方便的设计吗？




Answer 1

看看javadoc：

CompletableFuture#complete（）

如果尚未完成，请将get（）和相关方法返回的值设置为给定值。

CompletableFuture#cancel（）

如果尚未完成，请使用C撤消Exception完成此CompletableFuture。尚未完成的依赖CompletableFutures也将异常完成，并使用由此引起的CompletionException。

[...]

参数：

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

CompletableFuture对象与任何可能对它们有可访问引用的线程完全无关。换句话说，这些对象并不意味着给线程发信号。这种类型的未来基本上是带有注册侦听器选项的结果的持有者。

whenXyz和thenAbc的所有行为都在CompletableFuture对象中注册。该行为何时执行取决于很多事情：哪个线程将来完成（成功或例外），是否使用*Async方法注册了continuations，等等。这在javadoc中有详细说明。

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

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