Parallele Dateiverarbeitung: ForkJoin, Parallel Streams

1. Parallele Streams (parallelStream): einfach und bequem

Wenn Sie bereits mit der Stream-API gearbeitet haben, wissen Sie, wie bequem sich Kollektionen filtern, transformieren und sammeln lassen. Ein wichtiger Bonus: Jeden Stream kann man mit nur einer Zeile parallel machen – parallel() einschalten, und die Elemente der Kollektion werden gleichzeitig verarbeitet.

Das ist besonders nützlich für unabhängige Operationen über eine Menge von Dateien: Zeilen zählen, einen Substring finden, kopieren, komprimieren usw.

Beispiel: paralleles Zählen der Zeilen in allen Dateien eines Verzeichnisses

Variante 1: sequentiell

import java.nio.file.*;
import java.io.IOException;
import java.util.List;

public class LogLineCounter {
    public static void main(String[] args) throws IOException {
        Path logDir = Paths.get("logs");
        long totalLines = 0;
        try (DirectoryStream<Path> stream = Files.newDirectoryStream(logDir, "*.log")) {
            for (Path file : stream) {
                long lines = Files.lines(file).count();
                totalLines += lines;
            }
        }
        System.out.println("Gesamtzahl der Zeilen in allen Logs: " + totalLines);
    }
}

Kommentar: Alles geschieht nacheinander, eine Datei nach der anderen. Wenn es viele und große Dateien sind, muss man lange warten.

Variante 2: parallel!

import java.nio.file.*;
import java.io.IOException;
import java.util.stream.Stream;

public class LogLineCounterParallel {
    public static void main(String[] args) throws IOException {
        Path logDir = Paths.get("logs");
        try (Stream<Path> files = Files.list(logDir)) {
            long totalLines = files
                .filter(path -> path.toString().endsWith(".log"))
                .parallel() // Das ist die ganze Magie!
                .mapToLong(file -> {
                    try (Stream<String> lines = Files.lines(file)) {
                        return lines.count();
                    } catch (IOException e) {
                        e.printStackTrace();
                        return 0L;
                    }
                })
                .sum();
            System.out.println("Gesamtzahl der Zeilen in allen Logs: " + totalLines);
        }
    }
}

Kommentar: Die Schlüsselzeile – .parallel(). Auf einem Mehrkernprozessor läuft das Programm in der Regel deutlich schneller.

Wie funktioniert das?

• parallel() verwandelt einen normalen Stream in einen parallelen. Unter der Haube wird der gemeinsame ForkJoinPool verwendet (standardmäßig so viele Threads wie Kerne).
• Jede Datei wird unabhängig verarbeitet, die Ergebnisse werden über Terminal-Operationen aggregiert (z. B. sum()).
• Gibt es nur wenige Dateien, kann es keine Beschleunigung geben; bei Hunderten ist der Gewinn gewöhnlich deutlich.

Wichtig!

• Parallele Streams beschleunigen die I/O-Operationen an sich nicht; sie ermöglichen lediglich, mehrere Operationen gleichzeitig auszuführen. Auf schnellen Datenträgern (SSD) hilft das, auf langsamen (HDD) stößt man schnell auf den Flaschenhals der Platte.

2. ForkJoinPool: „Teile und herrsche“ in der Praxis

ForkJoin ist ein Framework für parallele Berechnungen nach dem Prinzip „Teile und herrsche“: Eine große Aufgabe wird in Teilaufgaben zerlegt, parallel ausgeführt und die Ergebnisse zusammengeführt. Gesteuert wird das von einem speziellen Pool – dem ForkJoinPool. Er wird „hinter den Kulissen“ auch von parallelen Streams verwendet, lässt sich aber für mehr Flexibilität auch direkt steuern.

Dieses Modell eignet sich besonders für rekursive Strukturen (Verzeichnisbäume), große Datenarrays und Aufgaben, die sich leicht in unabhängige Teile zerlegen lassen.

Beispiel: rekursive Suche im Verzeichnisbaum

Wir finden alle ".txt"-Dateien (einschließlich verschachtelter Ordner) und zählen die Gesamtzahl der Zeilen.

import java.nio.file.*;
import java.util.concurrent.*;
import java.util.*;
import java.io.IOException;
import java.util.stream.Collectors;

public class FolderLineCounter extends RecursiveTask<Long> {
    private final Path dir;

    public FolderLineCounter(Path dir) {
        this.dir = dir;
    }

