parallelism_homework

parallelism/src/main/java/ru/hh/school/homework/DirAction.java

@@ -0,0 +1,86 @@
+package ru.hh.school.homework;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.RecursiveAction;
+import java.util.concurrent.Future;
+
+import org.slf4j.Logger;
+import static org.slf4j.LoggerFactory.getLogger;
+
+public class DirAction extends RecursiveAction {
+
+    /*
+    Класс, описывающий итерирование по содержимому файла. Если натыкаемся на директорию,
+    рекурсивно вызывается новый экземпляр DirAction
+     */
+
+    private static final Logger LOGGER = getLogger(Launcher.class);
+
+    private Path dirPath;
+
+    // мапа для "топовых" слов "припрятана" в private-аттрибут для потокобезопасности
+    private Map <String, Long> wordMap = new HashMap<>();
+
+    DirAction(Path newDirPath) {
+        dirPath = newDirPath;
+    }
+
+    @Override
+    public void compute() {
+        javaFileIter(dirPath);
+    }
+
+    private void javaFileIter(Path dirPath) {
+        try {
+            for (Path filePath : Files.newDirectoryStream(dirPath)) {
+                if (Files.isDirectory(filePath)) {
+                    DirAction nestedDirAction = new DirAction(filePath);
+                    nestedDirAction.fork();
+                }
+                if (!Files.isDirectory(filePath) && filePath.toString().endsWith(".java")) {
+                    wordMap = StaticMethods.mapCombiner(wordMap,
+                            StaticMethods.naiveCount(filePath));
+                }
+            }
+
+            // из мапы "топовых" слов, собранных по отдельным файлам, собирает top-10
+            // для данной директории dirPath
+            wordMap = StaticMethods.mapTop(wordMap);
+
+            this.outputStringAssemble();
+        }
+        catch (IOException e) {
+            throw new RuntimeException(e);
+        } finally {
+            // даже несмотря на этот join какой-то тред не закрывается
+            // даже после проведения всех требуемых вычислений :(
+            this.join();
+        }
+    }
+
+    private void outputStringAssemble () {
+        SearchCall searchCall = new SearchCall(dirPath);
+        try {
+            String outputString = "";
+            ArrayList<Future<String>> futureList = new ArrayList<>();
+            for (String word : wordMap.keySet()) {
+                Future<String> future = searchCall.getNewString(word);
+                futureList.add(future);
+            }
+            for (Future<String> future : futureList) {
+                outputString = outputString.concat(future.get());
+            }
+            LOGGER.debug(outputString);
+        } catch (InterruptedException | ExecutionException e){
+            throw new RuntimeException();
+        } finally {
+            searchCall.shutdownExecutor();
+        }
+    }
+}

parallelism/src/main/java/ru/hh/school/homework/Launcher.java

@@ -1,81 +1,26 @@
 package ru.hh.school.homework;
 
-import java.io.IOException;
-import java.nio.file.Files;
 import java.nio.file.Path;
-import java.util.Map;
-import java.util.stream.Stream;
-import org.jsoup.Jsoup;
-import org.jsoup.nodes.Document;
-import org.jsoup.nodes.Element;
-import static java.util.Collections.reverseOrder;
-import static java.util.Map.Entry.comparingByValue;
-import static java.util.function.Function.identity;
-import static java.util.stream.Collectors.counting;
-import static java.util.stream.Collectors.groupingBy;
-import static java.util.stream.Collectors.toMap;
+import java.util.concurrent.ForkJoinPool;
 
 public class Launcher {
 
-  public static void main(String[] args) throws IOException {
-    // Написать код, который, как можно более параллельно:
-    // - по заданному пути найдет все "*.java" файлы
-    // - для каждого файла вычислит 10 самых популярных слов (см. #naiveCount())
-    // - соберет top 10 для каждой папки в которой есть хотя-бы один java файл
-    // - для каждого слова сходит в гугл и вернет количество результатов по нему (см. #naiveSearch())
-    // - распечатает в консоль результаты в виде:
-    // <папка1> - <слово #1> - <кол-во результатов в гугле>
-    // <папка1> - <слово #2> - <кол-во результатов в гугле>
-    // ...
-    // <папка1> - <слово #10> - <кол-во результатов в гугле>
-    // <папка2> - <слово #1> - <кол-во результатов в гугле>
-    // <папка2> - <слово #2> - <кол-во результатов в гугле>
-    // ...
-    // <папка2> - <слово #10> - <кол-во результатов в гугле>
-    // ...
-    //
-    // Порядок результатов в консоли не обязательный.
-    // При желании naiveSearch и naiveCount можно оптимизировать.
+    public static void main(String[] args){
 
-    // test our naive methods:
-    testCount();
-    testSearch();
-  }
+        Path path = Path.of("C:\\1_Konstantin\\1_hh\\concurrency\\hh-school-1\\" +
+                "parallelism\\src\\main\\java\\ru\\hh\\school");
 
-  private static void testCount() {
-    Path path = Path.of("d:\\projects\\hh-school\\parallelism\\src\\main\\java\\ru\\hh\\school\\parallelism\\Runner.java");
-    System.out.println(naiveCount(path));
-  }
+        // рекурсивный характер задачи вынудил меня прибегнуть к ForkJoinPool
+        ForkJoinPool pool =
+                new ForkJoinPool(Runtime.getRuntime().availableProcessors());
 
-  private static Map<String, Long> naiveCount(Path path) {
-    try {
-      return Files.lines(path)
-        .flatMap(line -> Stream.of(line.split("[^a-zA-Z0-9]")))
-        .filter(word -> word.length() > 3)
-        .collect(groupingBy(identity(), counting()))
-        .entrySet()
-        .stream()
-        .sorted(comparingByValue(reverseOrder()))
-        .limit(10)
-        .collect(toMap(Map.Entry::getKey, Map.Entry::getValue));
-    }
-    catch (IOException e) {
-      throw new RuntimeException(e);
-    }
-  }
-
-  private static void testSearch() throws IOException {
-    System.out.println(naiveSearch("public"));
-  }
+        DirAction startDirAction = new DirAction(path);
 
-  private static long naiveSearch(String query) throws IOException {
-    Document document = Jsoup //
-      .connect("https://www.google.com/search?q=" + query) //
-      .userAgent("Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/72.0.3626.110 Safari/537.36 Viv/2.3.1440.48") //
-      .get();
+        pool.invoke(startDirAction);
 
-    Element divResultStats = document.select("div#resultStats").first();
-    return Long.valueOf(divResultStats.text().replaceAll("[^0-9]", ""));
-  }
-
-}
+        // даже несмотря на это программа не завершается
+        // и после проведения всех требуемых вычислений:
+        // видимо, где-то не сджойнился тред
+        pool.shutdownNow();
+    }
+}

