Równoległe przejścia systemu plików: Files.walk + parallel() i ForkJoin

1. Problem: jak efektywnie przetworzyć wiele plików w katalogu

We współczesnych aplikacjach często pojawia się zadanie: przetworzyć dużą liczbę plików w folderze i jego podkatalogach. Na przykład:

• Zliczyć całkowitą liczbę linii we wszystkich ".java"-plikach projektu.
• Znaleźć wszystkie pliki zmienione w ostatnim miesiącu.
• Skopiować lub usunąć pliki według określonego kryterium.

Jeśli plików jest mało, wystarczy zwykła pętla. Ale przy tysiącach i dziesiątkach tysięcy, zwłaszcza gdy dla każdego pliku wykonywana jest „ciężka” operacja (odczyt, parsowanie, analiza), czas rośnie znacząco.

Pytanie: jak przyspieszyć przetwarzanie dużej liczby plików?
Odpowiedź: użyć równoległości — przetwarzać pliki jednocześnie w wielu wątkach.

2. Narzędzia do przechodzenia po systemie plików

Files.walk()

W Java 8+ pojawił się wygodny sposób przechodzenia drzewa katalogów — metoda Files.walk() z pakietu java.nio.file. Zwraca strumień Stream<Path> — wszystkie pliki i foldery, zaczynając od wskazanego katalogu.

Przykład:

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

Path start = Paths.get("src");
try (Stream<Path> stream = Files.walk(start)) {
    stream.forEach(System.out::println);
}

• Files.walk(start) — zwraca strumień wszystkich plików i folderów, włącznie z podkatalogami.
• Można wskazać maksymalną głębokość przejścia: Files.walk(start, 3).

Files.find()

Jeśli trzeba od razu filtrować po kryterium (np. tylko ".java"-pliki), użyj Files.find():

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

Path start = Paths.get("src");

try (Stream<Path> stream = Files.find(
        start,
        Integer.MAX_VALUE,
        (path, attr) -> path.toString().endsWith(".java"))) {
    stream.forEach(System.out::println);
}

• Files.find() przyjmuje filtr (BiPredicate<Path, BasicFileAttributes>), który otrzymuje ścieżkę i atrybuty pliku.

3. Przetwarzanie równoległe: parallel() i ForkJoinPool

Strumienie równoległe: .parallel()

Każdy Stream ma metodę parallel(). Po jej wywołaniu przetwarzanie elementów odbywa się w wielu wątkach.

Files.walk(start)
    .parallel()
    .forEach(path -> processFile(path));

Każdy plik będzie przetwarzany równolegle (w miarę możliwości), co jest szczególnie efektywne przy „ciężkich” operacjach: odczyt, parsowanie, obliczenia.

Jak to działa pod spodem? ForkJoinPool

Strumienie równoległe używają wspólnej puli wątków — ForkJoinPool.commonPool(). To „inteligentna” pula, która rozdziela zadania między wątki.

• Domyślnie liczba wątków = liczbie dostępnych procesorów: Runtime.getRuntime().availableProcessors().
• Model równoległy „fork/join” dobrze nadaje się do niezależnych zadań — jak przetwarzanie pojedynczych plików.

Kiedy używać .parallel()?

• Gdy przetwarzanie każdego pliku jest niezależne od pozostałych.
• Gdy operacja jest „ciężka” (obciąża CPU lub długo czeka na IO).
• Gdy plików jest dużo (setki, tysiące).

Nie należy używać strumieni równoległych:

• Jeśli plików jest mało (narzut na równoległość może przewyższyć zysk).
• Jeśli wymagany jest ścisły porządek lub istnieją zależności między elementami.

4. Alternatywy i dostrajanie równoległości

Kiedy lepiej użyć ExecutorService?

Strumienie równoległe są dobre w prostych przypadkach. Ale jeśli trzeba:

• Kontrolować dokładną liczbę wątków (dla zadań IO-bound korzystne jest mieć więcej wątków niż rdzeni).
• Zarządzać kolejkami, anulowaniem, ponownymi próbami, obsługą błędów.
• Budować bardziej złożone potoki zadań.

Wtedy użyj ExecutorService:

import java.nio.file.*;
import java.util.concurrent.*;

ExecutorService executor = Executors.newFixedThreadPool(8);
Files.walk(start)
    .filter(Files::isRegularFile)
    .forEach(path -> executor.submit(() -> processFile(path)));
