0

Вижу данный подход:

1.Поток Отбирает файлы с нужным расширением

Создаем Map где ключ Path а Содержание List-файла И добавляем туда все найденные файлы

2.Поток берет данные с этой карты и Ищит в значениях совпадение по тексту!

Дальше заносит Путь файла в Лист=получаем результат.

Последовательно программа работает! Причем довольна быстро!

Вопрос в том как настроить именно параллельную работу! Вот мои потуги по этому поводу!Результат не убедительный, так как программа вс равно работает последовательно!

1ый поток запускает метод WalkTree

public class ThreadSearch extends Thread {


    private String dirr;

    private MyFileVisitorParallSearch myfl;

    public ThreadSearch(String dirr, MyFileVisitorParallSearch myfl) {
        this.dirr = dirr;
        this.myfl = myfl;
    }

    @Override
    public void run() {
        try {
            Thread.sleep(10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("Дерево каталогов начиная с каталога: " + dirr + "\n");
        try {
            Files.walkFileTree(Paths.get(dirr), myfl);
        } catch (IOException e) {
            System.out.println("Ошибка Ввода Вывода " + e);
        }

    }
}

Мой интерфейс Визитор

public class MyFileVisitorParallSearch extends SimpleFileVisitor<Path> {


    private String extc;


    private List<String> resultList = new ArrayList<>();

    private HashMap<Path, List<String>> map = new HashMap<>();


    public MyFileVisitorParallSearch(String extc) {

        this.extc = extc;


    }

    public HashMap<Path, List<String>> getMap() {

        return map;
    }


    public List<String> getResultlist() {
        return resultList;
    }



    @Override

    public FileVisitResult preVisitDirectory(Path path, BasicFileAttributes attribs) throws IOException {

        //Ignore Dirr

        Path ignorepath1 = Paths.get("C:\\projects\\java-a-to-z\\.git");

        Path ignorepath2 = Paths.get("C:\\projects\\java-a-to-z\\.idea");


        if (path.equals(ignorepath1) || path.equals(ignorepath2)) {

            return FileVisitResult.SKIP_SUBTREE;

        } else {

            return FileVisitResult.CONTINUE;
        }

    }

    @Override
    public FileVisitResult visitFile(Path path, BasicFileAttributes attributes) throws IOException {

        String myRegex = "\\w+\\." + extc;

        Pattern p = Pattern.compile(myRegex);
        Matcher m = p.matcher(String.valueOf(path));

        if (m.find()) {
            List<String> list = Files.lines(path).collect(Collectors.toList());

            map.put(path, list);
            return FileVisitResult.CONTINUE;

        }
        return FileVisitResult.SKIP_SUBTREE;

    }

}

Второй поток работает с мапой и передает результат в Лист результатов

public class ThreadFindText extends Thread {

    private String text;

    private List<String> resultList = new ArrayList<>();

    private MyFileVisitorParallSearch myfl;

    int count = 0;


    public ThreadFindText(String text, MyFileVisitorParallSearch myfl) {

        this.text = text;
        this.myfl = myfl;

    }

    public List<String> getResultList() {

        return resultList;
    }


    public void run() {

        try {
            Thread.sleep(10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }


        for (Path path : myfl.getMap().keySet()) {
            for (String element : myfl.getMap().get(path)) {

                Pattern pfind = Pattern.compile("\\b" + text + "\\b");
                Matcher mfind = pfind.matcher(element);

                if (mfind.find()) {
                    myfl.getResultlist().add(String.valueOf(path));
                    break;

                }

            }
        }
    }
}

Метод Маин

public class MainTest {

    public static void main(String[] args) throws InterruptedException {

        String dir = "C:";

        String text = "MyFileVisitorParallSearch";

        String extc = "java";

        MyFileVisitorParallSearch myfl = new MyFileVisitorParallSearch(extc);
        ThreadFindText threadFind = new ThreadFindText(text, myfl);
        ThreadSearch threadSearch = new ThreadSearch(dir, myfl);



        Long startTime = System.currentTimeMillis();

        threadSearch.start();
        threadSearch.join();

        threadFind.start();
        threadFind.join();

        Long finishTime = System.currentTimeMillis() - startTime;


        for (String element : myfl.getResultlist()) {

            System.out.println(element);
        }

        System.out.println(finishTime);
    }


}
3
  • Метод main, если уж на то пошло, должен "задержать" выполнение программы. Его задача мониторить состояние потоков, вы же, повесили на него счётчик, который не нисёт никакого смысла XD, дельта времени не будет говорить о реальном времени работы потоков, она покажет скорость запуска потока и join функции. У потоков должна быть общая память. Для того чтобы коллекция в общей памяти могла параллельно модифицироваться и читаться - необходимо использовать коллекции которые это поддерживают, либо прибегать к хитростям. Сейчас же, ваш код не получится запустить параллельно.
    – test123
    16 янв 2018 в 9:53
  • 1. Первый поток может искать файлы и читать их содержимое в общую память. 2. Второй поток может извлекать из общей памяти содержимое файлов и искать вхождение, после чего выбрасывать просмотренные файлы и сохранять соответствующие критерию поиска содержимого в другую коллекцию. В таком случае - код будет работать параллельно. Ну, и не забывайте, что скорость в многопоточном варианте - упадёт, может, даже значительно. main может мониторить текущий список файлов в общей памяти, и если он пуст - и потоки отработали, значит все файлы успешно просмотрены.
    – test123
    16 янв 2018 в 9:58
  • 1
    А вы смотрели, где узкое место: в чтении или в фильтрации? 8 июл 2020 в 11:20

1 ответ 1

0

Насколько я понял, то распараллелить нужно именно чтение файлов, которые получаем от FileVisitor'а? Это будет примерно так:

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.*;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;

public class TextSearcherInTree {
    private static final String TARGET_STRING = "Bill Gates".toLowerCase();
    private static final String START_TREE_PATH = "C:\\Program Files\\";

    private static int totalFiles, totalDirs = 0;
    private static AtomicInteger matches = new AtomicInteger(0);
    private static AtomicInteger totalFails = new AtomicInteger(0);
    private static AtomicInteger jobs = new AtomicInteger(0);

    public static void main(String[] args) throws Exception {
        final int PARALLELISM_LEVEL = Runtime.getRuntime().availableProcessors() * 4;
        ExecutorService service = Executors.newFixedThreadPool(PARALLELISM_LEVEL);

        Files.walkFileTree(Paths.get(START_TREE_PATH), new SimpleFileVisitor<Path>() {
            @Override
            public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
                ++totalFiles;
                service.submit(getJob(file));
                return FileVisitResult.CONTINUE;
            }

            @Override
            public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
                ++totalDirs;
                return FileVisitResult.CONTINUE;
            }

            @Override
            public FileVisitResult visitFileFailed(Path file, IOException exc) throws IOException {
                totalFails.incrementAndGet();
                System.out.println("Visit fail on '" + file.toAbsolutePath().toString() + "'");
                return FileVisitResult.CONTINUE;
            }
        });
        service.shutdown();
        service.awaitTermination(10, TimeUnit.MINUTES);

        System.out.println("Tree of '" + START_TREE_PATH + "': files=" + totalFiles + ", dirs=" + totalDirs + ", fails=" + totalFails);
        System.out.println("Founded " + matches.get() + " matches of '" + TARGET_STRING + "' in " + jobs.get() + " jobs");
    }

    private static Runnable getJob(Path file) {
        return () -> {
            jobs.incrementAndGet();
            try (Stream<String> lines = Files.lines(file, StandardCharsets.UTF_8)) {
                lines/*.parallel()*/.map(String::toLowerCase).filter(line -> line.contains(TARGET_STRING)).forEach(line -> {
                    matches.incrementAndGet();
                    System.out.println(file.toAbsolutePath().toString() + "\tline: " + line);
                });
            } catch (IOException e) {
                totalFails.incrementAndGet();
                System.out.println(e.toString());
            }
        };
    }
}

Если нужно параллельное чтение строк файла(для поиска) - раскомментируйте /.parallel()/

Если нужно распаралелить ещё и обход всех деревьев (walkFileTree), то примерно так, но не думаю, что это хорошая идея для HDD:

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.*;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;

public class TextSearcherInConcurrentTree {
    private static final String TARGET_STRING = "Bill Gates".toLowerCase();
    private static final String START_TREE_PATH = "C:\\Program Files\\";