parallelism/src/main/java/ru/hh/school/homework/SearchCall.java

@@ -0,0 +1,42 @@
+package ru.hh.school.homework;
+
+import java.nio.file.Path;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+public class SearchCall {
+
+    /*
+    *
+    Класс, реализующий асинхронную отсылку запросов на гугл.
+
+    Не придумал, как засунуть эти задачи в общий ForkJoinPool,
+    поэтому для каждой директории будет создаваться свой класс SearchCall,
+    в котором будет скромный тредпул на пять потоков.
+
+    Таким образом, наибольшее количество потоков, которое может получиться:
+    20 = 5 (SearchCall) * 4 (на главном ForkJoinPool)
+     *
+     */
+
+    private Path dirPath;
+    private ExecutorService executor = Executors.newFixedThreadPool(5);
+
+    SearchCall(Path newDirPath) {
+        dirPath = newDirPath;
+    }
+
+    public Future <String> getNewString (String word) {
+        return executor.submit(() -> {
+            return String.format("%s - %s - %d \n",
+                    dirPath.toString(), word, StaticMethods.naiveSearch(word));
+        });
+    }
+
+    // Конечно, после выполнения запросов весь этот класс будет убит garbege collector'ом,
+    // но все же я решил предусмотреть остановку Executor-сервиса
+    public void shutdownExecutor(){
+        executor.shutdownNow();
+    }
+}

parallelism/src/main/java/ru/hh/school/homework/StaticMethods.java

@@ -0,0 +1,63 @@
+package ru.hh.school.homework;
+
+import org.jsoup.Jsoup;
+import org.jsoup.nodes.Document;
+import org.jsoup.nodes.Element;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static java.util.Collections.reverseOrder;
+import static java.util.Map.Entry.comparingByValue;
+import static java.util.function.Function.identity;
+import static java.util.stream.Collectors.*;
+
+public class StaticMethods {
+
+    protected static Map<String, Long> mapTop(Map<String, Long> wordMap) {
+        return wordMap.entrySet()
+            .stream()
+            .sorted(comparingByValue(reverseOrder()))
+            .limit(10)
+            .collect(toMap(Map.Entry::getKey, Map.Entry::getValue));
+    }
+
+    // сливает две мапы в одну, суммируя численные значения с одинаковым ключом
+    protected static Map<String, Long> mapCombiner (
+            Map<String, Long> map1, Map<String, Long> map2) {
+        return Stream.concat(map1.entrySet().stream(), map2.entrySet().stream())
+                .collect(Collectors.groupingBy(Map.Entry::getKey,
+                     Collectors.summingLong(Map.Entry::getValue)));
+    }
+
+
+    protected static Map<String, Long> naiveCount(Path path) {
+        try {
+            return Files.lines(path)
+            .flatMap(line -> Stream.of(line.split("[^a-zA-Z0-9]")))
+            .filter(word -> word.length() > 3)
+            .collect(groupingBy(identity(), counting()))
+            .entrySet()
+            .stream()
+            .sorted(comparingByValue(reverseOrder()))
+            .limit(10)
+            .collect(toMap(Map.Entry::getKey, Map.Entry::getValue));
+        }
+        catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    protected static long naiveSearch(String query) throws IOException {
+        Document document = Jsoup //
+        .connect("https://www.google.com/search?q=" + query) //
+        .userAgent("Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/72.0.3626.110 Safari/537.36 Viv/2.3.1440.48") //
+        .get();
+        Element divResultStats = document.select("div#resultStats").first();
+        return Long.valueOf(divResultStats.text().replaceAll("[^0-9]", ""));
+    }
+}