优化多个文件的并行处理

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());

这很有帮助。然而，我想改进两件事：

>

顺序操作以固定顺序执行，而不是首先处理任何可用的结果。如何更改它？

有数千个文件需要处理，启动数千个磁盘请求可能会导致磁盘垃圾。通过使用执行器。newFixedThreadPool（10）我限制了这个数字，但我正在寻找更好的。理想情况下，它应该是自调整的，以便在不同的计算机上最佳工作（例如，当RAID和/或NCQ可用时，发出更多请求，等等）。我不认为这可以基于找出硬件配置，但测量处理速度并基于它进行优化应该是可能的。知道吗？

Answer 1

顺序操作按固定顺序执行，而不是先处理任何可用的结果。我怎样才能改变它？

假设：每个<代码>一些操作（文件） 调用将花费可变的时间，因此，您希望在收到MyFileData后立即处理它，但不要与另一个顺序操作同时处理。

您可以通过设置生产者/消费者队列来实现这一点。

生产者是您在示例中执行的可调用对象，带有添加的位，您将结果添加到等待处理的工作队列中。

Consumer是sequentialOperation（）调用-它在自己的线程中运行，并且只有一个线程。这个线程所做的就是占据队列的头，并对其进行处理，重复执行，直到程序结束。

这样，您可以最大限度地利用机器上的所有资源。

带有一些示例代码的相关帖子：使用队列的生产者/消费者线程

编辑：我想你可能需要一个快速的样本，因为它对任何以前从未做过的人来说都非常不透明

public class Main {

    private final ExecutorService producerExecutor = Executors.newFixedThreadPool(10);
    private final ExecutorService consumerExecutor = Executors.newFixedThreadPool(1);
    private final LinkedBlockingQueue<MyData> queue = new LinkedBlockingQueue();//or some other impl

    abstract class Producer implements Runnable{
        private final File file;
        Producer(File file) {
            this.file = file;
        }

        public void run() {
            MyData result = someLongAssOperation(file);
            queue.offer(result);
        }

        public abstract void someLongAssOperation(File file);
    }

    abstract class Consumer implements Runnable {
        public void run() {
            while (true) {
                sequentialOperation(queue.take());  
            }
        }

        public abstract void sequentialOperation(MyData data);
    } 

    private void start() {
        consumerExecutor.submit(new Consumer(){
            //implement sequentialOperation here
        });

        for (File f : files) {
            producerExecutor.submit(new Producer(file) {
                //implement the someLongAssOperation()
            });
        }

    }

    public static void main(String[] args) {
        new Main().start();     
    } 

}

Answer 2

顺序操作按固定顺序执行，而不是先处理任何可用的结果。我怎样才能改变它？

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

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