多线程海量文件读取

Question

我仍在全神贯注地了解Java中并发的工作方式。我知道（如果您订购的是OO Java 5并发模型）则分别实现a Task或Callablewith
run()或or call()方法，并且它应该使您尽可能多地并行实现该方法。

但是我仍然不了解Java并发编程的内在知识：

• 一个是怎样Task的run()方法分配给执行工作的同时适量的？

作为一个具体的例子，如果我有一个I / O绑定的readMobyDick()方法，该方法将从本地系统上的文件中读取Herman Melville的
Moby Dick 的全部内容到内存中。且仅说我希望此readMobyDick()方法是并发的，并由3个线程处理，其中：

• 线程＃1将书的前1/3读入内存
• 线程＃2将书的第二个1/3读入内存
• 线程＃3将书的最后1/3读入内存

我是否需要将Moby
Dick分成三个文件，并将每个文件传递给他们自己的任务，还是只readMobyDick()从实现的run()方法内部进行调用II，并且（以某种方式）Executor知道如何在线程之间分解工作。

我是一个非常视觉化的学习者，因此非常感谢使用正确方法实现此目的的任何代码示例！谢谢！

Answer 1

您可能偶然地选择了并行活动的绝对最差示例！

实际上，从单个机械磁盘并行读取要比使用单个线程读取速度慢，因为实际上您是在每个线程轮流运行时将机械头弹跳到磁盘的不同部分。最好将其保留为单线程活动。

让我们再举一个例子，它与您的例子相似，但实际上可以提供一些好处：假设我想在庞大的单词列表中搜索某个单词的出现（该列表甚至可能来自磁盘文件，但是像我一样表示，由单个线程读取）。假设我可以像您的示例中那样使用3个线程，每个线程都在巨大单词列表的1/3上进行搜索，并保留一个本地计数器来显示搜索到的单词出现的次数。

在这种情况下，您希望将列表分为三部分，将每个部分传递给一个不同的对象，该对象的类型实现Runnable并在该run方法中实现搜索。

运行时本身不知道如何进行分区或类似的操作，您必须自行指定。还有许多其他的分区策略，每种策略都有自己的优点和缺点，但是我们现在可以坚持使用静态分区。

让我们看一些代码：

class SearchTask implements Runnable {
     private int localCounter = 0;
     private int start; // start index of search
     private int end;
     private List<String> words;
     private String token;

     public SearchTask(int start, int end, List<String> words, String token) {
         this.start = start;
         this.end = end;
         this.words = words;
         this.token = token;
     }

     public void run() {
         for(int i = start; i < end; i++) {
              if(words.get(i).equals(token)) localCounter++;
         }
     }

     public int getCounter() { return localCounter; }
}

// meanwhile in main :)

List<String> words = new ArrayList<String>();
// populate words 
// let's assume you have 30000 words

// create tasks
SearchTask task1 = new SearchTask(0, 10000, words, "John");
SearchTask task2 = new SearchTask(10000, 20000, words, "John");
SearchTask task3 = new SearchTask(20000, 30000, words, "John");

// create threads for each task
Thread t1 = new Thread(task1);
Thread t2 = new Thread(task2);
Thread t3 = new Thread(task3);

// start threads
t1.start();
t2.start();
t3.start();

// wait for threads to finish
t1.join();
t2.join();
t3.join();

// collect results
int counter = 0;
counter += task1.getCounter();
counter += task2.getCounter();
counter += task3.getCounter();

这应该很好地工作。请注意，在实际情况下，您将构建更通用的分区方案。您也可以使用ExecutorServiceand实现，Callable而不是Runnable如果要返回结果。

因此，一个使用更高级构造的替代示例：

class SearchTask implements Callable<Integer> {
     private int localCounter = 0;
     private int start; // start index of search
     private int end;
     private List<String> words;
     private String token;

     public SearchTask(int start, int end, List<String> words, String token) {
         this.start = start;
         this.end = end;
         this.words = words;
         this.token = token;
     }

     public Integer call() {
         for(int i = start; i < end; i++) {
              if(words.get(i).equals(token)) localCounter++;
         }
         return localCounter;
     }        
}

// meanwhile in main :)

List<String> words = new ArrayList<String>();
// populate words 
// let's assume you have 30000 words

// create tasks
List<Callable> tasks = new ArrayList<Callable>();
tasks.add(new SearchTask(0, 10000, words, "John"));
tasks.add(new SearchTask(10000, 20000, words, "John"));
tasks.add(new SearchTask(20000, 30000, words, "John"));

// create thread pool and start tasks
ExecutorService exec = Executors.newFixedThreadPool(3);
List<Future> results = exec.invokeAll(tasks);

// wait for tasks to finish and collect results
int counter = 0;
for(Future f: results) {
    counter += f.get();
}