java - 多线程读取大量文件

Question

我仍在思考 Java 中并发的工作原理。我知道(如果您订阅了 OO Java 5 并发模型)您使用 run() 或 Task 或 Callable 或call() 方法(分别)，您应该尽可能多地并行化已实现的方法。

但我仍然不理解 Java 中并发编程的内在特性:

• Task 的 run() 方法如何分配要执行的正确数量的并发工作？

举个具体的例子，如果我有一个 I/O 绑定(bind)的 readMobyDick() 方法，它将 Herman Melville 的 Moby Dick 的全部内容从本地系统上的文件。假设我希望此 readMobyDick() 方法并发并由 3 个线程处理，其中:

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

我是否需要将 Moby Dick 分为三个文件并将它们分别传递给各自的任务，或者我是否只从已实现的 run( ) 方法和(不知何故)Executor 知道如何在线程之间分解工作。

我是一个视觉型学习者，所以非常感谢任何正确方法的代码示例!谢谢!

Answer 1

您可能不小心选择了并行 Activity 中绝对最糟糕的例子!

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

让我们再举一个例子，它与你的相似，但实际上可以提供一些好处:假设我想在一个巨大的单词列表中搜索某个单词的出现(这个列表甚至可以来自磁盘文件，但就像我说的，通过一个线程阅读)。假设我可以像您的示例一样使用 3 个线程，每个线程搜索庞大单词列表的 1/3，并保留一个本地计数器，记录搜索到的单词出现的次数。

在这种情况下，您希望将列表分成 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();

这应该能很好地工作。请注意，在实际情况下，您会构建更通用的分区方案。如果您希望返回结果，您也可以使用 ExecutorService 并实现 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();
}