executor.shutdown();

Dostrajanie ForkJoinPool

Domyślnie wspólna pula używa liczby wątków równej liczbie procesorów. Można to zmienić przez właściwość systemową (zanim po raz pierwszy użyjesz strumieni równoległych):

System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "16");

• Po wywołaniu wszystkie strumienie równoległe będą używać do 16 wątków.

CPU-bound vs IO-bound zadania

• CPU-bound: intensywnie obciążają procesor (obliczenia, parsowanie, kompresja). Liczba wątków ≈ liczbie rdzeni.
• IO-bound: dużo oczekiwania na dysk/sieć. Często opłaca się mieć więcej wątków niż rdzeni.

Strumienie równoległe nie zawsze są optymalne dla zadań IO-bound — częściej wygrywa własny ExecutorService z większą pulą.

5. Przykład: równoległe wyszukiwanie i przetwarzanie plików

Zliczmy łączną liczbę linii we wszystkich ".java"-plikach projektu, używając równoległego przejścia.

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

public class LineCounter {
    public static void main(String[] args) throws IOException {
        Path start = Paths.get("src");

        long totalLines = Files.walk(start)
            .parallel() // przetwarzanie równoległe!
            .filter(p -> p.toString().endsWith(".java"))
            .mapToLong(LineCounter::countLines)
            .sum();

        System.out.println("Łączna liczba linii kodu: " + totalLines);
    }

    // Metoda zliczania linii w pliku
    private static long countLines(Path path) {
        try (Stream<String> lines = Files.lines(path)) {
            return lines.count();
        } catch (IOException e) {
            System.err.println("Błąd odczytu pliku: " + path);
            return 0;
        }
    }
}

Co się dzieje:

• Files.walk(start) — przejście wszystkich ścieżek.
• parallel() — włączamy przetwarzanie równoległe.
• filter(...) — zostawiamy tylko ".java"-pliki.
• mapToLong(...) — liczymy linie w każdym pliku.
• sum() — sumujemy wynik.

Zalety: wykorzystuje się wiele wątków, a kod pozostaje zwięzły.

6. Ważne niuanse i typowe błędy

• Nie wszystkie zadania przyspieszają dzięki równoległości. Dla niewielkich zbiorów plików lub szybkich operacji narzut może spowolnić program.
• Zamykaj zasoby. Przy pracy z plikami używaj try-with-resources — dzięki temu deskryptory nie „uciekną”. Na przykład Files.lines(path) w try(...).
• Zagnieżdżona równoległość. Uruchamianie strumieni równoległych wewnątrz innych zadań równoległych (nested parallelism) rzadko jest efektywne i może prowadzić do degradacji wydajności.
• Efekty uboczne. Unikaj zapisu do wspólnych struktur/plików bez synchronizacji. Preferuj „czyste” operacje na elementach.

7. Schemat: jak działa równoległe przejście po plikach

flowchart TD
    A["Files.walk(start)"] --> B["Stream<Path>"]
    B --> C{".parallel()?"}
    C -- Nie --> D[Zwykły forEach]
    C -- Tak --> E["Równoległy forEach (ForkJoinPool)"]
    E --> F[Przetwarzanie plików w wielu wątkach]

8. Typowe błędy przy równoległym przetwarzaniu plików

Błąd nr 1: Używanie strumieni równoległych do małych zadań — narzut jest większy niż zysk.

Błąd nr 2: Oczekiwanie, że strumienie równoległe przyspieszą zadania IO-bound tak samo jak CPU-bound. Dla IO częściej potrzebny jest ExecutorService z większą pulą.

Błąd nr 3: Nieobsłużone wyjątki w lambdach — bez obsługi IOException strumień może się przerwać, a wynik okazać się niepełny.

Błąd nr 4: Wyścigi przy zapisie do wspólnych zmiennych lub plików — synchronizuj dostęp albo unikaj efektów ubocznych.

Błąd nr 5: Zapomniano zamknąć zasoby — używaj try-with-resources dla wszystkich operacji na plikach.

Błąd nr 6: Próba zmiany ForkJoinPool.commonPool() po pierwszym użyciu — konfigurację przez System.setProperty(...) trzeba wykonać wcześniej.

Błąd nr 7: Używanie strumieni równoległych wewnątrz innych strumieni równoległych — często prowadzi do degradacji wydajności.