为什么我的线程不等待 CompletableFutures 用 `allOf()` 完成？

Question

我正在学习 Java 1.8 中的 CompletableFuture，但在尝试理解 allOf 时遇到了困难。似乎主线程没有等待任何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

来自 Javadoc

返回一个新的 CompletableFuture，当所有给定的 CompletableFuture 都完成时，它就完成了。

Future 是一个异步任务，它在您调用 get 之前不会阻塞（只有在任务仍在运行时才会阻塞）。

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

继续程序之前的 CompletableFutures，如：CompletableFuture.allOf(c1, c2, c3).join();。

在我看来，您的 (2) 解决方案既不丑也不意外。

Answer 2

您看到的行为并不意外。当您创建一个CompletableFuture 时，您基本上是在安排一个异步运行的工作。

在使用allOf之前需要了解CompletableFuture

假设我们像这样创建CompletableFuture

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

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

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

• 如果没有提供ExecutorService，则使用ForkJoinPool#commonPool提供的，提供Daemon Threads的线程池。

• 任务() -> myLongRunningOperation() 将提交给ExecutorService，无论是否有人在等待myFuture 的结果，即无论是调用myFuture.join() 还是myFuture.get()。

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

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

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

• 如果你关心结果，需要等待未来 完成，根据您的需要致电join() 或get() 异常处理。

• 如果您不关心结果，但想申请 等待未来完成，然后创建一个自定义执行器 创建non-daemon 线程的服务。