如何并行运行boolean类型的CompletableFutures，并在所有任务返回TRUE时在第一个假结果或返回时退出？

Question

所以我有一个用例，我有四个任务a，B，C，D，我想并行运行。这些任务的响应类型是布尔型的。

在并行运行它们之后，我想在以下任一条件下停止执行

1. 从任一任务接收第一个false时返回。
2. 4个任务全部完成，没有一个返回false。

我的项目使用的是Java，springboot

所以我试图使用可完成的未来来实现这一点。我可以在第一步成功，就像下面的代码一样。但不是第二步。有什么帮助吗？

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.function.Consumer;
import java.util.function.Predicate;


public class CompFutu {
    public static <T> CompletableFuture<T> firstMatchingCondition(
            List<? extends CompletionStage<? extends T>> work,
            Predicate<? super T> successFullResultCondition) {

        CompletableFuture<T> result = new CompletableFuture<>();

        Consumer<T> tester = workUnit -> {
            if (successFullResultCondition.test(workUnit)) {
                result.complete(workUnit);
            }
        };

        work.forEach(s -> s.thenAccept(tester));
        return result;
    }

    public static void main(String[] args) throws Exception {

        List<CompletableFuture<Boolean>> list= Arrays.asList(
                CompletableFuture.supplyAsync(()-> {
                    try {
                        System.err.println(Thread.currentThread().getName());
                        Thread.sleep(1000);
                        System.err.println("DONE(true 1) "+Thread.currentThread().getName());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return true;
                }),
                CompletableFuture.supplyAsync(()-> {
                    try {
                        System.err.println(Thread.currentThread().getName());
                        Thread.sleep(20_000);
                        System.err.println("DONE(false 20) "+Thread.currentThread().getName());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return false;
                }),
                CompletableFuture.supplyAsync(()-> {
                    try {
                        System.err.println(Thread.currentThread().getName());
                        Thread.sleep(5000);
                        System.err.println("DONE(false 5) "+Thread.currentThread().getName());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return false;
                })
        );

        firstMatchingCondition(list, i->!i).join();

        System.err.println("Done");

    }
}

Answer 1

创造另一个未来，当所有的未来都完成时，它将被完成，除了你的未来，一旦你满足了条件，它将被完成。然后，将它们结合到一个未来中，一旦其中一个完成，这个未来就会完成：

CompletableFuture<?> firstFalse = firstMatchingCondition(list, i -> !i);
CompletableFuture<?> all
    = CompletableFuture.allOf(list.toArray(new CompletableFuture<?>[0]));

CompletableFuture.anyOf(firstFalse, all).join();

您可以将创建内联到单个表达式中。

CompletableFuture.anyOf(
    firstMatchingCondition(list, i -> !i),
    CompletableFuture.allOf(list.toArray(new CompletableFuture<?>[0]))
).join();

但重要的是要记住，这只会影响启动线程在继续之前等待的时间。这不会取消其他未来，而且由于CompletableFuture通常不支持中断，因此无法停止正在进行的评估。只有他们的结果会被忽略。