Параллельная версия Files.walkFileTree (Java или Scala)
Кто-нибудь знает какой-либо параллельный эквивалент java Files.walkFileTree или что-то подобное? Это может быть библиотека Java или Scala.
3 ответа
Давайте предположим, что выполнения обратного вызова для каждого файла достаточно.
Этот код не будет обрабатывать циклы в файловой системе - вам нужен реестр, где вы были для этого (например, java.util.concurrent.ConcurrentHashMap). Вы можете добавить всевозможные улучшения, например, сообщать об исключениях вместо того, чтобы просто игнорировать их.
import java.io.File
import scala.util._
def walk(f: File, callback: File => Unit, pick: File => Boolean = _ => true) {
Try {
val (dirs, fs) = f.listFiles.partition(_.isDirectory)
fs.filter(pick).foreach(callback)
dirs.par.foreach(f => walk(f, callback, pick))
}
}
Сбор файлов с использованием сгиба вместо foreach не намного сложнее, но я оставляю это в качестве упражнения для читателя. (A ConcurrentLinkedQueue возможно, достаточно быстрый, чтобы в любом случае принять их все в обратном вызове, если только у вас не очень медленные потоки и отличная файловая система.)
Как уже отмечали другие, обход дерева файлов почти наверняка связан с вводом-выводом, а не с процессором, поэтому польза от многопоточного обхода дерева файлов вызывает сомнения. Но если вы действительно хотите, вы могли бы свернуть свой собственный с ForkJoinPool или похожие.
import java.io.IOException;
import java.nio.file.FileVisitResult;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;
public class MultiThreadedFileTreeWalk {
private static class RecursiveWalk extends RecursiveAction {
private static final long serialVersionUID = 6913234076030245489L;
private final Path dir;
public RecursiveWalk(Path dir) {
this.dir = dir;
}
@Override
protected void compute() {
final List<RecursiveWalk> walks = new ArrayList<>();
try {
Files.walkFileTree(dir, new SimpleFileVisitor<Path>() {
@Override
public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
if (!dir.equals(RecursiveWalk.this.dir)) {
RecursiveWalk w = new RecursiveWalk(dir);
w.fork();
walks.add(w);
return FileVisitResult.SKIP_SUBTREE;
} else {
return FileVisitResult.CONTINUE;
}
}
@Override
public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
System.out.println(file + "\t" + Thread.currentThread());
return FileVisitResult.CONTINUE;
}
});
} catch (IOException e) {
e.printStackTrace();
}
for (RecursiveWalk w : walks) {
w.join();
}
}
}
public static void main(String[] args) throws IOException {
RecursiveWalk w = new RecursiveWalk(Paths.get(".").toRealPath());
ForkJoinPool p = new ForkJoinPool();
p.invoke(w);
}
}
В этом примере каждый каталог проходит по отдельному потоку. Вот учебник для библиотеки Java 7 fork/join.
Это упражнение не такое короткое, как ответ Scala, и не такое Java-подобное, как ответ Java.
Идея состояла в том, чтобы начать параллельные прогулки с чем-то вроде потока на устройство.
Обходчики находятся в потоках ForkJoinPool, поэтому, когда они запускают будущее для каждого теста пути, это разветвленные задачи в пуле. Тест каталога использует управляемую блокировку, когда он читает каталог и ищет файлы.
Результат возвращается путем выполнения обещания в зависимости от будущего пути тестирования. (Здесь нет механизма для обнаружения завершения с пустыми руками.)
Более интересный тест включал бы чтение zip-файлов, поскольку распаковка потребляет некоторый процессор.
Интересно, сделает ли paulp что-нибудь умное с глубоким списком?
import util._
import collection.JavaConverters._
import concurrent.{ TimeoutException => Timeout, _ }
import concurrent.duration._
import ExecutionContext.Implicits._
import java.io.IOException
import java.nio.file.{ FileVisitResult => Result, _ }
import Result.{ CONTINUE => Go, SKIP_SUBTREE => Prune, TERMINATE => Stop }
import java.nio.file.attribute.{ BasicFileAttributes => BFA }
object Test extends App {
val fileSystem = FileSystems.getDefault
val starts = (if (args.nonEmpty) args.toList else mounts) map (s => (fileSystem getPath s))
val p = Promise[(Path, BFA)]
def pathTest(path: Path, attrs: BFA) =
if (attrs.isDirectory ) {
val entries = blocking {
val res = Files newDirectoryStream path
try res.asScala.toList finally res.close()
}
List("hello","world") forall (n => entries exists (_.getFileName.toString == n))
} else {
path.getFileName.toString == "enough"
}
def visitor(root: Path) = new SimpleFileVisitor[Path] {
def stopOrGo = if (p.isCompleted) Stop else Go
def visiting(path: Path, attrs: BFA) = {
future { pathTest(path, attrs) } onComplete {
case Success(true) => p trySuccess (path, attrs)
case Failure(e) => p tryFailure e
case _ =>
}
stopOrGo
}
override def preVisitDirectory(dir: Path, attrs: BFA) = (
if ((starts contains dir) && dir != root) Prune
else visiting(dir, attrs)
)
override def postVisitDirectory(dir: Path, e: IOException) = {
if (e != null) p tryFailure e
stopOrGo
}
override def visitFile(file: Path, attrs: BFA) = visiting(file, attrs)
}
//def walk(p: Path): Path = Files walkFileTree (p, Set().asJava, 10, visitor(p))
def walk(p: Path): Path = Files walkFileTree (p, visitor(p))
def show(store: FileStore) = {
val ttl = store.getTotalSpace / 1024
val used = (store.getTotalSpace - store.getUnallocatedSpace) / 1024
val avail = store.getUsableSpace / 1024
Console println f"$store%-40s $ttl%12d $used%12d $avail%12d"
store
}
def mounts = {
val devs = for {
store <- fileSystem.getFileStores.asScala
if store.name startsWith "/dev/"
if List("ext4","fuseblk") contains store.`type`
} yield show(store)
val devstr = """(\S+) \((.*)\)""".r
(devs.toList map (_.toString match {
case devstr(name, dev) if devs.toList exists (_.name == dev) => Some(name)
case s => Console println s"Bad dev str '$s', skipping" ; None
})).flatten
}
starts foreach (f => future (walk(f)))
Try (Await result (p.future, 20.seconds)) match {
case Success((name, attrs)) => Console println s"Result: ${if (attrs.isDirectory) "dir" else "file"} $name"
case Failure(e: Timeout) => Console println s"No result: timed out."
case Failure(t) => Console println s"No result: $t."
}
}