以很小的内存占用执行数百万个可运行的

Question

我有N个愿望是ID。对于每一个ID，我都需要执行一个Runnable（即，我不关心返回值），并等待它们全部完成。每个Runnable的运行时间从几秒到几分钟不等，并行运行大约100个线程是安全的。

在我们当前的解决方案中，我们使用Executors.NewFixedThreadPool()，对每个ID调用submit()，然后对每个返回的Future调用get（）。

代码工作得很好，而且非常简单，因为我不必处理线程、复杂的等待逻辑等。它有一个缺点：内存占用。

所有仍然排队的Runnable都消耗内存（比long所需的多8个字节：这些是我的Java类，有一些内部状态），所有的N个未来实例也消耗内存（这些也是有状态的Java类，我只用于等待，但不需要实际结果）。我查看了一个堆转储，估计N=1000万占用的内存略高于1GiB。一个数组中的1000万个长操作只会消耗76个MIB。

有没有一种方法只在内存中保存ID，最好不诉诸低级并发编程来解决这个问题？

Answer 1

这是我通常使用生产者/消费者模式和协调两者的BlockingQueue所做的事情，或者使用Akka Actor，如果我手头有项目的话。

但是我想我应该建议一些安静的，有点不同的东西，依赖于Java的流行为。

直觉是，流的惰性执行将被用来抑制工作单元、未来及其结果的创建。

public static void main(String[] args) {
    // So we have a list of ids, I stream it
    // (note : if we have an iterator, you could group it by a batch of, say 100,
    // and then flat map each batch)
    LongStream ids = LongStream.range(0, 10_000_000L);
    // This is were the actual tasks will be dispatched
    ExecutorService executor = Executors.newFixedThreadPool(4);

    // For each id to compute, create a runnable, which I call "WorkUnit"
    Optional<Exception> error = ids.mapToObj(WorkUnit::new)
             // create a parralel stream
             // this allows the stream engine to launch the next instructions concurrently
            .parallel()
            // We dispatch ("parallely") the work units to a thread and have them execute
            .map(workUnit -> CompletableFuture.runAsync(workUnit, executor))
            // And then we wait for the unit of work to complete
            .map(future -> {
                try {
                    future.get();
                } catch (Exception e) {
                    // we do care about exceptions
                    return e;
                } finally {
                    System.out.println("Done with a work unit ");
                }
                // we do not care for the result
                return null;
            })
            // Keep exceptions on the stream
            .filter(Objects::nonNull)
            // Stop as soon as one is found
            .findFirst();


    executor.shutdown();
    System.out.println(error.isPresent());
}

老实说，我不是很确定这种行为是由规范保证的，但从我的经验来看，它是有效的。每个并行“chunck”抓取几个ID，将其馈送到管道（映射到工作单元、分派到线程池、等待结果、过滤异常），这意味着活动工作单元的数量与执行器的数量相当快地达到了平衡。

如果要对并行“chunck”的数量进行微调，那么应该在下面进行跟踪：Java 8并行流中的自定义线程池（Custom thread pool in Java 8 parallel stream