优化许多文件的并行处理

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

sequentialOperation以固定顺序执行,而不是先处理可用的结果.我该怎么改变它？

这正是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)我限制了这个数字,但是我正在寻找更好的东西.

我不是百分百肯定那个.我想这取决于你有多少磁盘,但我认为磁盘访问部分不应该分成太多的线程(每个磁盘一个线程可能是明智的):如果许多线程同时访问一个磁盘,它会花更多的时间寻求而不是阅读.