    private static final int PARALLELISM_LEVEL = Runtime.getRuntime().availableProcessors() * 2;
    private static final ExecutorService service = Executors.newFixedThreadPool(PARALLELISM_LEVEL);

    private static AtomicInteger totalFiles = new AtomicInteger(0);
    private static AtomicInteger totalDirs = new AtomicInteger(0);
    private static AtomicInteger totalFails = new AtomicInteger(0);
    private static AtomicInteger matches = new AtomicInteger(0);
    private static AtomicInteger jobs = new AtomicInteger(0);

    public static void main(String[] args) throws Exception {
        RecursiveWalk w = new RecursiveWalk(Paths.get(START_TREE_PATH).toRealPath());
        ForkJoinPool p = new ForkJoinPool(PARALLELISM_LEVEL);
        p.invoke(w);
        p.awaitQuiescence(10, TimeUnit.MINUTES);

        service.shutdown();
        service.awaitTermination(10, TimeUnit.MINUTES);

        System.out.println("Tree of '" + START_TREE_PATH + "': files=" + totalFiles.get() + ", dirs=" + totalDirs.get() + ", fails=" + totalFails.get());
        System.out.println("Founded " + matches.get() + " matches of '" + TARGET_STRING + "' in " + jobs.get() + " jobs");
    }

