Какие плюсы и минусы параллельных стримов?

# Параллельные стримы (Parallel Streams)

Что это

Параллельный стрим — это способ обработки элементов коллекции в несколько потоков одновременно, вместо последовательной обработки. Используется для увеличения производительности на многоядерных процессорах.

// Последовательный стрим
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
Int sum = numbers.stream()
    .filter(n -> n % 2 == 0)
    .map(n -> n * 2)
    .reduce(0, Integer::sum);

// Параллельный стрим
Int sum = numbers.parallelStream()
    .filter(n -> n % 2 == 0)
    .map(n -> n * 2)
    .reduce(0, Integer::sum);

ПЛЮСЫ параллельных стримов

1. Автоматический параллелизм

// Без параллельных стримов
List<Integer> numbers = generateMillionNumbers();
int sum = 0;
for (int num : numbers) {
    if (num % 2 == 0) {
        sum += num * 2;
    }
}

// С параллельными стримами
int sum = numbers.parallelStream()
    .filter(n -> n % 2 == 0)
    .map(n -> n * 2)
    .reduce(0, Integer::sum);

// Система автоматически распределяет работу между потоками

2. Производительность на больших данных

// Бенчмарк: 100 миллионов элементов

public class StreamPerformance {
    public static void main(String[] args) {
        List<Integer> numbers = new ArrayList<>();
        for (int i = 0; i < 100_000_000; i++) {
            numbers.add(i);
        }
        
        // Последовательный стрим: ~5000ms
        long start = System.currentTimeMillis();
        long count = numbers.stream()
            .filter(n -> isPrime(n))
            .count();
        System.out.println("Sequential: " + (System.currentTimeMillis() - start) + "ms");
        
        // Параллельный стрим на 4-ядерном процессоре: ~1500ms
        start = System.currentTimeMillis();
        count = numbers.parallelStream()
            .filter(n -> isPrime(n))
            .count();
        System.out.println("Parallel: " + (System.currentTimeMillis() - start) + "ms");
        // Ускорение в 3-4 раза!
    }
    
    static boolean isPrime(int n) {
        if (n < 2) return false;
        for (int i = 2; i * i <= n; i++) {
            if (n % i == 0) return false;
        }
        return true;
    }
}

3. Простота использования

// Просто измените stream() на parallelStream()

// ДО:
List<Order> orders = getOrders();
List<OrderSummary> summaries = orders.stream()
    .map(this::enrichOrder)
    .map(this::calculateTotal)
    .collect(Collectors.toList());

// ПОСЛЕ:
List<OrderSummary> summaries = orders.parallelStream()
    .map(this::enrichOrder)
    .map(this::calculateTotal)
    .collect(Collectors.toList());
// Автоматически работает в параллель!

4. ForkJoinPool управляет потоками

// Java ForkJoinPool автоматически:
// - Распределяет задачи между потоками
// - Использует work-stealing алгоритм
// - Балансирует нагрузку

public class ForkJoinExample {
    public static void main(String[] args) {
        // parallelStream() использует общий ForkJoinPool
        ForkJoinPool commonPool = ForkJoinPool.commonPool();
        System.out.println("Available processors: " + 
            commonPool.getParallelism());
        // Обычно = кол-во ядер - 1
    }
}

5. Экономия памяти и кода

// Без параллельных стримов
List<User> users = getAllUsers();  // 1 млн
List<ReportRow> report = new ArrayList<>();
Object lock = new Object();

int numThreads = Runtime.getRuntime().availableProcessors();
ThreadPoolExecutor executor = 
    new ThreadPoolExecutor(numThreads, numThreads, 0L, TimeUnit.MILLISECONDS,
        new LinkedBlockingQueue<>());

int batchSize = users.size() / numThreads;
for (int i = 0; i < numThreads; i++) {
    int start = i * batchSize;
    int end = (i + 1) * batchSize;
    executor.execute(() -> {
        for (int j = start; j < end; j++) {
            synchronized(lock) {
                report.add(buildReportRow(users.get(j)));
            }
        }
    });
}
executor.shutdown();
executor.awaitTermination(1, TimeUnit.HOURS);

