线程池中的多个线程在同一列表中写入数据

Question

我有多个线程在我的threadPool中运行每个线程读取一个大文件并在List中返回该文件中的数据。

代码如下所示：

class Writer{


   ArrayList finalListWhereDataWillBeWritten = new Array<Integer>()
   for(query q : allQueries){ //all the read queries to read file

      threadPool.submit(new GetDataFromFile(fileName,filePath));        

   }//all the read queries have been submitted.

}

现在我知道以下代码段将出现在我的代码中的某个位置，但我不知道将其放置在哪里。因为如果我在for循环中的提交（）之后放置它，它就不会添加它，因为每个文件都非常大，可能还没有完成它的处理。

synchronized(finalListWhereDataWillBeWritten){

  //process the data obtained from single file and add it to target list 
      finalListWhereDataWillBeWritten.addAll(dataFromSingleThread); 

}

那么，有谁能告诉我，我应该把这段代码放在哪里，以及我需要确保哪些其他事情，以避免出现关键部分问题。

class GetDataFromFile implements Runnable<List<Integer>>{

   private String fileName;
   private String filePath;

   public List<Integer> run(){
       //code for streaming the file fileName 
       return dataObtainedFromThisFile;
   }

}

如果我只是在线程中并行读取文件中的数据，并将它们放在共享列表中，那么我需要在代码中使用wait（）/notifyAll（）方法吗

Answer 1

更新请考虑Marko提供的答案，这要好得多

如果要确保在处理列表之前完成所有线程，请执行以下操作：

import java.util.List;
import java.util.Vector;

public class ThreadWork {

  public static void main(String[] args) {

    int count = 5;
    Thread[] threads = new ListThread[count];
    List<String> masterList = new Vector<String>();

    for(int index = 0; index < count; index++) {
      threads[index] = new ListThread(masterList, "Thread " + (index + 1));
      threads[index].start();
    }
    while(isOperationRunning(threads)) {
      // do nothing
    }

    System.out.println("Done!! Print Your List ...");

    for(String item : masterList){
      System.out.println("[" + item + "]");
    }
  }

  private static boolean isOperationRunning(Thread[] threads) {
    boolean running = false;

    for(Thread thread : threads) {
      if(thread.isAlive()) {
        running = true;
        break;
      }
    }
    return running;
  }
}

class ListThread extends Thread {
  private static String items[] = { "A", "B", "C", "D"};
  private List<String> list;
  private String name;

  public ListThread(List<String> masterList, String threadName) {
    list = masterList;
    name = threadName;
  }

  public void run() {
    for(int i = 0; i < items.length;++i) {
      randomWait();
      String data = "Thread [" + name + "][" + items[i] + "]";
      System.out.println( data );
      list.add( data );
    }
  }

  private void randomWait() {
    try {
      Thread.currentThread();
      Thread.sleep((long)(3000 * Math.random()));
    }
    catch (InterruptedException x) {}
  }
}

Answer 2

与其重新发明轮子，不如简单地实现Callable

final ExecutorService threadPool = 
    Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
final List<Future<List<Integer>>> futures = new ArrayList<>();
for(query q : allQueries) {
  futures.add(threadPool.submit(new GetDataFromFile(fileName, filePath)));
}
for (Future<List<Integer>> f : futures) {
    finalListWhereDataWillBeWritten.addAll(f.get());
}

这都是假设你低于Java8。对于Java 8，您当然会使用并行流：

final List<Integer> finalListWhereDataWillBeWritten =
  allQueries.parallelStream()
            .flatMap(q -> getDataFromFile(q.fileName, q.filePath))
            .collect(toList());