1

У меня поиск который собирает txt файлы в очередь ConcurrentLinkedQueue, и потом в нескольких потоках одновременно проверяет содержимое этих файлов на соответствие заданной строке. В общем многопоточный поиск файла содержащего какую-то строку. Когда один из потоков находит заданную строку, он останавливает все остальные. Сам обход файловой системы у меня выполняется в отдельном потоке.

Для пула потоков использую ExecutorService.

Но когда я начал тестировать код то получилось что правильно срабатывает только один раз из 10 примерно, во всех остальных случаях падает с java.util.concurrent.CancellationException, причем указывает на строку в тесте где вызывается метод get() у Future.

Помогите пожалуйста разобраться что я не так делаю?

Основной класс в котором ExecutorService создается:

public class ParallelSearch implements Parallel {
    // останавливает все потоки (вызывает cancel в цикле по threads).
    private final Stopper stopper;
    // лист запущенных потоков.
    private final List<Future<File>> threads;
    // размер пула потоков.
    private final int amountThreads;
    // очередь в которой храняться файлы.
    private final Queue<File> container;
    private final ExecutorService service;
    // поток поиска txt файлов по диску.
    private Thread searchByFileSystem;

    public ParallelSearch(final int amountThreads) {
        this.service = Executors.newFixedThreadPool(amountThreads);
        this.amountThreads = amountThreads;

        this.container = new ConcurrentLinkedQueue<>();
        this.threads =   new ArrayList<>(amountThreads);
        this.stopper =   new Stopper();
    }

    @Override
    public List<Future<File>> startParallelSearch(
            final String pathToRoot,
            final String targetText) {

        this.searchByFileSystem(pathToRoot);
        this.searchByFilesContent(targetText);
        return this.threads;
    }

    // поток обходит файловую систему в поисках txt файлов и добавляет в очередь.
    private void searchByFileSystem(final String pathToRoot) {
        final Thread searchByFileSystem = new Thread(new Runnable() {
            @Override
            public void run() {
                new CollectingTXTFiles(
                        pathToRoot, container)
                        .collectTXTFromRoot();
            }
        });
        this.service.submit(searchByFileSystem);
        this.searchByFileSystem = searchByFileSystem;
    }

    // добавляем в пул потоки для поиска строки по содержимому txt файлов
    private void searchByFilesContent(final String targetText) {
        for (int i = 0; i < (this.amountThreads - 1); i++) {
            final Future<File> taskRead =
                    this.service.submit(new Callable<File>() {
                        @Override
                        public File call() throws Exception {
                            return new SearchByFileContent(container, stopper)
                                    .scanQueue(targetText);
                        }
                    });
            // добавляем потоки поиска по контенту в лист Future'ов
            this.threads.add(taskRead);
        }
    }

    /**
     * Останавливает все потоки ссылка его экземпляр есть у всех потоков.
     * Если один из потоков найдет искомую строку, вызовет stopAllSearchingThread().
     */
    class Stopper {
        void stopAllSearchingThread() {
            searchByFileSystem.interrupt();
            for (Future future : threads) {
                future.cancel(true);
            }
        }
    }
}

Вот так я вызываю:

@Test
public void whenThen() throws ExecutionException, InterruptedException {
    Parallel parallel = new ParallelSearch(5);
    final List<Future<File>> futures = parallel.startParallelSearch(dir.getAbsolutePath(), "test");

    String result = null;

    for (Future<File> f : futures) {
        String name = f.get().getName();
        if (!name.equals("-1")) {
            result = name;
        }
    }

    Assert.assertThat(result, is("test2.txt"));
}

Класс который по файловой системе ищет txt файлы в паралельном потоке, запускается тем-же объектом ExecutorService.:

public class CollectingTXTFiles implements Collector {
    private final String root;
    private final Queue<File> files;

    public CollectingTXTFiles(final String root,
                              final Queue<File> files) {
        this.root = root;
        this.files = files;
    }

    @Override
    public void collectTXTFromRoot() throws IllegalArgumentException {
        final File root = new File(this.root);

        if (!root.exists() || !root.isDirectory()) {
            throw new IllegalArgumentException(String.format(
                    "Directory: '%s' not found", this.root));
        }

        this.selectTXTFiles(root);
    }

    private void selectTXTFiles(final File file) {
        if (Thread.currentThread().isInterrupted()) {
            return;
        }

        if (file.getName().endsWith(".txt") &&
                !file.isDirectory()) {

            this.files.add(file);
        }

        if (file.isDirectory()) {
            final File[] childes = file.listFiles();
            assert Objects.nonNull(childes);

            for (File f : childes) {
                selectTXTFiles(f);
            }
        }
    }
}

Класс осуществляющий поиск строки, по содержанию каждого файла в отдельности:

public class SearchByFileContent implements SearchByFile {
    private final Queue<File> paths;
    private final ParallelSearch.Stopper stopper;


    public SearchByFileContent(final Queue<File> paths,
                               final ParallelSearch.Stopper stopper) {
        this.stopper = stopper;
        this.paths = paths;
    }

    @Override
    public File scanQueue(final String target) {
        while (this.paths.size() != 0 && !Thread.currentThread().isInterrupted()) {
            System.out.println(paths.size());
            final File next = this.paths.poll();
            final File result = getFileWhichContains(target, next);

            if (result != null && !result.getName().equals("-1")) {
                this.stopper.stopAllSearchingThread();
                return result;
            }
        }
        return new File("-1");
    }

    private File getFileWhichContains(final String target, final File file) {
        final StringBuilder sb = new StringBuilder();
        if (file == null) return new File("-1");
        try (BufferedReader reader = new BufferedReader(
                new FileReader(file))
        ) {

            String line;

            while (Objects.nonNull(line = reader.readLine())) {
                sb.append(line);
            }

            if (new String(sb).contains(target)) {
                return file;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return new File("-1");
    }
}

Последние 2 класса, в однопоточном режиме, юнит тесты проходят нормально.

Может быть у меня обход файлов идет впереди их сбора в очередь? Или что-то еще? Помогите пожалуйста это решить.

1 ответ 1

1
 final Thread searchByFileSystem = new Thread(new Runnable() {
            @Override
            public void run() {
                new CollectingTXTFiles(
                        pathToRoot, container)
                        .collectTXTFromRoot();
            }
        });
        this.service.submit(searchByFileSystem);

А зачем создаете Thread если потом передаете его в ExecutorService? Делайте сразу Runnable. Вы же тред не запускаете через start() а через пул.

А вообще причина простая. Когда вы делаете get() вы ждете завершение работы таска и возврата им результата. Представьте что первый таск из списка работает 1 мин, а второй 10 секунд (и при этом находит текст). В этой ситуации вы получите исключение, т.к. через 10 секунд будет найдет файл, все таски получат событие отмены, а ваш тест ждет ответа от того таска, который работал но его отменили.

Что бы все таки получить результат работы вашего кода, стоит использовать систему колбеков, например на базе Consumer. Для этого надо:

Добавить в базовый класс ссылку на Consumer и передать его в конструкторе, т.е.

public class ParallelSearch implements Parallel {
   private Consumer<File> callback;
   ...
   public ParallelSearch(final int amountThreads, Consumer<File> callback) {
        this.callback = callback;

  public Consumer<File> getCallback() { 
        return callback;
  }

  public void setCallback(Consumer<File> callback){
       this.callbacl = callback;
  } 

Затем, модифицировать метод scanQueue (только прокиньте в класс с ним ссылку на ParallelSearch) изменив

if (result != null && !result.getName().equals("-1")) {
            this.stopper.stopAllSearchingThread();
            return result;
}

на

 if (result != null && !result.getName().equals("-1")) {
      if (parallelSearch.getConsumer() != null){
          synchronized (parallelSearch.getConsumer()){
                 if (parallelSearch.getConsumer() != null){
                     parallelSearch.getConsumer().accept(result);
                     parallelSearch.setConsumer(null);
                 }
          }
      }      
      this.stopper.stopAllSearchingThread();
      return result;
 }

Единственное, что скорей всего саму ссылку на Consumer надо завернуть в AtomicReference, но это уж сами.

4
  • А как же мне тогда поступить? Мне надо где-то вызвать результат а другого способа кроме get() вроде нет... Должен же быть какой-то способ получить результат именно от того потока который нашел?
    – Pavel
    3 апр 2017 в 16:25
  • Самый простой и правильный - кинуть calbacl из задачи, которая нашла файл. Т.е. просто передавайте в ваши потоки поиска файлов ссылку на Consumer в котором вы уже будете обрабатывать результат поиска. Но только учтите, что может быть ситуация когда она вызовется нескольким задачами, если они одновременно нашли файл и не успели проверить себя на interrupted и завершится. Есть и множество других вариантов - от CompltableFuture до Condition и барьеров. Но это сложней. 3 апр 2017 в 16:35
  • Если я пошлю в каждый поток ссылку на Consumer так же как я сделал это со Stopper то что дальше? у этого Consumer будет поле которое должно принять результат но как мне понять что результат получен? проверять что оно не null и засыпать? Или как это делается?
    – Pavel
    3 апр 2017 в 17:04
  • Дополнил ответ. Заодно решил проблему с возможность нескольких вызовов консюмера, которую озвучил в предыдущем комментарии. 3 апр 2017 в 17:22

Ваш ответ

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

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