Iterieren einer Ordnerstruktur parallel zu Dateien.NewDirectoryStream, warten ⇐ Java

[Anonymous]

Ich verwende Dateien.NewDirectoryStream, um eine Verzeichnisstruktur rekursiv zu iterieren. Ich möchte mit einem Ausführungsdienst verwenden, um das Verzeichnis parallel zu erreichen (Dateien sind über eine langsame Netzwerkverbindung und Latenz führt dazu, dass unser ursprünglicher Suchansatz für ein Thread die meiste Zeit blockiert wird).

Code: Select all

ExecutorService executorService = Executors.newFixedThreadPool(numThreads);

public void walkDirectory(Path directory) {
try (DirectoryStream stream = Files.newDirectoryStream(directory)) {
for (Path entry : stream) {

DosFileAttributes attribs =  getBasicFileAttributes(entry);
if (attribs.isDirectory()) {
executorService.submit(() -> walkDirectory(entry));
} else {
process(entry, attribs);
}
}
}
}
< /code>
Ich muss jetzt blockieren, bis die Iteration abgeschlossen ist.  Beachten Sie auch, dass die Threads des Executor -Dienstes Jobs an denselben Executor -Service senden müssen. Besserer /eleganterer Weg, um dieses Parallelisierungsziel zu erreichen? Testen: < /p>
Ich erstelle einen Atomicinteger -Zähler, der die Anzahl der zu Fuß erreichten Verzeichnisse im Auge behält.  Wenn dieser Zähler auf Null fällt, wissen wir, dass der Traversal vollständig ist. Das. < /p>
Wenn jemand Feedback hat, würde ich es schätzen, es zu hören.  < /p>
public class FolderWalker {

private final ExecutorService executorService;
private final PathProcessor processor;
private final PathFilter fileFilter;
private final PathFilter directoryFilter;
private final ExceptionHandler exceptionHandler;
private final AtomicInteger activeThreads = new AtomicInteger(0);
private final Object lock = new Object();

private boolean keepRunning = true;

public static interface PathProcessor{
public void process(Path p, DosFileAttributes attribs) throws Exception;
}

public static interface ExceptionHandler{
public boolean handle(Path p, Exception e); // return true to continue traversal
}

public static interface PathFilter{
public boolean accept(Path p, DosFileAttributes attribs);

public static PathFilter ACCEPT_ALL = (p, attribs) -> true;
}

public FolderWalker(int numThreads, PathFilter directoryFilter, PathFilter fileFilter, PathProcessor processor, ExceptionHandler exceptionHandler) {
this.executorService = createExecutorService(numThreads);
this.directoryFilter = directoryFilter;
this.fileFilter = fileFilter;
this.processor = processor;
this.exceptionHandler = exceptionHandler;
}

private static ExecutorService createExecutorService(int numThreads) {

// if threads are all in use, then the calling thread has to do the work - this will keep us from having excessive memory overhead of submitted jobs
return new ThreadPoolExecutor(numThreads, numThreads, 0, TimeUnit.MILLISECONDS,
new SynchronousQueue(),
new ThreadPoolExecutor.CallerRunsPolicy());
}

public void walkDirectory(Path directory) {
activeThreads.incrementAndGet();
executorService.submit(() -> walkDirectoryInternal(directory));
}

private void walkDirectoryInternal(Path directory) {
int activeThreadCount = activeThreads.get();
System.out.println(Thread.currentThread() + " - Before: Looking in " + directory + " - Active folders: " + activeThreadCount);

try {
try (DirectoryStream stream = Files.newDirectoryStream(directory)) {
for (Path entry : stream) {
// TODO: How do we want to handle exceptions?
//if (thrownException != null) return;
if (Thread.interrupted()) return;
if (!keepRunning) return;

try {
DosFileAttributes attribs =  getBasicFileAttributes(entry);
if (attribs.isDirectory() && directoryFilter.accept(entry, attribs)) {
activeThreads.incrementAndGet();
executorService.submit(() -> walkDirectoryInternal(entry));
} else if (fileFilter.accept(entry, attribs)) {
processor.process(entry, attribs);
}
} catch (Exception e) {
if (!exceptionHandler.handle(entry, e))
keepRunning = false;
}
}
} catch (Exception e) {
if (!exceptionHandler.handle(directory, e))
keepRunning = false;
}
} finally {
activeThreadCount = activeThreads.decrementAndGet();
System.out.println(Thread.currentThread() + " - After: Looking in " + directory + " - Active folders: "  + activeThreadCount);
if (activeThreadCount == 0) {
synchronized(lock) {
lock.notifyAll();
}
}
}

}

private DosFileAttributes getBasicFileAttributes(Path p) throws IOException {
return Files.getFileAttributeView(p, DosFileAttributeView.class).readAttributes();
}

public void await() throws Exception {
synchronized(lock) {
while (activeThreads.get() != 0)
lock.wait();
}

executorService.shutdown();
System.out.println("Thread pool is shut down");
try {
if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
executorService.shutdownNow();
}
} catch (InterruptedException e) {
executorService.shutdownNow();
}
System.out.println("Thread pool is terminated");

}

}