// С параллельными стримами
List<ReportRow> report = users.parallelStream()
    .map(this::buildReportRow)
    .collect(Collectors.toList());
// 1 строка вместо 20!

МИНУСЫ параллельных стримов

1. Overhead на координацию

// Если данных мало, параллелизм замедляет
public void demonstrateOverhead() {
    List<Integer> small = Arrays.asList(1, 2, 3, 4, 5);
    
    // Последовательный: 0.1ms
    small.stream()
        .map(n -> n * 2)
        .collect(Collectors.toList());
    
    // Параллельный: 5ms (из-за создания потоков и синхронизации)
    small.parallelStream()
        .map(n -> n * 2)
        .collect(Collectors.toList());
    // 50x МЕДЛЕННЕЕ!
}

// Правило: используйте параллельные стримы только если:
// - Коллекция >= 10,000 элементов
// - Операция сложная (долгая обработка)

2. Проблемы с потокобезопасностью

public class ThreadSafetyIssue {
    static int counter = 0;
    
    public static void main(String[] args) {
        List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
        
        // ❌ НЕПРАВИЛЬНО: counter не потокобезопасна
        numbers.parallelStream()
            .forEach(n -> counter++);  // Race condition!
        
        System.out.println(counter);  // Может быть < 5!
        
        // ✅ ПРАВИЛЬНО: используйте reduce
        int sum = numbers.parallelStream()
            .reduce(0, Integer::sum);
        System.out.println(sum);  // Всегда 15
    }
}

// Другие проблемы:
// - forEach() может дать неожиданный порядок
// - Обновление shared state из лямбды опасно
// - forEachOrdered() работает медленнее

3. Непредсказуемый порядок

public void orderingIssue() {
    List<String> data = Arrays.asList("A", "B", "C", "D", "E");
    
    // Последовательный: A, B, C, D, E
    data.stream()
        .forEach(System.out::println);
    
    // Параллельный: C, E, A, D, B (случайный порядок)
    data.parallelStream()
        .forEach(System.out::println);
    
    // Если порядок важен:
    data.parallelStream()
        .forEachOrdered(System.out::println);  // A, B, C, D, E
    // Но это медленнее!
}

4. Использование общего ForkJoinPool

public class ForkJoinPoolIssue {
    public static void main(String[] args) {
        // Проблема: все parallelStream() используют ОДИН общий пул
        
        // Поток 1: долгий parallelStream
        new Thread(() -> {
            longComputation();
        }).start();
        
        // Поток 2: быстрый parallelStream заблокирован!
        new Thread(() -> {
            // Это может ждать, пока Поток 1 освободит пул
            quickTask();
        }).start();
        
        // Решение: создать свой ForkJoinPool
        ForkJoinPool customPool = new ForkJoinPool(4);
        customPool.execute(() -> {
            List<Integer> data = generateData();
            long result = data.parallelStream()
                .filter(n -> expensiveOperation(n))
                .count();
        });
    }
}

5. Сложность отладки

public void debuggingIssue() {
    List<Integer> data = Arrays.asList(1, 2, 3, 4, 5);
    
    // ❌ Сложно отладить (потоки переплетаются)
    data.parallelStream()
        .map(n -> {
            System.out.println("Processing " + n);  // Хаотичный порядок
            return n * 2;
        })
        .collect(Collectors.toList());
    
    // Breakpoint в IDE может вести себя странно
    // Логирование становится сложным
}

6. Проблемы с I/O операциями

public class IOIssue {
    static final String[] urls = {"url1", "url2", ..., "url1000"};
    
    public void badParallelIO() {
        // ❌ ПЛОХО: параллельные I/O операции
        List<String> results = Arrays.stream(urls)
            .parallelStream()
            .map(this::fetchFromNetwork)  // Сеть медленная
            .collect(Collectors.toList());
        // Может перегрузить сеть, создать слишком много соединений
        // parallelStream() создаст столько потоков, сколько ядер
        // Но сеть может быть узким местом!
    }
    