    @Override
    protected Long compute() {
        List<FolderLineCounter> subTasks = new ArrayList<>();
        long lines = 0;
        try (DirectoryStream<Path> stream = Files.newDirectoryStream(dir)) {
            for (Path entry : stream) {
                if (Files.isDirectory(entry)) {
                    FolderLineCounter task = new FolderLineCounter(entry);
                    task.fork(); // Unteraufgabe starten
                    subTasks.add(task);
                } else if (entry.toString().endsWith(".txt")) {
                    try (Stream<String> fileLines = Files.lines(entry)) {
                        lines += fileLines.count();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        // Ergebnisse aus den Unteraufgaben einsammeln
        for (FolderLineCounter task : subTasks) {
            lines += task.join();
        }
        return lines;
    }

    public static void main(String[] args) {
        Path root = Paths.get("big_folder");
        ForkJoinPool pool = new ForkJoinPool();
        FolderLineCounter counter = new FolderLineCounter(root);
        long totalLines = pool.invoke(counter);
        System.out.println("Gesamtzahl der Zeilen in allen .txt-Dateien: " + totalLines);
    }
}

Was passiert hier:

• Für jeden Ordner wird eine eigene Aufgabe erstellt (FolderLineCounter), für Unterverzeichnisse – eigene Unteraufgaben (fork()).
• Dateien werden vor Ort gezählt, die Ergebnisse nach dem join() aller Unteraufgaben aufsummiert.

Worin liegt der Vorteil ForkJoin?

• Funktioniert effizient mit großen Hierarchien (Verzeichnisbäume).
• Lastet die CPU-Kerne maximal aus.
• Erlaubt eine genaue Kontrolle der Parallelisierung und der Aufgabenabgrenzung.

3. Praktische Anwendungsfälle

Massenhafte Dateiverarbeitung

Zum Beispiel sollen Tausende Fotos in einen Sicherungsordner kopiert werden.

import java.nio.file.*;
import java.util.List;
import java.util.stream.Collectors;

public class ParallelFileCopier {
    public static void main(String[] args) throws Exception {
        Path sourceDir = Paths.get("photos");
        Path destDir = Paths.get("photos_backup");
        Files.createDirectories(destDir);

        List<Path> files = Files.list(sourceDir)
                .filter(Files::isRegularFile)
                .collect(Collectors.toList());

        files.parallelStream().forEach(file -> {
            try {
                Path destFile = destDir.resolve(file.getFileName());
                Files.copy(file, destFile, StandardCopyOption.REPLACE_EXISTING);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });

        System.out.println("Alle Dateien wurden kopiert!");
    }
}

Kommentar: Jede Datei wird in einem eigenen Thread kopiert. Bei einer großen Anzahl von Dateien ist die Beschleunigung spürbar.

Paralleles Komprimieren/Entpacken

Analog lassen sich Komprimierung, Hash-Neuberechnung, Bildformat-Konvertierung usw. über parallelStream() oder einen eigenen ForkJoinPool parallelisieren.

4. Wichtige Hinweise und Einschränkungen

• I/O-Operationen profitieren nicht immer von Parallelität. Wenn das Laufwerk oder das Netzwerk der Flaschenhals ist, erhöht eine Hundertschaft paralleler Aufgaben nur die Ressourcenkonkurrenz.
• Starten Sie nicht zu viele Threads. Standardmäßig verwenden parallele Streams den gemeinsamen ForkJoinPool.commonPool() mit einem Parallelitätsgrad ≈ der Zahl der Kerne. Dies lässt sich über die Eigenschaft "java.util.concurrent.ForkJoinPool.common.parallelism" anpassen – tun Sie das jedoch mit Bedacht.
• Vergessen Sie die Synchronisation nicht. Wenn mehrere Threads in dieselbe Datei/ dasselbe Objekt schreiben – verwenden Sie Synchronisation und Queues; für unabhängige Dateien ist Synchronisation nicht nötig.

5. Kurze Einführung in FileChannel und positionsbasierten Zugriff

Für fortgeschrittene Szenarien (z. B. paralleles Lesen verschiedener Teile einer großen Datei) verwenden Sie java.nio.channels.FileChannel, das positionsbasiertes Lesen/Schreiben unterstützt.

Beispiel: Lesen verschiedener Dateibereiche in unterschiedlichen Threads

import java.nio.channels.FileChannel;
import java.nio.file.*;
import java.nio.ByteBuffer;

public class FileChunkReader implements Runnable {
    private final Path path;
    private final long position;
    private final int size;

    public FileChunkReader(Path path, long position, int size) {
        this.path = path;
        this.position = position;
        this.size = size;
    }

    @Override
    public void run() {
        try (FileChannel channel = FileChannel.open(path, StandardOpenOption.READ)) {
            ByteBuffer buffer = ByteBuffer.allocate(size);
            channel.read(buffer, position);
            // Datenverarbeitung
            System.out.println("Gelesen: " + buffer.position() + " Byte ab Position " + position);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

Kommentar: Starten Sie mehrere solcher Aufgaben – jede liest ihren eigenen Bereich. Aber Vorsicht: Nicht alle Laufwerke und Dateisysteme mögen starke parallele Last.

6. Typische Fehler bei der parallelen Dateiverarbeitung

Fehler Nr. 1: Paralleles Schreiben in eine Datei ohne Synchronisation. Daten werden vermischt und beschädigt. Verwenden Sie Queues, Pufferung und synchronisiertes Schreiben.

Fehler Nr. 2: Zu viel Parallelität. Parallele Streams verursachen auf schwacher Hardware bzw. bei kleinen Dateien Overhead und können die Ausführung verlangsamen.

Fehler Nr. 3: Ignorieren von I/O-Fehlern. In parallelen Streams gehen Ausnahmen leicht „verloren“ – behandeln Sie sie innerhalb der Lambdas, protokollieren Sie sie und berücksichtigen Sie Ausfälle.

Fehler Nr. 4: Nicht geschlossene Ressourcen. Verwenden Sie stets try-with-resources für Streams/Kanäle, sonst drohen Lecks und seltsame Fehler.

Fehler Nr. 5: „Magie“ von parallel() erwarten. Parallelität beschleunigt nur bei ausreichender Arbeitslast und verfügbaren Ressourcen (CPU, schnelles Laufwerk). Der Aufruf parallel() ist für sich genommen kein Allheilmittel.