在 Java 中使用 ExecutorService 的多线程帮助 [重复]

Question

我正在尝试搜索单词列表并查找多个文件中所有单词的总数。

我的逻辑是为每个文件设置单独的线程并获取计数。最后，我可以汇总从每个线程获得的总数。

假设，我有 50 个文件，每个文件 1MB。当我使用多个线程时，性能并没有提高。 FILE_THREAD_COUNT 并没有改善我的总执行时间。当我的线程数为 1 或 50 时，我得到几乎相同的执行时间。

我在使用执行器服务时做错了吗？

这是我的代码。

public void searchText(List<File> filesInPath, Set<String> searchWords) {
    try {
        BlockingQueue<File> filesBlockingQueue = new ArrayBlockingQueue<>(filesInPath.size());
        filesBlockingQueue.addAll(filesInPath);

        ExecutorService executorService = Executors.newFixedThreadPool(FILE_THREAD_COUNT);
        int totalWordCount = 0;
        while (!filesBlockingQueue.isEmpty()) {
            Callable<Integer> task = () -> {
                int wordCount = 0;
                try {
                    File file = filesBlockingQueue.take();
                    try (BufferedReader bufferedReader = new BufferedReader(new FileReader(file))) {
                        String currentLine;
                        while ((currentLine = bufferedReader.readLine()) != null) {
                            String[] words = currentLine.split("\\s+");
                            for (String word : words) {
                                for (String searchWord : searchWords) {
                                    if (word.contains(searchWord)) {
                                        wordCount++;
                                    }
                                }
                            }
                        }
                    } catch (Exception e) {
                        // Handle error
                    }
                } catch (Exception e) {
                    // Handle error
                }
                return wordCount;
            };
            totalWordCount += executorService.submit(task).get();
        }
        System.out.println("Final word count=" + totalWordCount);
        executorService.shutdown();
    } catch (Exception e) {
        // Handle error
    }
}

Answer 1

是的，你做错了什么。

问题出在这里：

executorService.submit(task).get()

您的代码提交一个任务，然后等待它完成，这并没有实现任何并行；任务按顺序运行。而你的 BlockingQueue 没有任何价值。

并行运行任务的方式是先提交所有个任务，收集返回的Futures，然后调用@987654324 @他们所有人。像这样：

List<Future<Integer>> futures = filesInPath.stream()
    .map(<create your Callable>)
    .map(executorService::submit)
    .collect(toList());

for (Future future : futures)
    totalWordCount += future.get();
}

您实际上可以通过中间列表（如上）在一个流中执行此操作，但随后立即对其进行流式传输，但是您必须在某些代码中包装对 Future#get 的调用以捕获检查的异常 - 我离开作为读者的练习。