    private static Runnable getJob(Path file) {
        return () -> {
            jobs.incrementAndGet();
            try (Stream<String> lines = Files.lines(file, StandardCharsets.UTF_8)) {
                lines/*.parallel()*/.map(String::toLowerCase).filter(line -> line.contains(TARGET_STRING)).forEach(line -> {
                    matches.incrementAndGet();
                    System.out.println(file.toAbsolutePath().toString() + "\tline: " + line);
                });
            } catch (IOException e) {
                totalFails.incrementAndGet();
                System.out.println(e.toString());
            }
        };
    }

    private static class RecursiveWalk extends RecursiveAction {
        private final Path path;

        RecursiveWalk(Path path) {
            this.path = path;
        }

        @Override
        protected void compute() {
            final List<RecursiveWalk> walks = new ArrayList<>();
            try {
                Files.walkFileTree(path, new SimpleFileVisitor<Path>() {
                    @Override
                    public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
                        totalFiles.incrementAndGet();
                        service.submit(getJob(file));
                        return FileVisitResult.CONTINUE;
                    }

                    @Override
                    public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
                        if (!dir.equals(RecursiveWalk.this.path)) {
                            RecursiveWalk w = new RecursiveWalk(dir);
                            w.fork();
                            walks.add(w);
                            return FileVisitResult.SKIP_SUBTREE;
                        } else {
                            totalDirs.incrementAndGet();
                            return FileVisitResult.CONTINUE;
                        }
                    }

                    @Override
                    public FileVisitResult visitFileFailed(Path file, IOException exc) throws IOException {
                        totalFails.incrementAndGet();
                        System.out.println("Visit fail on '" + file.toAbsolutePath().toString() + "'");
                        return FileVisitResult.CONTINUE;
                    }
                });
            } catch (IOException e) {
                e.printStackTrace();
            }

            for (RecursiveWalk walk : walks) {
                walk.join();
            }
        }
    }
}

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.