Strumienie równoległe: składnia i zastosowanie

1. Przypominamy Stream API

Znacie już Stream API — to wygodny sposób pracy z kolekcjami, który pozwala pisać zwięzły i czytelny kod do przetwarzania danych: filtrowania, sortowania, zliczania itd.

Oto klasyczny przykład:

List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);

int sum = numbers.stream()
    .filter(n -> n % 2 == 0)
    .mapToInt(n -> n)
    .sum();

System.out.println(sum); // 6 (2 + 4)

W tym przykładzie kolekcja zamienia się w strumień (stream()), z którego wybierane są tylko liczby parzyste, następnie są przekształcane do int, a wynik sumowany wywołaniem sum().

Stream API skraca i uczytelnia kod: zamiast opisywać krok po kroku, jak dokładnie wszystko się odbywa, mówisz po prostu, co chcesz otrzymać. Dodatkowo w razie potrzeby łatwo przełączysz się na przetwarzanie równoległe — jednym wierszem.

2. Strumienie równoległe: składnia i zasada działania

Jak zrobić strumień równoległy?

To proste: zamiast stream() użyj parallelStream(). Albo wywołaj metodę .parallel() dla istniejącego strumienia.

List<Integer> numbers = ...;

int sum = numbers.parallelStream()
    .filter(n -> n % 2 == 0)
    .mapToInt(n -> n)
    .sum();

Albo tak:

numbers.stream()
    .parallel() // zamieniamy na strumień równoległy
    .filter(...)
    .map(...)
    .sum();

Co dzieje się pod maską?

• Kolekcja jest automatycznie dzielona na części.
• Każda część jest przetwarzana w osobnym wątku (używany jest ForkJoinPool — specjalna pula wątków).
• Wyniki są łączone w wartość końcową.

To znaczy, jeśli masz wielordzeniowy procesor, przetwarzanie naprawdę odbywa się równolegle — na przykład filtrowanie i sumowanie mogą działać jednocześnie na kilku rdzeniach.

Gdzie to jest szczególnie przydatne?

• Przetwarzanie dużych kolekcji (dziesiątki tysięcy elementów i więcej).
• Złożone obliczenia dla każdego elementu.
• Brak konieczności zachowania ścisłej kolejności przetwarzania.

Przykład: porównanie strumienia sekwencyjnego i równoległego

Spójrzmy na prosty przykład z przetwarzaniem dużej listy.

import java.util.*;
import java.util.stream.*;

public class ParallelStreamDemo {
    public static void main(String[] args) {
        List<Integer> numbers = IntStream.rangeClosed(1, 10_000_000)
                                         .boxed()
                                         .collect(Collectors.toList());

        // Strumień sekwencyjny
        long time1 = System.currentTimeMillis();
        long count1 = numbers.stream()
            .filter(n -> n % 2 == 0)
            .count();
        long time2 = System.currentTimeMillis();
        System.out.println("Sekwencyjnie: " + (time2 - time1) + " ms, parzystych: " + count1);

        // Strumień równoległy
        long time3 = System.currentTimeMillis();
        long count2 = numbers.parallelStream()
            .filter(n -> n % 2 == 0)
            .count();
        long time4 = System.currentTimeMillis();
        System.out.println("Równolegle: " + (time4 - time3) + " ms, parzystych: " + count2);
    }
}

Wypróbuj ten kod na swoim komputerze — najprawdopodobniej strumień równoległy przetworzy kolekcję szybciej (zwłaszcza na procesorze wielordzeniowym). Ale nie zawsze! O niuansach — poniżej.

3. Jak to działa: ForkJoinPool i automatyczny podział

Strumienie równoległe używają pod spodem ForkJoinPool.commonPool(), który automatycznie zarządza liczbą wątków (zwykle — według liczby dostępnych rdzeni procesora).

Schematycznie:

+-----------------------------+
|      Twoja kolekcja         |
+-----------------------------+
| 1  | 2  | 3  | ... | 10 mln |
+----+----+----+-----+--------+
   |    |    |           |
   v    v    v           v
[Wątek1][Wątek2]...[WątekN]
   |    |    |           |
   +----+----+-----------+
        |
   [Łączenie wyniku]

Każdy wątek przetwarza swoją część, potem wyniki są łączone.

4. Ograniczenia i pułapki

Strumienie równoległe to nie magiczny przycisk „przyspiesz wszystko”. Czasem nawet spowalniają wykonanie!

Kiedy równoległość się nie opłaca:

• Kolekcja jest mała (do ~1000 elementów).
• Operacja na każdym elemencie jest bardzo szybka (na przykład po prostu n * 2).
• Istotny jest ścisły porządek przetwarzania (na przykład dla sekwencyjnego zapisu do pliku).

Dlaczego? Tworzenie i synchronizacja wątków też kosztują czas. Jeśli samo zadanie jest „drobne”, narzut może przewyższyć zysk z równoległości.

Efekty uboczne — wróg równoległości

Jeśli operacje wewnątrz strumienia zmieniają zmienne zewnętrzne, bądź ostrożny!

Zły przykład:

List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
int[] sum = {0};

numbers.parallelStream().forEach(n -> sum[0] += n);

System.out.println(sum[0]); // ??? (spodziewasz się 15, a możesz dostać cokolwiek)

Dlaczego? Bo kilka wątków jednocześnie modyfikuje jedną zmienną — powstaje race condition (warunek wyścigu). Wartość końcowa może być niepoprawna.

Poprawny sposób — używać metod strumienia, które zwracają wynik:

int sum = numbers.parallelStream().mapToInt(n -> n).sum();

