Java中高效的多线程文件处理

Question

在我的代码中，这个方法使用生产者-消费者设计来读取许多文件并处理它们，以便在NLP算法中使用。完成服务在处理文件时收集文件，这意味着docId != index。考虑到我需要在docID == index的其他地方检索概率，我已经实现了一个优先级队列来排序。但是，我这样做的方法是从期货列表中依次检索文档，然后将它们放到队列中。

我知道这个实现不是很好。我一直盯着concurrentskiplistset，认为为了更好的性能，我应该能够实现它，而不存在这些问题。对于并发和Java，我是全新的，所以我肯定有很多需要批评的地方。

我如何使这个多线程更有效？

矢量发生器：

public class Vectorizer {
public Vectorizer(){
}
public PriorityQueue<Document> readAll(File fileDir) 
throws InterruptedException, ExecutionException, IOException{
    //read each file. When each file is vectorized, put it in a minibatch.
    //producer-consumer threading structure.
    int NUM_THREADS = Runtime.getRuntime().availableProcessors();
    BlockingQueue<LinkedList<String>> queue = new ArrayBlockingQueue<>(50);
    ExecutorService service = Executors.newFixedThreadPool(NUM_THREADS);
    CompletionService<Document> completionService = 
               new ExecutorCompletionService<Document>(service);
    //TODO: is list the most efficient structure to handle future?
    List<Future<Document>> docs = new ArrayList<Future<Document>>();
    for (int i = 0; i < (NUM_THREADS - 1); i++) {
        docs.add(completionService.submit(new DocumentConsumer(queue)));
    }
    // Wait for ReadFile to complete
    service.submit(new ReadFile(queue, fileDir)).get();
    service.shutdownNow();  // interrupt CPUTasks
    // Wait for DocumentConsumer to complete
    service.awaitTermination(365, TimeUnit.DAYS);

    //do things with processed docs.
    //should I be doing this, though?
    PriorityQueue<Document> Documents = new PriorityQueue<Document>();
    for(Future<Document> d : docs){
        try{
        Document doc = d.get();
        Documents.add(doc);
        System.out.println(Integer.toString(doc.Cj));
        } catch(ExecutionException e) {
            e.getCause();e.printStackTrace();
        }
    }
    return Documents;
}

文档对象(没有比较和重载)：

public class Document implements Comparator<Document>{
int docId, Cj;
HashMap<String,Integer> termTable;
public Document(int id,int Cj, HashMap<String,Integer> termMap){
    this.docId = id;
    this.Cj = Cj;

    this.termTable = termMap;
}

docId是在文件读取期间设置的。我读取一个文件并将计数附加到第一行，然后在处理过程中对其进行设置。

其余的代码是可用的这里。

Answer 1

我没有看你描述的具体性能问题。不过，我的总评是：

可读性

for (int i = 0; i < (NUM_THREADS - 1); i++) {
    docs.add(completionService.submit(new DocumentConsumer(queue)));
}

从0开始到somevalue - 1结束。你为什么不从1点开始呢？你甚至不对指数做任何事，所以这并不重要。

        e.getCause();e.printStackTrace();

不要将多个语句放在同一行中。它会把你的代码变成一堵代码墙。你觉得理解一堵墙很难吗？试试密码墙..。

命名

方法和变量使用camelCase。对类型使用PascalCase。常量使用ALL_CAPS。这意味着..。

int NUM_THREADS = Runtime.getRuntime().availableProcessors();

应该是最终的

PriorityQueue<Document> Documents = new PriorityQueue<Document>();

应该重命名为documents。

public Document(int id,int Cj, HashMap<String,Integer> termMap){

应该将Cj重命名为cj。..。更好的办法是把它重命名为解释它是什么的东西。现在我不知道cj是什么。