CompletableFuture 的实例无法获得预期结果

Question

    CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
        System.out.println("enter into completableFuture()");
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("start to out of completableFuture()");
        return "a";
    });

    System.out.println("do something else");

    cf1.thenApply(v -> v + " b").thenAcceptAsync(v ->
            System.out.println(v)
    );

    System.out.println("finalize...");

    //cannot get expected result, if this line was comment out.
    //TimeUnit.SECONDS.sleep(10);

代码如上。

在 jdk8 中写一个使用 CompletableFuture 的例子时，我很困惑。

我必须添加最后一行

TimeUnit.SECONDS.sleep(10);

得到预期的结果。

如果我不让其主线程休眠，我想知道程序是否已经结束。如果没有，为什么我不能得到输出？

非常感谢您的宝贵时间。

Answer 1

当没有非守护线程在运行时，JVM 会终止，因此如果异步操作仅由守护线程执行，它将在主线程终止时终止，而不是继续后台操作。

有几种方法可以解决这个问题。

1. 如果后台计算形成单个依赖链，您可以使用最后一个操作等待其完成，因为它的完成意味着所有先前阶段的完成。让主线程等待直到完成，将 JVM 的终止推迟到该点：

   CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
       System.out.println("enter into completableFuture()");
       LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(1));
       System.out.println("start to out of completableFuture()");
       return "a";
   });
   System.out.println("do something else");
   CompletableFuture last
       = cf1.thenApply(v -> v + " b").thenAcceptAsync(System.out::println);
   System.out.println("finalize...");
   last.join();
   

2. 考虑documentation of CompletableFuture：

   所有没有显式 Executor 参数的 async 方法都使用ForkJoinPool.commonPool() 执行（除非它不支持至少两个并行级别，在这种情况下，会创建一个新线程来运行每个任务）。

   由于使用守护线程是该 F/J 公用池的属性，因此我们可以使用该知识来等待在这种情况下所有待处理任务的完成，这与这些待处理任务之间的依赖关系无关：

   CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
       System.out.println("enter into completableFuture()");
       LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(1));
       System.out.println("start to out of completableFuture()");
       return "a";
   });
   System.out.println("do something else");
   cf1.thenApply(v -> v + " b").thenAcceptAsync(System.out::println);
   System.out.println("finalize...");
   if(ForkJoinPool.getCommonPoolParallelism()>1)
       ForkJoinPool.commonPool().awaitQuiescence(1, TimeUnit.DAYS);
   

3. 使用显式执行程序，它不会使用守护线程。 JRE提供的线程池执行器，抛开ForkJoinPool，默认使用非守护线程：

   final ExecutorService threadPool = Executors.newCachedThreadPool();
   CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
       System.out.println("enter into completableFuture()");
       LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(1));
       System.out.println("start to out of completableFuture()");
       return "a";
   }, threadPool);
   System.out.println("do something else");
   cf1.thenApply(v -> v + " b").thenAcceptAsync(System.out::println);
   System.out.println("finalize...");
   threadPool.shutdown();
   

   请注意，threadPool.shutdown(); 并不意味着等待，也不会停止待处理的任务；它只会停止接受新任务，并确保一旦处理完所有待处理的任务，池中的线程将终止。与supplyAsync 一起使用后，您可以直接放置它，而无需更改行为。

   所以第三种解决方案是唯一让main 线程退出的解决方案，JVM 会继续运行，直到处理完所有待处理的后台任务，因为它们在非守护线程中运行。

Answer 2

您可以暂停CompletableFuture，直到CompletableFuture#join 完成，例如：

CompletableFuture<Void> stage = cf1.thenApply(v -> v + " b").thenAcceptAsync(v ->
        System.out.println(v)
);

System.out.println("finalize...");

//   v--- the main thread wait until the stage is completed
stage.join();