Java中无限并行流映射减少管道的时间限制答案

【问题标题】：Infinite parallel stream map reduce pipeline with time limit in JavaJava中无限并行流映射减少管道的时间限制
【发布时间】：2019-11-27 11:20:17
【问题描述】：

我正在尝试在 Java 中实现这种并发策略。我真的不在乎我使用的是流还是执行器，这里重要的是效率。

我想在给定的时间期限内并行运行尽可能多的模拟。单个模拟的运行速度比超时快。只要底层线程池有容量，就应该提交新的模拟。

+----------------+         +--------------+           +------------+
|                |  param  |              | Solution  |            |
| SimCoordinator +-------->+  Simulation  +----+----->+ reduce(min)|
|                |         |              |    |      |            |
+-------+--------+         +--------------+    |      +------------+
        ^                                      |
        +--------------------------------------+

类看起来像：

class SimCoordinator {
    double getParam(); //changes every time
    double putSolution(Solution s);
}

class Simulation {
    Solution run(double param); // takes a while to compute
}

协调器向任务提供参数。

然后运行任务并获得解决方案。

解决方案必须反馈给协调器，这将改变参数的生成方式。

每个解决方案都有成本；我想以最低成本保存解决方案，并在运行模拟时执行此操作，而不是在截止日期之后。

Simulation.run() 是一个长时间的操作，因此它应该与其他模拟同时运行。假设我有一个 N 个线程的线程池（或者并行流正在工作）。最初，simcoordinator 为所有线程提供模拟，当模拟完成时，它会创建新的。

解决此类问题的最佳方法是什么？

【问题讨论】：

您正在描述一个因果循环，不清楚当每个模拟都需要基于前一个模拟的参数时，并行性应该发挥作用。
@Holger，抱歉不清楚。 Simulation.run() 是一个长时间的操作，因此它应该与其他模拟同时运行。假设我有一个 N 个线程的线程池（或者并行流正在工作）。最初 simcoordinator 为所有线程提供模拟，当模拟完成时，它会创建新的。我会更新我的问题
只需创建一个Thread Pool 和n 线程，其中n 是您拥有的cpu 核心数...
所以单个模拟的运行速度应该比超时快得多，所以当有容量时您可能会提交更多？
@Holger 完全正确。如果没有超时，模拟将永远提交。

标签： java parallel-processing java-stream java-threads

【解决方案1】：

所以基本上你所拥有的是

模拟 - 一个Function<Double, Double>
根据旧参数及其结果计算新参数，即BiFunction<Double, Double, Double>
记录最低参数/分数集的评估，即Consumer<Double, Double>。
一个计时器用完，阻止新的模拟开始。

总之你可以拥有类似的东西

class SimulationDriver {
    private ExecutorService service;
    AtomicBoolean finished = new AtomicBoolean();
    private ScheduledExecutorService timer = Executors.newScheduledThreadPool(1);

    private Function<Double, Double> simulation;
    private BiFunction<Double, Double, Double> calculateNewParameter;

    private Map<Double, Double> results = new ConcurrentHashMap<>();

    private SimulationDriver(Function<Double, Double> sim,
                             BiFunction<Double, Double, Double> newParam) {
        simulation = sim;
        calculateNewParameter = newParam;
    }

    public void run(int timeout, Runnable callback, Double... params) {
        if (service != null) throw new InvalidStateException("already running");

        // initial simulations, one in each thread
        service = Executors.newFixedThreadPool(params.length);
        for (Double d : params) {
            submitForSimulation(d);
        }
        // after timeout: set finished to prevent new simulations and notify caller
        timer.schedule(() -> { 
            finished.set(true);
            service.shutdown();
            callback.run(); // tell outside world we're finished
        }, timeout, ChronoUnit.SECONDS);
    }

    void submitForSimulation(Double d) {
        if (!finished.get()) { // don't start new simulations after time ran out
            CompletableFuture.supplyAsync(() -> simulation.apply(d), service);
            result.thenAccept(r -> results.put(d, r)); // store param/result in map
            // calculate next parameter and run simulation with that
            result.thenAccept(r -> calculateNewParameter.apply(d, r))
                  .thenAccept(this::submitForSimulation);
        }
    }
}

【讨论】：