使用并发类并行处理目录中的文件

Question

我试图弄清楚如何使用 java.util.concurrent 包中的类型来并行处理目录中的所有文件。

我熟悉 Python 中的 multiprocessing 包，使用起来非常简单，所以理想情况下我正在寻找类似的东西：

public interface FictionalFunctor<T>{
  void handle(T arg);
}

public class FictionalThreadPool {
  public FictionalThreadPool(int threadCount){
    ...
  }
  public <T> FictionalThreadPoolMapResult<T> map(FictionalFunctor<T> functor, List<T> args){
    // Executes the given functor on each and every arg from args in parallel. Returns, when
    // all the parallel branches return.
    // FictionalThreadPoolMapResult allows to abort the whole mapping process, at the least.
  }
}

dir = getDirectoryToProcess();
pool = new FictionalThreadPool(10);   // 10 threads in the pool
pool.map(new FictionalFunctor<File>(){ 
  @Override
  public void handle(File file){
    // process the file
  }
}, dir.listFiles());

我感觉java.util.concurrent 中的类型允许我做类似的事情，但我完全不知道从哪里开始。

有什么想法吗？

谢谢。

编辑 1

按照答案中给出的建议，我写了这样的东西：

public void processAllFiles() throws IOException {
  ExecutorService exec = Executors.newFixedThreadPool(6);
  BlockingQueue<Runnable> tasks = new LinkedBlockingQueue<Runnable>(5); // Figured we can keep the contents of 6 files simultaneously.
  exec.submit(new MyCoordinator(exec, tasks));
  for (File file : dir.listFiles(getMyFilter()) {
    try {
      tasks.add(new MyTask(file));
    } catch (IOException exc) {
      System.err.println(String.format("Failed to read %s - %s", file.getName(), exc.getMessage()));
    }
  }
}

public class MyTask implements Runnable {
  private final byte[] m_buffer;
  private final String m_name;

  public MyTask(File file) throws IOException {
    m_name = file.getName();
    m_buffer = Files.toByteArray(file);
  }

  @Override
  public void run() {
    // Process the file contents
  }
}

private class MyCoordinator implements Runnable {
  private final ExecutorService m_exec;
  private final BlockingQueue<Runnable> m_tasks;

  public MyCoordinator(ExecutorService exec, BlockingQueue<Runnable> tasks) {
    m_exec = exec;
    m_tasks = tasks;
  }

  @Override
  public void run() {
    while (true) {
      Runnable task = m_tasks.remove();
      m_exec.submit(task);
    }
  }
}

我认为代码的工作原理是：

1. 文件一个接一个地被读取。
2. 文件内容保存在专用的MyTask 实例中。
3. 一个容量为 5 的阻塞队列来容纳任务。我依靠服务器一次最多保留 6 个文件的内容的能力 - 5 个在队列中，另一个完全初始化的任务等待进入队列。
4. 一个特殊的MyCoordinator 任务从队列中获取文件任务并将它们分派到同一个池中。

好的，所以有一个错误 - 可以创建超过 6 个任务。即使所有池线程都忙，也会提交一些。我打算稍后解决它。

问题是它根本不起作用。 MyCoordinator 线程在第一次删除时阻塞 - 这很好。但它永远不会解除阻塞，即使新任务已放入队列中。谁能告诉我我做错了什么？

Answer 1

您要查找的线程池是ExecutorService 类。您可以使用newFixedThreadPool 创建一个固定大小的线程池。这使您可以轻松实现生产者-消费者模式，池为您封装了所有队列和工作器功能：

ExecutorService exec = Executors.newFixedThreadPool(10);

然后您可以以类型实现Runnable（或Callable，如果您还想获得结果）的对象形式提交任务：

class ThreadTask implements Runnable {
    public void run() {
       // task code
    }
}

...

exec.submit(new ThreadTask());
// alternatively, using an anonymous type
exec.submit(new Runnable() {
           public void run() {
              // task code
           }
      });

关于并行处理多个文件的重要建议：如果您有一个机械磁盘保存文件，明智的做法是使用单个线程逐个读取它们并提交每个文件文件到上面的线程池任务中，进行处理。不要并行执行实际读取，因为这会降低性能。

Answer 2

比使用 ExecuterService 更简单的解决方案是实现您自己的生产者-消费者方案。有一个创建任务并提交到 LinkedBlockingQueue 或 ArrayBlockingQueue 的线程，并有检查此队列以检索任务并执行它们的工作线程。您可能需要一种特殊的任务名称 ExitTask 来强制工作人员退出。因此，在作业结束时，如果您有 n 个工作人员，您需要将 n 个 ExitTasks 添加到队列中。

Answer 3

基本上，@Tudor 所说的，使用 ExecutorService，但我想扩展他的代码，我总是觉得编辑别人的帖子很奇怪。这是您将提交给 ExecutorService 的概要：

public class MyFileTask implements Runnable {
   final File fileToProcess;

   public MyFileTask(File file) {
      fileToProcess = file;
   }

   public void run() {
      // your code goes here, e.g.
      handle(fileToProcess);
      // if you prefer, implement Callable instead
   }
}

另见my blog post here for some more details if you get stuck

由于处理文件经常导致 IOExceptions，我更喜欢 Callable（它可以抛出检查的异常）而不是 Runnable，但是 YMMV。