java - 以较小的内存占用量执行数百万次 Runnable

Question

我有 N 个 long，它们是 ID。对于每个 ID，我都需要执行一个 Runnable(即，我不关心返回值)并等待所有这些都完成。每个 Runnable 可能需要几秒到几分钟，并行运行大约 100 个线程是安全的。

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

代码运行良好并且非常简单，因为我不必处理线程、复杂的等待逻辑等。它有一个缺点:内存占用。

所有仍在排队的 Runnable 消耗内存(比 long 需要的多 8 个字节:这些是我的具有某些内部状态的 Java 类)，并且所有 N 个 Future 实例也消耗内存(这些是具有状态也是如此，我只用它来等待，但我不需要实际结果)。我查看了堆转储，我估计 N=1000 万占用了略多于 1 GiB 的内存。一个数组中的 1000 万个 long 只会消耗 76 MiB。

有没有办法只将 ID 保存在内存中来解决这个问题，最好不要求助于低级并发编程？

Answer 1

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

但我认为我会根据 Java 的 Stream 行为提出一些不同的建议。

直觉是，流的惰性执行将用于限制工作单元、 future 及其结果的创建。

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，将其提供给管道(映射到工作单元，分派(dispatch)到线程池，等待结果，过滤异常)，这意味着很快达到平衡平衡 Activity 工作单元的数量与 executor 的数量。

如果要微调并行“ block ”的数量，应该在此处跟进:Custom thread pool in Java 8 parallel stream