java - 使用 CompletionService 处理大量任务

Question

我需要在多核机器上处理大量(>1亿)请求(每个请求是处理一个数据文件中的一行，涉及到一些与远程系统的I/O。虽然细节做没关系，具体任务是从一些数据文件中加载一个分布式的 Hazelcast map )。执行将通过 ThreadPoolExecutor 处理.一个线程将读取文件，然后将数据提交给多个独立线程以将其放入 map 中。该机器有 32 个内核，因此有足够的空间用于 map 的并行加载。

由于请求量大，通常创建任务并排队到执行器服务的方法不可行，因为排队的任务会占用太多内存。

这带来了 ExecutorCompletionService .有了它，任务将在前一个操作完成时提交，这通过调用 take()(或 poll()，如适用)获知。当执行程序服务的所有线程都被使用时，这将正常工作。然而，“加载所有线程”还没有完成。分为两个阶段:

• 填满队列:当池中仍有未使用的线程时，将任务提交给 ExecutorCompletionService，不要等到提交更多线程

• 填充队列:一旦所有线程都被使用，只有在前一个任务完成后才提交任务。因此，将尽可能快地提供行，但不会更快，也不会排队。

上面可以编码，但是我想知道上面的逻辑是否已经实现了，我不知何故错过了。我问是因为这看起来很常见。

Answer 1

您可以在创建ThreadPoolExecutor 时指定BlockingQueue 实现。如果您要避免的只是创建过多的 Runnable 对象，那么您可以使用有界 BlockingQueue，例如ArrayBlockingQueue有一个线程将项目推送到队列，当队列达到容量时将被阻塞。