时间戳

首页 > 解决方案 > 用于 Java CompletableFuture 组合的线程？

java - 用于 Java CompletableFuture 组合的线程？

问题描述

我开始熟悉 JavaCompletableFuture组合，使用过 JavaScript Promise。基本上，组合只是在指定的执行程序上安排了链式命令。但是我不确定执行组合时哪个线程正在运行。

假设我有两个执行者，executor1并且executor2；为简单起见，假设它们是单独的线程池。我安排了一个CompletableFuture（使用非常宽松的描述）：

CompletableFuture<Foo> futureFoo = CompletableFuture.supplyAsync(this::getFoo, executor1);

然后，当完成后，我使用第二个执行器将其转换Foo为：Bar

CompletableFuture<Bar> futureBar .thenApplyAsync(this::fooToBar, executor2);

我知道这getFoo()将从executor1线程池中的线程调用。我知道这fooToBar()将从executor2线程池中的线程调用。

但是什么线程用于实际组合，即在getFoo()完成之后futureFoo()完成；但在命令fooToBar()被安排之前executor2？换句话说，哪个线程实际运行代码以在第二个执行程序上调度第二个命令？

executor1调度是否作为调用的同一线程的一部分执行getFoo()？如果是这样，这个可完成的未来组合是否等同于我fooToBar()在executor1任务的第一个命令中手动安排自己？

标签： javaparallel-processingcompletable-futureexecutor

解决方案

这是故意未指定的。Async在实践中，当调用没有后缀的变体并表现出类似的行为时，它将由处理链接操作的相同代码处理。

所以当我们使用下面的测试代码时

CompletableFuture.supplyAsync(() -> {
    LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(1));
    return "";
}, r -> new Thread(r, "A").start())
.thenAcceptAsync(s -> {}, r -> {
    System.out.println("scheduled by " + Thread.currentThread());
    new Thread(r, "B").start();
});

它可能会打印

scheduled by Thread[A,5,main]

因为完成前一个阶段的线程被用来安排依赖的动作。

但是，当我们使用

CompletableFuture<String> first = CompletableFuture.supplyAsync(() -> "",
    r -> new Thread(r, "A").start());
LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(1));
first.thenAcceptAsync(s -> {}, r -> {
    System.out.println("scheduled by " + Thread.currentThread());
    new Thread(r, "B").start();
});

它可能会打印

scheduled by Thread[main,5,main]

当主线程调用thenAcceptAsync时，第一个未来已经完成，主线程将自行安排动作。

但这不是故事的结局。当我们使用

CompletableFuture<String> first = CompletableFuture.supplyAsync(() -> {
    LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(5));
    return "";
}, r -> new Thread(r, "A").start());

Set<String> s = ConcurrentHashMap.newKeySet();
Runnable submitter = () -> {
    String n = Thread.currentThread().getName();
    do {
        for(int i = 0; i < 1000; i++)
            first.thenAcceptAsync(x -> s.add(n+" "+Thread.currentThread().getName()),
                Runnable::run);
    } while(!first.isDone());
};
Thread b = new Thread(submitter, "B");
Thread c = new Thread(submitter, "C");
b.start();
c.start();
b.join();
c.join();
System.out.println(s);

它不仅可以打印组合B A和C A来自第一个场景和B B来自C C第二个场景。在我的机器上，它还可以重现地打印组合B C，并C B指示一个线程传递给的动作thenAcceptAsync被另一个线程同时调用thenAcceptAsync不同的动作提交给执行程序。

这与该答案中描述的评估传递给thenApply（没有）的函数的线程的场景相匹配。正如一开始所说，这是我所期望的，因为这两件事很可能由相同的代码处理。但与评估传递给的函数的线程不同，在文档中甚至没有提到调用方法的线程。所以理论上，另一个实现可以使用一个完全不同的线程，而不是调用未来的方法，也不会完成它。AsyncthenApplyexecuteExecutor

推荐阅读