递归实现 CompletableFuture

Question

我有三个任务：

createSimulationScenarios()、runSimulation() 和 getSimulationOutput()

这些任务以迭代方式完成：

1. createSimulationScenarios() 创建新的模拟文件
2. runSimulation() 运行模拟。如果有很多模拟，可以并行批量执行。
3. 模拟完成后，getSimulationOutput() 就会立即处理输出数据。只要特定的模拟已经完成，也可以并行进行。

然后，当所有模拟完成并获得所有输出时，createSimulationScenarios() 将根据前一批的输出创建一组新的模拟文件。依此类推...

因为 each runSimulation() 和 each getSimulationOutput() 可以并行完成，我想到使用CompletableFuture来实现。

for (i = 0; i < noSimulations; ++i) {
    CompletableFuture<Void> runSim = CompletableFuture.runAsync(new runSimulation(), simulationExecutor);
    CompletableFuture<Void> getOutput = runSim.thenCompose(x ->
                                        CompletableFuture.runAsync(new getSimulationOutput(), outputExecutor));
}

但是，我陷入了如何将此块连接到另一个用于 createSimulationScenarios() 的 CompletableFuture 的问题，并要求它仅在所有模拟和输出完成时才开始执行（其中是下一次迭代的开始）。

有什么想法吗？

Answer 1

看看 ForkJoinTask，它似乎非常适合计算量大的任务，可以在不同的环境中以最佳方式运行

RecursiveTask 参考显示了一个简单的示例执行

Answer 2

您可以坚持使用迭代，而不是使用递归，但似乎无限循环正是您正在寻找的。此示例使用无限流。 CompletableFuture 调用可能需要根据所需的确切行为进行更改（翻译：我怀疑它会编译，只是显示代码可能是什么样子）。

Stream.generate(() -> createSimulationScenarios())
    .map(simulations -> IntStream.range(0, simulations.size())
       .mapToObj(i -> CompletableFuture.runAsync(new runSimulation(), simulationExecutor)
          .thenCompose(x -> CompletableFuture.runAsync(new getSimulationOutput(), 
                            outputExecutor)))
       .collect(Collectors.toList())) // force tasks to be submitted
    .forEach(futureList -> futureList.stream().forEach(CompletableFuture::join));