优化许多文件的并行处理

Question

我有一个程序处理大量文件，其中每个文件都需要做两件事：首先，读取并处理一部分文件，然后MyFileData存储结果。第一部分可以并行化，第二部分不能并行化。

顺序执行所有操作非常慢，因为CPU必须等待磁盘，然后工作一点，然后发出另一个请求，然后再次等待…

我做了以下

class MyCallable implements Callable<MyFileData> {
    MyCallable(File file) {
        this.file = file;
    }
    public MyFileData call() {
        return someSlowOperation(file);
    }
    private final File file;
}

for (File f : files) futures.add(executorService.submit(new MyCallable(f)));
for (Future<MyFileData> f : futures) sequentialOperation(f.get());

这很有帮助。但是，我想改善两点：

• 在sequentialOperation获取一个固定的顺序，而不是处理任何结果，请首先执行。我该如何更改？

• 有成千上万的文件要处理，启动成千上万的磁盘请求可能会导致磁盘损坏。通过使用Executors.newFixedThreadPool(10)我限制了这个数字，但是我正在寻找更好的东西。理想情况下，它应该是自调整的，以使其在不同的计算机上最佳工作（例如，在RAID和/或NCQ可用时发出更多请求，等等）。我认为这可能不是基于发现硬件配置，但是应该可以 以某种方式 测量处理速度并基于它进行优化。任何想法？

Answer 1

serialOperation将以固定的顺序执行，而不是先处理任何可用的结果。我该如何更改？

这正是CompletionService的工作：它并行处理任务，并在完成任务时将其返回，而与提交顺序无关。

简化（未经测试）的示例：

int NUM_THREADS = Runtime.getRuntime().availableProcessors();
ExecutorService executor = Executors.newFixedThreadPool(NUM_THREADS);
CompletionService<MyFileData> completionService = new ExecutorCompletionService<MyFileData>(executor);

for (File f : files) futures.add(completionService.submit(new MyCallable(f)));

for(int i = 0; i < futures.size(); i++) {
    Future<MyFileData> next = completionService.take();
    sequentialOperation(next.get());
}

有成千上万的文件要处理，启动成千上万的磁盘请求可能会导致磁盘损坏。通过使用Executors.newFixedThreadPool（10），我限制了这个数字，但是我正在寻找更好的东西。

我不确定那一个。我想这取决于您有多少个磁盘，但是我会认为磁盘访问部分不应拆分成太多线程（每个磁盘一个线程可能是明智的）：如果有多个线程同时访问一个磁盘，它将花费比阅读更多的时间。