CompletableFuture中的静态ScheduledThreadPoolExecutor。延迟器

Question

在java-9中，在CompletableFuture类中引入了新方法completeOnTimeout

public CompletableFuture<T> completeOnTimeout(T value, long timeout,
                                              TimeUnit unit) {
    if (unit == null)
        throw new NullPointerException();
    if (result == null)
        whenComplete(new Canceller(Delayer.delay(
                                       new DelayedCompleter<T>(this, value),
                                       timeout, unit)));
    return this;
}

我不明白的是为什么它在其实现中使用静态SchduledThreadPoolExecitor：

    static ScheduledFuture<?> delay(Runnable command, long delay,
                                    TimeUnit unit) {
        return delayer.schedule(command, delay, unit);
    }

在哪里

    static final ScheduledThreadPoolExecutor delayer;
    static {
        (delayer = new ScheduledThreadPoolExecutor(
            1, new DaemonThreadFactory())).
            setRemoveOnCancelPolicy(true);
    }

对我来说，这是一种非常奇怪的方法，因为它可能会成为整个应用程序的瓶颈：唯一的schduledThreadPoolExecitor，只有一个线程保留在所有可能的CompletableFuture任务的池内？

我错过了什么？

P、 它看起来像：

>

此代码的作者不愿意提取此逻辑，而是更愿意重用ScheduledThreadPoolExecutor，

这显然导致了这样一个带有静态变量的解决方案，因为为每个可完成的未来创建一个新的执行器是非常低效的。

但我的怀疑仍然存在，因为我发现一般方法很奇怪。

Answer 1

当然，这是一个有待作者回答的问题。无论如何，这是我对这件事的看法。

我不明白的是为什么它在其实现中使用静态schduledThreadPoolExecitor：

...

对我来说，这是一种非常奇怪的方法，因为它可能会成为整个应用程序的瓶颈：唯一的schduledThreadPoolExecitor，只有一个线程保留在所有可能的CompletableFuture任务的池内？

你说得对。ScheduledThreadPoolExecutor可以运行任意代码。具体来说，orTimeout（）和completeOnTimeout（）将异常调用completeeexceptionally（）和complete（），默认情况下，这会同步调用依赖项。

为了避免这种行为，您必须使用您自己的CompletionStage或CompletableFuture的子类，这使得非*Async方法始终调用*Async方法。自Java9以来，这要容易得多，因为它覆盖了newInComplete eFuture（）。

它看起来像：

1） 此代码的作者不愿意提取此逻辑，而是更愿意重用ScheduledThreadPoolExecutor，

当ForkJoinPool出现在Java7中时，它缺少一个公共线程池。Java8引入了静态公共池（），在引入的CompletableFuture和Stream类中默认使用（以及其他）。

似乎他们不愿意透露一个共同的预定遗嘱执行人。这将与公共线程池一样有用，以避免部署许多很少使用的调度执行器。

如果您需要具有静态间隔的延迟任务，那么就需要CompletableFuture。delayedExecutor（）可能已经足够好了，因为包装对象的开销很小。

对于可变间隔，每次创建包装执行器都会有额外的开销，但在创建过程中已经创建了一些对象，例如内部取消器、超时、延迟完成器和任务提交器类的新实例。

我们多久需要在可变时间间隔内延迟许多任务？纯异步代码可能会在不同的超时情况下一直这样做，但由于我们没有暴露调度的执行程序本身，因此我们要么假设这个开销，要么使用另一个静态调度程序。

2）这显然导致了静态变量的这种解决方案，因为为每个CompletableFuture创建一个新的执行程序效率非常低。

确切地

Answer 2

你是对的，这可能会成为瓶颈，但不是完成本身，它只是在CompletableFuture中设置一个变量。那个单线程可以在一秒钟内完成数百万个未来。关键的方面是完成可能会触发完成线程中依赖阶段的评估。

所以

Executor neverDone = r -> {};
long t0 = System.nanoTime();
CompletableFuture<String> c11 =
    CompletableFuture.supplyAsync(() -> "foo", neverDone)
        .completeOnTimeout("timeout", 2, TimeUnit.SECONDS)
        .thenApply(s -> {
            System.out.println("long dependent action 1 "+Thread.currentThread());
            LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(5));
            return s;
        });
CompletableFuture<String> c12 =
    CompletableFuture.supplyAsync(() -> "bar", neverDone)
        .completeOnTimeout("timeout", 2, TimeUnit.SECONDS)
        .thenApply(s -> {
            System.out.println("long dependent action 2 "+Thread.currentThread());
            LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(5));
            return s;
        });
System.out.println("set up");
CompletableFuture.allOf(
    c11.thenAccept(System.out::println),
    c12.thenAccept(System.out::println)
).join();
System.out.println(Math.round((System.nanoTime()-t0)*1e-9)+" s");

将打印

set up
long dependent action 1 Thread[CompletableFutureDelayScheduler,5,main]
timeout
long dependent action 2 Thread[CompletableFutureDelayScheduler,5,main]
timeout
12 s

使用...异步链接方法将消除该问题

Executor neverDone = r -> {};
long t0 = System.nanoTime();
CompletableFuture<String> c11 =
    CompletableFuture.supplyAsync(() -> "foo", neverDone)
        .completeOnTimeout("timeout", 2, TimeUnit.SECONDS)
        .thenApplyAsync(s -> {
            System.out.println("long dependent action 1 "+Thread.currentThread());
            LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(5));
            return s;
        });
CompletableFuture<String> c12 =
    CompletableFuture.supplyAsync(() -> "bar", neverDone)
        .completeOnTimeout("timeout", 2, TimeUnit.SECONDS)
        .thenApplyAsync(s -> {
            System.out.println("long dependent action 2 "+Thread.currentThread());
            LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(5));
            return s;
        });
System.out.println("set up");
CompletableFuture.allOf(
    c11.thenAccept(System.out::println),
    c12.thenAccept(System.out::println)
).join();
System.out.println(Math.round((System.nanoTime()-t0)*1e-9)+" s");

将打印

set up
long dependent action 2 Thread[ForkJoinPool.commonPool-worker-2,5,main]
long dependent action 1 Thread[ForkJoinPool.commonPool-worker-9,5,main]
timeout
timeout
7 s

结论是，当您有一个可能很长的评估时，您应该始终通过其中一个异步方法来链接is。考虑到在使用不带“…Async”后缀的方法时无法控制执行线程（也可能是调用链接方法的线程或任何其他调用“完成方法”的线程，请参阅此答案），这是您始终应该做的。