以较小的内存占用执行数百万个 Runnable

Question

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

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

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

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

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

Answer 1

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

但我想我会建议一些更安静的东西，依赖于 Java 的 Stream 行为。

直觉是，流的延迟执行将用于限制工作单元、期货及其结果的创建。

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，将其馈送到管道（映射到工作单元，分派到线程池，等待结果，过滤异常），这意味着很快就达到了平衡平衡活动工作单元的数量与executor 的数量。

如果要微调并行“块”的数量，应该在这里跟进：Custom thread pool in Java 8 parallel stream

Answer 2

是的：你可以有一个共享的 long 队列。你向执行器提交 n 个Runnables，其中 n 是执行器中的线程数，在 run 方法结束时，你从队列中获取下一个 long，并重新提交一个新的 Runnable。

Answer 3

与其创建数百万个 Runnable，不如创建特定的线程池，将 long 作为任务。 不要使用 Future.get() 等待任务完成，而是使用 CountdownLatch。

那个线程池可以这样实现：

int N = 1000000;// number of tasks;
int T = 100; // number of threads;
CountdownLatch latch = new CountdownLatch(N);
ArrayBlockingQueue<Long> queue = new ArrayBlockingQueue<>();

for (int k=0; k<N; k++) {
   queue.put(createNumber(k));
}
for (int k=0; k<T; k++) {
  new WorkingThread().start();
}
CountdownLatch.await();

class WorkingThread extends Thread {
  public void run() {
      while (latch.getCount() != 0) {
           processNumber(queue.take());
           latch.countDown();
      }
  }
}

Answer 4

使用ExecutorCompletionService 怎么样？类似于以下内容（可能包含错误，我没有测试过）：

import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorCompletionService;
import java.util.function.LongFunction;

public class Foo {

  private final ExecutorCompletionService<Void> completionService;
  private final LongFunction<Runnable> taskCreator;
  private final long maxRunning; // max tasks running or queued

  public Foo(Executor executor, LongFunction<Runnable> taskCreator, long maxRunning) {
    this.completionService = new ExecutorCompletionService<>(executor);
    this.taskCreator = taskCreator;
    this.maxRunning = maxRunning;
  }

  public synchronized void processIds(long[] ids) throws InterruptedException {
    int completed = 0;

    int running = 0;
    for (long id : ids) {
      if (running < maxRunning) {
        completionService.submit(taskCreator.apply(id), null);
        running++;
      } else {
        completionService.take();
        running--;
        completed++;
      }
    }

    while (completed < ids.length) {
      completionService.take();
      completed++;
    }

  }

}

---

上述的另一个版本可以使用Semaphore 和CountDownLatch，而不是CompletionService。

public static void processIds(long[] ids, Executor executor,
                              int max, LongFunction<Runnable> taskSup) throws InterruptedException {
  CountDownLatch latch = new CountDownLatch(ids.length);
  Semaphore semaphore = new Semaphore(max);

  for (long id : ids) {
    semaphore.acquire();

    Runnable task = taskSup.apply(id);
    executor.execute(() -> {
      try {
        task.run();
      } finally {
        semaphore.release();
        latch.countDown();
      }
    });

  }

  latch.await();
}