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

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

我熟悉Python中的多处理程序包,该程序包非常易于使用,因此理想情况下,我正在寻找类似的东西：

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

我认为代码的工作方式是：

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

好,所以有一个错误-可以创建6个以上的任务.即使所有池线程都忙,也将提交一些.我已计划稍后解决.

问题在于它根本不起作用. MyCoordinator线程在第一次删除时阻塞-很好.但是,即使将新任务放在队列中,它也不会畅通无阻.谁能告诉我我在做什么错？

解决方法:

您正在寻找的线程池是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
           }
      });

关于并行处理多个文件的一个重要建议：如果您有一个包含文件的机械磁盘,则明智的做法是使用单个线程一次读取它们,并将每个文件提交给如上所述的线程池任务,以便处理.不要并行读取实际读数,因为这会降低性能.

标签：multithreading,java-util-concurrent,java
来源： https://codeday.me/bug/20191201/2077456.html