为什么我的线程不等待CompletableFutures以“allOf（）`”完成？

Question

我正在学习Java 1.8中的CompletableFuture，在理解所有代码时遇到了困难。主线程似乎不会等待任何可完成的未来完成。

看见https://github.com/nurkiewicz/reactive/blob/master/src/test/java/be/more/reactive/S03_AllOf.java例如，我正在测试。

测试作业在打印任何结果之前完成。

有两种（丑陋的？）方法可以规避这一点：1）在主线程上设置超时并等待两者都完成。2）在末尾设置一个. get（），它将成为一个阻塞任务。

这是为什么呢？

代码片段：

package be.more.reactive;

import be.more.reactive.util.BaseTest;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CompletableFuture;

public class S03_AllOf extends BaseTest {

    private static final Logger log = LoggerFactory.getLogger(S03_AllOf.class);

    private final CompletableFuture<String> futureResult1 = getFutureQueryResult("1"); //.exceptionally() ??
    private final CompletableFuture<String> futureResult2 = getFutureQueryResult("2");
    private final CompletableFuture<String> futureResult3 = getFutureQueryResult("3");
    private final CompletableFuture<String> futureResult4 = getFutureQueryResult("4");

    @Test
    public void allOf() throws Exception {
        final CompletableFuture<Void> futureResult = CompletableFuture.allOf(   //Void ?? I want List<String>
                futureResult1, futureResult2, futureResult3, futureResult4
        );

//        futureResult.thenAccept((Void vd) -> vd.??)   //no, it won't work

        futureResult.thenRun(() -> {
            try {
                log.debug("Query result 1: '{}'", futureResult1.get());
                log.debug("Query result 2: '{}'", futureResult2.get());
                log.debug("Query result 3: '{}'", futureResult3.get());
                log.debug("Query result 4: '{}'", futureResult4.get());   //a lot of manual work

                log.debug("Now do on complete");    //handling onComplete
            } catch (Exception e) {
                log.error("", e);
            }
        });

    }

}

在BaseTest中：

protected CompletableFuture<String> getFutureQueryResult(final String queryId) {
    return CompletableFuture.supplyAsync(
            () -> db.apply(new Query(queryId))

    );
}

以DB为单位。Java语言

package be.more.reactive.db;

import java.util.concurrent.TimeUnit;

import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
import static org.apache.commons.lang3.RandomUtils.nextInt;
import static org.apache.commons.lang3.RandomUtils.nextLong;

public class DB {
    public String apply(Query query) {
        try {
            TimeUnit.SECONDS.sleep(nextLong(2, 4));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return String.format("%s_%s", randomAlphabetic(nextInt(4, 12)), query.getId());
    }
}

Answer 1

您看到的行为并非意外。当您创建一个可完成的未来时，您基本上是在计划一个异步运行的工作。

我们可以这样创建一个可完成的未来

var myFuture = CompletableFuture.supplyAsync(() -> myLongRunningOperation());

>

CompletableFuture将在单独的线程上调用myLongRunningAction。

CompletableFuture使用ExecutorService运行任务，可以在创建CompletableFuture期间提供。

如果未提供ExecutorService，它将使用ForkJoinPool#commonPool提供的服务，该服务提供守护程序线程的线程池。

任务（）-

在您的测试方法中，这是幕后发生的事情

@Test
public void allOf() throws Exception {
    // Schedules a computation (futureResult) on a different thread whose only 
    // work is to wait for the futures futureResult1, futureResult2, futureResult3 
    // and futureResult4 to complete
    final CompletableFuture<Void> futureResult = CompletableFuture.allOf(
            futureResult1, futureResult2, futureResult3, futureResult4
    );

    //  Schedules a computation that prints the results AFTER the futureResult is complete.
    futureResult.thenRun(() -> {
        try {
            log.debug("Query result 1: '{}'", futureResult1.get());
            log.debug("Query result 2: '{}'", futureResult2.get());
            log.debug("Query result 3: '{}'", futureResult3.get());
            log.debug("Query result 4: '{}'", futureResult4.get());
            log.debug("Now do on complete");
        } catch (Exception e) {
            log.error("", e);
        }
    });

    // Nothing more to do, so exit

}

但是，当您调用时。join（）或。get（）执行测试的线程（主线程）将在退出之前等待计划的计算完成。

因此，如果您希望您的测试等待计划的计算在它存在之前完成，

//  Schedules a computation that prints the results ONCE the futureResult is complete.
final CompletableFuture<Void> myFuture = futureResult.thenRun(() -> {
    try {
        log.debug("Query result 1: '{}'", futureResult1.get());
        log.debug("Query result 2: '{}'", futureResult2.get());
        log.debug("Query result 3: '{}'", futureResult3.get());
        log.debug("Query result 4: '{}'", futureResult4.get());   //a lot of manual work

        log.debug("Now do on complete");    //handling onComplete
    } catch (Exception e) {
        log.error("", e);
    }
});

// Wait for the myFuture to complete (sucessfully or throw an exception) before continuing.
myFuture.get();

在主线程上设置超时以等待将来完成是一种反模式。

>

如果您不关心结果，但希望应用程序等待将来完成，那么请创建一个自定义执行器服务，创建非守护进程线程。

Answer 2

来自Javadoc

返回一个新的CompletableFuture，该新的CompletableFuture在所有给定CompletableFutures完成时完成。

未来是一个异步任务，在调用get之前不会阻塞（仅当任务仍在运行时才会阻塞）。

在这种情况下，CompleteableFuture是所有CompletableFutures的复合Future。这个未来仍然是一个阻塞异步调用，您必须调用get或加入以等待所有期货完成。同样，来自javadoc

在继续一个程序之前完成未来，如：CompletableFuture。所有（c1、c2、c3）。join（） 。

在我看来，你的解决方案既不是丑陋的，也不是意外的功能。