Stream API vs Flux: różnice dla programistów Java | Przewodnik

# Stream API vs Flux: różnice architektoniczne i praktyczne zastosowanie

Zrozumienie różnic między Stream API a Flux jest kluczowe dla programistów Java pracujących z systemami asynchronicznymi. Pomimo podobieństwa metod (filter, map, limit), te narzędzia rozwiązują zasadniczo różne zadania. Stream API służy do przetwarzania skończonych kolekcji w trybie synchronicznym, natomiast Flux z Project Reactor został stworzony do reaktywnego przetwarzania danych strumieniowych z obsługą backpressure i asynchroniczności. Ten artykuł omawia kluczowe różnice architektoniczne i pomaga wybrać odpowiednie narzędzie do konkretnego zadania.

Podstawy Stream API: synchroniczne przetwarzanie kolekcji

Stream API to wbudowany mechanizm Java do deklaratywnego przetwarzania kolekcji. Jego kluczowa cecha: praca z skończonymi zbiorami danych w ramach jednego wątku wykonania. Po wywołaniu .stream() tworzony jest sekwencyjny strumień, przetwarzający elementy w tym samym wątku, w którym został zainicjowany (np. main).

Ważne jest pamiętanie o podziale operacji:

• Operacje pośrednie (filter, map, flatMap, limit, skip) — tworzą łańcuch transformacji bez natychmiastowego wykonania. Zwracają nowy strumień i mogą być łączone w nieograniczonej liczbie.
• Operacje terminalne (collect, forEach, reduce) — uruchamiają wykonanie całego łańcucha i zamykają strumień. Po ich wywołaniu ponowne użycie strumienia jest niemożliwe.

Rozważmy przykład filtracji i transformacji danych:

List<Integer> numbers = List.of(1, 2, 3, 4, 5);
numbers.stream()
    .filter(n -> n % 2 == 0)
    .map(n -> n * 2)
    .forEach(System.out::println); // 4, 8

Tutaj:

• filter zostawia liczby parzyste
• map podwaja każdy element
• forEach uruchamia wykonanie i wyświetla wynik

Krytyczne ograniczenie Stream API to jednorazowe użycie. Próba ponownego wywołania operacji terminalnej spowoduje IllegalStateException. To fundamentalna różnica w porównaniu do strumieni reaktywnych.

Flux: reaktywne przetwarzanie danych strumieniowych

Flux z Project Reactor implementuje specyfikację Reactive Streams, służącą do asynchronicznego przetwarzania nieograniczonych strumieni danych. W odróżnieniu od Stream API, Flux obsługuje:

• Wykonanie asynchroniczne
• Backpressure (regulację prędkości konsumowania danych)
• Przetwarzanie błędów za pomocą operatorów reaktywnych
• Subskrypcję zdarzeń w dowolnym wątku

Kluczowe różnice architektoniczne:

• Operacja terminalna — .subscribe() jest obowiązkowa do uruchomienia przetwarzania. Bez niej Flux pozostaje „zimny” i nie generuje danych.
• Leniwą inicjalizacja — dane są generowane dopiero po subskrypcji.
• Obsługę nieograniczonych strumieni — Flux może pracować z nieskończonymi źródłami (np. dane z czujników w czasie rzeczywistym).

Przykład równoważnej operacji w Flux:

Flux.just(1, 2, 3, 4, 5)
    .filter(n -> n % 2 == 0)
    .map(n -> n * 2)
    .subscribe(System.out::println); // 4, 8

Tutaj .subscribe() nie tylko uruchamia przetwarzanie, ale zarządza cyklem życia strumienia. Flux może obsługiwać błędy za pomocą .onErrorResume() i regulować przepływ za pomocą .limitRate().

Krytyczne różnice: tabela porównawcza

Dla lepszej przejrzystości podsumujmy kluczowe różnice w strukturyzowanej liście:

• Model wykonania

- Stream API: synchroniczne, blokujące przetwarzanie

- Flux: asynchroniczne, nieblokujące przetwarzanie z obsługą backpressure

• Typ danych

- Stream API: działa tylko ze skończonymi kolekcjami

- Flux: obsługuje skończone i nieskończone strumienie

• Przetwarzanie błędów

- Stream API: wyjątki są rzucane bezpośrednio (wymagają try/catch)

- Flux: wbudowane przetwarzanie błędów za pomocą operatorów (onErrorResume, retry)

• Ponowne użycie

- Stream API: jednorazowe użycie po operacji terminalnej

- Flux: wielokrotna subskrypcja na jednym strumieniu (przy odpowiedniej konfiguracji)

• Równoległość

- Stream API: .parallelStream() rozdziela zadania po rdzeniach CPU

- Flux: .publishOn() i .subscribeOn() zarządzają kontekstem wykonania

Praktyczny przypadek: transformacja danych w rzeczywistych scenariuszach

Rozważmy zadanie aktualizacji danych w kolekcji. Mamy HashMap z pracownikami i ich poziomami. Wymagane:

• Dla pracowników Junior zmienić poziom na Middle
• Dodać oznaczenie „(awansowany)"
• Zwrócić zaktualizowaną strukturę

Realizacja za pomocą Stream API:

Map<String, String> updatedEmployees = employees.entrySet().stream()
    .map(entry -> {
        String grade = entry.getValue();
        if ("Junior".equals(grade)) {
            return Map.entry(entry.getKey(), "Middle (increased)");
        }
        return entry;
    })
    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

Ten sam scenariusz za pomocą Flux:

Map<String, String> updatedEmployees = Flux.fromStream(employees.entrySet().stream())
    .map(entry -> {
        if ("Junior".equals(entry.getValue())) {
            return Map.entry(entry.getKey(), "Middle (increased)");
        }
        return entry;
    })
    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))
    .block();

Kluczowe różnice w realizacji:

• Flux wymaga jawnego .block() do pobrania wyniku (narusza podejście reaktywne)
• Stream API blokuje wątek wykonania do zakończenia operacji
• W systemach reaktywnych preferowany jest nieblokujący sposób za pomocą .subscribe()

Co ważne: kluczowe wnioski

• Stream API — optymalny do synchronicznego przetwarzania skończonych kolekcji w ramach jednego wątku. Używaj go do operacji ETL, transformacji danych w logice biznesowej.
• Flux — rozwiązuje zadania asynchronicznego przetwarzania danych strumieniowych. Stosuj przy pracy z żądaniami sieciowymi, czujnikami, szynami zdarzeń.
• Backpressure — krytyczna funkcja Flux, nieobecna w Stream API. Zapewnia stabilność systemu przy wysokiej obciążeniu.
• Błędy wykonania — w Flux przetwarzane reaktywnie, w Stream API wymagają tradycyjnego try/catch.
• Wydajność — Stream API jest efektywniejszy dla krótkich operacji synchronicznych, Flux — dla długich procesów asynchronicznych.

— Editorial Team