Почему есть вариации в ответе при использовании нескольких потоков для подсчета частоты слов в большом файле?

Question

Моя цель - подсчитать частоту каждого слова при чтении большого файла с использованием нескольких потоков. Я реализую интерфейс Runnable для достижения многопоточности. Но при выполнении программы я не всегда получаю правильный ответ. Иногда он дает правильный результат, а иногда нет. Но при использовании интерфейса Callable вместо Runnable программа выполняется правильно без ошибок.

Это основной класс:

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class WordFrequencyRunnableTest {

    public static void main(String[] args) throws IOException {
        long startTime = System.currentTimeMillis();
        String filePath = "C:/Users/Mukesh Kumar/Desktop/data.txt";
        WordFrequencyRunnableTest runnableTest = new WordFrequencyRunnableTest();
        Map<String, Integer> wordFrequencies = runnableTest.parseLines(filePath);
        runnableTest.printResult(wordFrequencies);
        long elapsedTime = System.currentTimeMillis() - startTime;
        System.out.println("Total execution time in millis: " + elapsedTime);
    }

    public Map<String, Integer> parseLines(String filePath) throws IOException {
        Map<String, Integer> wordFrequencies = new HashMap<>();
        try (BufferedReader bufferedReader = new BufferedReader(new FileReader(filePath))) {
            String eachLine = bufferedReader.readLine();
            while (eachLine != null) {
                List<String> linesForEachThread = new ArrayList<>();
                while (linesForEachThread.size() != 100 && eachLine != null) {
                    linesForEachThread.add(eachLine);
                    eachLine = bufferedReader.readLine();
                }
                WordFrequencyUsingRunnable task = new WordFrequencyUsingRunnable(linesForEachThread, wordFrequencies);
                Thread thread = new Thread(task);
                thread.start();
            }
        }
        return wordFrequencies;
    }

    public void printResult(Map<String, Integer> wordFrequencies) {
        wordFrequencies.forEach((key, value) -> System.out.println(key + " " + value));
    }
}

А это logi c class:

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

public class WordFrequencyUsingRunnable implements Runnable {

    private final List<String> linesForEachThread;
    private final Map<String, Integer> wordFrequencies;

    public WordFrequencyUsingRunnable(List<String> linesForEachThread, Map<String, Integer> wordFrequencies) {
        this.linesForEachThread = linesForEachThread;
        this.wordFrequencies = wordFrequencies;
    }

    @Override
    public void run() {
        List<String> currentThreadLines = new ArrayList<>(linesForEachThread);
        for (String eachLine : currentThreadLines) {
            String[] eachLineWords = eachLine.toLowerCase().split("([,.\\s]+)");
            synchronized (wordFrequencies) {
                for (String eachWord : eachLineWords) {
                    if (wordFrequencies.containsKey(eachWord)) {
                        wordFrequencies.replace(eachWord, wordFrequencies.get(eachWord) + 1);
                    }
                    wordFrequencies.putIfAbsent(eachWord, 1);
                }
            }
        }
    }
}

Надеюсь на хорошие отзывы и заранее благодарю за помощь.

Answer 1

Вы должны дождаться закрытия всех потоков, прежде чем печатать результаты.

public class WordFrequencyRunnableTest {

    List<Thread> threads = new ArrayList<>();
    public static void main(String[] args) throws IOException {
        ...
        ...
        Map<String, Integer> wordFrequencies = runnableTest.parseLines(filePath);
        for(Thread thread: threads)
        {
           thread.join();
        }
        runnableTest.printResult(wordFrequencies);
        ...
        ...
    }

    public Map<String, Integer> parseLines(String filePath) throws IOException {
        Map<String, Integer> wordFrequencies = new HashMap<>();
        try (BufferedReader bufferedReader = new BufferedReader(new FileReader(filePath))) {
            String eachLine = bufferedReader.readLine();
            while (eachLine != null) {
                List<String> linesForEachThread = new ArrayList<>();
                while (linesForEachThread.size() != 100 && eachLine != null) {
                    linesForEachThread.add(eachLine);
                    eachLine = bufferedReader.readLine();
                }
                WordFrequencyUsingRunnable task = new WordFrequencyUsingRunnable(linesForEachThread, wordFrequencies);
                Thread thread = new Thread(task);
                thread.start();
                threads.add(thread); // Add thread to the list.
            }
        }
        return wordFrequencies;
    }
}

PS - Вы можете использовать ConcurrentHashMap<String, AtomicInteger>, чтобы избежать синхронизации доступа к хэш-карте. Так программа будет работать быстрее.