У меня поиск который собирает txt
файлы в очередь ConcurrentLinkedQueue
, и потом в нескольких потоках одновременно проверяет содержимое этих файлов на соответствие заданной строке. В общем многопоточный поиск файла содержащего какую-то строку. Когда один из потоков находит заданную строку, он останавливает все остальные. Сам обход файловой системы у меня выполняется в отдельном потоке.
Для пула потоков использую ExecutorService
.
Вот такой Consumer
:
public class ConsumerSearch implements ConsumerGetter<File>, Consumer<File> {
private File result = null;
@Override
public void accept(File result) {
this.result = result;
}
@Override
public File getResult() {
return result;
}
}
Вот общий класс который создает потоки:
public class ParallelSearch implements Parallel {
private AtomicReference<ConsumerSearch<File>> consumer;
// останавливает все потоки (вызывает 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;
public ParallelSearch(final int amountThreads, Consumer<File> consumer) {
this.service = Executors.newFixedThreadPool(amountThreads);
this.amountThreads = amountThreads;
this.container = new LinkedTransferQueue<>();
this.consumer = new AtomicReference<>(consumer);
this.threads = new ArrayList<>(amountThreads);
this.stopper = new Stopper();
}
@Override
public AtomicReference<Consumer<File>> getConsumer() {
return consumer;
}
@Override
public void killConsumer() {
this.consumer.set(null);
}
@Override
public void startParallelSearch(
final String pathToRoot,
final String targetText) {
this.searchByFileSystem(pathToRoot);
this.searchByFilesContent(targetText);
}
// поток обходит файловую систему в поисках 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);
}
// добавляем в пул потоки для поиска строки по содержимому 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, ParallelSearch.this)
.scanQueue(targetText);
}
});
// добавляем в потоки поиск контенту
this.threads.add(taskRead);
}
}
/**
* Останавливает все потоки ссылка его экземпляр есть у всех потоков.
* Если один из потоков найдет искомую строку, вызовет stopAllSearchingThread().
*/
class Stopper {
void stopAllSearchingThread() {
for (Future future : threads) {
future.cancel(true);
}
}
}
}
Вот так выглядят сами потоки:
public class SearchByFileContent implements SearchByFile {
private final Queue<File> paths;
private final ParallelSearch.Stopper stopper;
private final Parallel parallelSearch;
public SearchByFileContent(final Queue<File> paths,
final ParallelSearch.Stopper stopper,
final Parallel parallelSearch) {
this.stopper = stopper;
this.paths = paths;
this.parallelSearch = parallelSearch;
}
@Override
public File scanQueue(final String target) {
while (this.paths.size() != 0 && !Thread.currentThread().isInterrupted()) {
final File next = this.paths.poll();
final File result = getFileWhichContains(target, next);
if (result != null && !result.getName().equals("-1")) {
if (parallelSearch.getConsumer() != null) {
synchronized (parallelSearch.getConsumer()) {
if (this.parallelSearch.getConsumer().get() != null) {
this.parallelSearch.getConsumer().get().accept(result);
this.parallelSearch.killConsumerReference();
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");
}
}
Вызываю вот так:
@Test
public void whenThen() throws ExecutionException, InterruptedException {
ConsumerSearch<File> consumer = new ConsumerSearch();
Parallel parallel = new ParallelSearch(5, consumer);
parallel.startParallelSearch(dir.getAbsolutePath(), "test");
final File result = consumer.getResult();
Assert.assertThat(result.getName(), is("test2.txt"));
}
Иногда работает, иногда падает с NullPointerException
в предпоследней строке теста final File result = consumer.getResult()
. Хотя вроде все логично и должно работать. Помогите поправить эту проблему.
this.searchByFileSystem(pathToRoot); this.searchByFilesContent(targetText);
вот эти две строки у вас могут выполняться параллельно, а по смыслу насколько я понялsearchByFilesContent
должен стартовать строго послеsearchByFileSystem
. То есть внутриsearchByFilesContent
вы запускаете вExecutor
еthis.service.submit(searchByFileSystem);
и сразу же, не дожидаясь выполнения, запускаетеthis.searchByFilesContent(targetText);
. Это вроде как неверно, нужно дождаться окончания чтения списка файлов.