Nie wszystkie kolekcje równie dobrze nadają się do równoległości

Niektóre kolekcje (na przykład zwykły ArrayList) dobrze dzielą się na części. Natomiast LinkedList lub strumień o nieskończonej liczbie elementów (na przykład Stream.generate(...)) — już nie.

5. Praktyka: porównanie wydajności

Przykład: znajdowanie liczby maksymalnej

import java.util.*;
import java.util.stream.*;

public class ParallelMaxDemo {
    public static void main(String[] args) {
        List<Integer> numbers = IntStream.rangeClosed(1, 30_000_000)
                                         .boxed()
                                         .collect(Collectors.toList());

        // Sekwencyjnie
        long t1 = System.currentTimeMillis();
        int max1 = numbers.stream().max(Integer::compareTo).get();
        long t2 = System.currentTimeMillis();
        System.out.println("Sekwencyjnie: " + (t2 - t1) + " ms, max = " + max1);

        // Równolegle
        long t3 = System.currentTimeMillis();
        int max2 = numbers.parallelStream().max(Integer::compareTo).get();
        long t4 = System.currentTimeMillis();
        System.out.println("Równolegle: " + (t4 - t3) + " ms, max = " + max2);
    }
}

Co zobaczymy? Na nowoczesnych wielordzeniowych procesorach strumień równoległy jest zwykle szybszy. Ale jeśli zamienisz 30_000_000 na 1000, różnicy nie będzie — a czasem równoległy bywa wolniejszy!

6. Przykłady użycia: filtrowanie, agregacja, sortowanie

Filtrowanie i zliczanie

List<String> names = Arrays.asList("Anya", "Boris", "Vasya", "Grisha", "Dasha", "Egor", "Zhenya");

long count = names.parallelStream()
    .filter(name -> name.length() == 4)
    .count();

System.out.println("Imion o długości 4: " + count);

Grupowanie

List<String> words = Arrays.asList("kot", "whale", "kot", "pies", "whale", "kot");

Map<String, Long> freq = words.parallelStream()
    .collect(Collectors.groupingBy(
        w -> w,
        Collectors.counting()
    ));

System.out.println(freq); // {pies=1, whale=2, kot=3}

Sortowanie (ale tutaj równoległość nie zawsze daje zysk!)

List<Integer> bigList = IntStream.rangeClosed(1, 5_000_000)
                                 .boxed()
                                 .collect(Collectors.toList());

long t1 = System.currentTimeMillis();
List<Integer> sorted = bigList.parallelStream()
    .sorted()
    .collect(Collectors.toList());
long t2 = System.currentTimeMillis();

System.out.println("Sortowanie równoległe: " + (t2 - t1) + " ms");

7. Ważne uwagi i zalecenia

Kiedy warto używać parallelStream()

• Kolekcja jest duża (dziesiątki tysięcy elementów i więcej).
• Operacja na elemencie jest „ciężka” (złożone obliczenia, praca z plikami/siecią).
• Brak zależności od kolejności elementów.
• Brak efektów ubocznych (nie są zmieniane zmienne zewnętrzne).

Kiedy NIE warto używać parallelStream()

• Kolekcja jest mała.
• Operacja jest szybka.
• Trzeba ściśle zachować porządek.
• Jest dostęp do współdzielonych zmiennych (rozważ kolekcje thread-safe albo inne podejścia).

Jak sprawdzić, ile wątków jest używanych?

Domyślnie — tyle, ile rdzeni procesora: Runtime.getRuntime().availableProcessors(). Można zmienić to zachowanie poprzez właściwość systemową:

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

Rób to tylko, jeśli rozumiesz konsekwencje — w przeciwnym razie możesz „zajechać” procesor i uzyskać spowolnienie.

8. Typowe błędy przy pracy ze strumieniami równoległymi

Błąd nr 1: efekty uboczne wewnątrz forEach
Wielu myśli: „Teraz równolegle wypełnię listę!”

List<Integer> result = new ArrayList<>();
IntStream.range(0, 1_000)
    .parallel()
    .forEach(result::add); // NIEBEZPIECZNE!
System.out.println(result.size()); // Wynik — losowy!

Dlaczego to źle? ArrayList nie jest bezpieczny wątkowo, a przy jednoczesnym dodawaniu z wielu wątków wynik jest nieprzewidywalny: mogą pojawić się braki, duplikaty, wyjątki.

Rozwiązanie: Używaj metod zbierających strumienia (collect), które same zapewniają bezpieczeństwo, albo specjalnych kolekcji.

List<Integer> result = IntStream.range(0, 1_000)
    .parallel()
    .boxed()
    .collect(Collectors.toList());

Błąd nr 2: oczekiwanie przyspieszenia przy małych zadaniach
Równoległość nie jest darmowa! Jeśli kolekcja jest mała, strumień równoległy może działać wolniej przez narzut planowania i synchronizacji.

Błąd nr 3: naruszenie porządku
Jeśli ważny jest porządek elementów (na przykład przy zapisie do pliku), nie używaj strumieni równoległych — porządek nie jest gwarantowany (albo będzie działać wolniej).

Błąd nr 4: używanie nieodpowiednich kolekcji
Niektóre kolekcje (na przykład LinkedList, niestandardowe struktury) słabo dzielą się na części — efektywność równoległości spada.

Błąd nr 5: ignorowanie bezpieczeństwa wątkowego przy zbieraniu wyników
Jeśli zbierasz wyniki ręcznie (na przykład dodajesz do listy), używaj kolekcji bezpiecznych wątkowo (CopyOnWriteArrayList, ConcurrentLinkedQueue) albo metod zbierających strumienia.