Как получить доступ к результату работы потоков над общим ресурсом

105

23 августа 2019, 04:10

Программа решает следующую задачу: есть несколько файлов с текстом, на каждый файл через пул потоков создается поток для открытия и чтения файла и составление из прочитанного текста списка из строк List<String>, далее предполагается использование общего для всех потоков Map<String, Long> для сбора статистики следующим образом - ключом(key) будет соответственно строка из списка, а значение(value) равно 1, но если такой ключ уже присутствует т.е. в списке и в других списках имеются одинаковые строки, то инкрементировать значение на единицу

Если первый файл содержит строки "Hello", второй соответственно две строки "Java" и "World", а третий файл "Hello", то результирующий map должен выглядеть так:

{"Hello"=2,"Java"=1, "World"=1}

Моя реализация: Все работает корректно

В методе makeStatistics()создается экземпляр класса (Statistic), где находится общий map, далее метод makeStatistics() с помощью метода getFilesPaths() вычисляет пути к файлам с расширением .txt и заносит из в список, далее создает фиксированный пул потоков размером, равным количеству текстовых файлов в папке(1 файл - 1 поток), потом в цикле создаются экземпляры классов для потока ReaderThread, в его конструктор передается экземпляр класса (Statistic) с общим map-ом и путь к файлу

public ArrayList<Path> getFilesPaths() {
    ArrayList<Path> filesNames = null;
    try (Stream<Path> files = Files.walk(Paths.get("D:\\DIR"))) {
        filesNames= files
                    .filter(file -> file.toString().endsWith(".txt"))
                    .collect(Collectors.toCollection(ArrayList::new));
        filesNames.forEach(System.out::println);
    } catch (IOException e) {
        e.printStackTrace();
    }
    return filesNames;
}
public void makeStatistics() {
    Statistic statistic = new Statistic();
    ArrayList<Path> logPaths = getFilesPaths();
    ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(logPaths.size());
    for (Path filePath : logPaths)
    {
        ReaderThread logReader = new ReaderThread(statistic, filePath);
        executor.execute(logReader);
    }
    executor.shutdown();
}

Класс потока выглядит так, он просто читает файл, по переданному в него пути и сохраняет строчки в список, потом в строке statistic.setStatisticMap(textLines); список отправляется на изменение общего map-а в классе Statistic

public class ReaderThread implements Runnable {
private Path filePath;
private Statistic statistic;
public ReaderThread(Statistic statistic, Path filePath) {
    this.filePath = filePath;
    this.statistic = statistic;
}
public Path getFile() {
    return filePath;
}
public void run() {   
        List<String> textLines;
        try(Stream<String> lineStream = Files.newBufferedReader(filePath).lines()) {
            textLines = lineStream
                    .collect(Collectors.toCollection(ArrayList::new));
            statistic.setStatisticMap(textLines);
        } catch (IOException e) {
            e.printStackTrace();
        }
}

А вот и класс с общим map-ом:

public class Statistic {
private Map<String, Long> statisticMap;
public Statistic() {
    statisticMap = new HashMap<>();
}
public synchronized void setStatisticMap(List<String> logLines) {
    statisticMap = logLines.stream()
                            .collect(Collectors.toMap(Function.identity(), v -> 1L, Long::sum));
    System.out.println(statisticMap);
}
public Map<String, Long> getStatisticMap() {
    return statisticMap;
}

Программа выводит на экран правильный результат, но меня интересует как и в каком месте получить доступ к уже построенному общему map-у(statisticMap), т.е. как и где получить доступ к statisticMap после того как его сформируют потоки и он будет заполнен?

Answer 1

executor.shutdown();

Не будет дожидаться выполнения всех переданных задач. Самый простой способ, воспользоваться методом invokeAll и немного переделать реализацию.

public class ReaderThread implements Callable<Object> {
    private Path filePath;
    private Statistic statistic;
    public ReaderThread(Statistic statistic, Path filePath) {
        this.filePath = filePath;
        this.statistic = statistic;
    }
    public Path getFile() {
        return filePath;
    }
    @Override
    public Object call() throws Exception {
        List<String> textLines;
        try (Stream<String> lineStream = Files.newBufferedReader(filePath).lines()) {
            textLines = lineStream.collect(Collectors.toCollection(ArrayList::new));
            statistic.setStatisticMap(textLines);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
}

И вызывать его через executor так:

    List<ReaderThread> tasks = new ArrayList<>();
    for (Path filePath : logPaths) {
        tasks.add(new ReaderThread(statistic, filePath));
    }
    try {
        executor.invokeAll(tasks);
    } catch (InterruptedException e) {
        e.printStackTrace();
    } finally {
        executor.shutdown();
    }
    System.out.println(statistic.getStatisticMap());