    public void goodIO() {
        // ✅ ХОРОШО: используйте ExecutorService для I/O
        ExecutorService executor = Executors.newFixedThreadPool(10);
        
        List<Future<String>> futures = Arrays.stream(urls)
            .map(url -> executor.submit(() -> fetchFromNetwork(url)))
            .collect(Collectors.toList());
        
        List<String> results = futures.stream()
            .map(future -> {
                try {
                    return future.get();
                } catch (Exception e) {
                    return null;
                }
            })
            .collect(Collectors.toList());
    }
}

Когда использовать параллельные стримы

✅ ИСПОЛЬЗУЙТЕ

// 1. Большие коллекции (> 10,000 элементов)
List<Integer> bigData = generateMillionIntegers();
long sum = bigData.parallelStream()
    .filter(n -> n % 2 == 0)
    .reduce(0, Integer::sum);

// 2. Сложные вычисления
List<Product> products = getProducts();  // 100,000
List<ProductReport> reports = products.parallelStream()
    .map(this::expensiveCalculation)
    .map(this::anotherExpensiveCalc)
    .collect(Collectors.toList());

// 3. Independent данные (без shared state)
List<Order> orders = getOrders();
List<OrderTotal> totals = orders.parallelStream()
    .map(this::calculateTotal)  // Каждый order независим
    .collect(Collectors.toList());

❌ НЕ ИСПОЛЬЗУЙТЕ

// 1. Маленькие коллекции
List<Integer> small = Arrays.asList(1, 2, 3, 4, 5);
int sum = small.parallelStream()  // Overhead > польза
    .reduce(0, Integer::sum);

// 2. I/O операции
List<String> urls = getUrls();
List<Response> responses = urls.parallelStream()
    .map(this::fetchUrl)  // Сеть медленная
    .collect(Collectors.toList());

// 3. Shared mutable state
List<Integer> data = getData();
List<Integer> results = new ArrayList<>();
data.parallelStream()
    .forEach(n -> results.add(n * 2));  // Race condition!

// 4. Порядок критичен
List<String> logs = getLog();
logoutput logs.parallelStream()
    .forEach(System.out::println);  // Порядок нарушится

Практический пример

public class ProcessingService {
    
    public List<ProcessedData> processLargeDataset(
        List<RawData> dataset) {
        
        // Хорошая кандидатура для parallelStream:
        // - Большой dataset
        // - Каждый элемент обрабатывается независимо
        // - Обработка требует времени
        // - Порядок не важен
        
        return dataset.parallelStream()
            .filter(this::isValidData)           // Фильтрация
            .map(this::transformData)            // Трансформация
            .map(this::enrichWithMetadata)       // Обогащение
            .filter(d -> d.getQualityScore() > 0.7)  // Контроль качества
            .collect(Collectors.toList());
    }
    
    // Но для I/O используем ExecutorService
    public List<RemoteData> fetchFromMultipleSources(
        List<String> sourceUrls) throws InterruptedException {
        
        ExecutorService executor = Executors.newFixedThreadPool(5);
        
        List<Future<RemoteData>> futures = sourceUrls.stream()
            .map(url -> executor.submit(() -> fetchFromSource(url)))
            .collect(Collectors.toList());
        
        List<RemoteData> results = futures.stream()
            .map(future -> {
                try {
                    return future.get(5, TimeUnit.SECONDS);
                } catch (TimeoutException e) {
                    return null;
                } catch (Exception e) {
                    logger.error("Fetch failed", e);
                    return null;
                }
            })
            .filter(Objects::nonNull)
            .collect(Collectors.toList());
        
        executor.shutdown();
        return results;
    }
}

Резюме

Главное правило: используйте параллельные стримы осторожно. Правильно используемые — они дают хорошие ускорения. Неправильно — становятся узким местом.