Какой ExecutorService используется в Parallel Stream?

ExecutorService в Parallel Stream

Parallel Stream в Java использует ForkJoinPool, а именно его static общий пул (Common Fork/Join Pool). Это важное знание для понимания поведения параллельных потоков и оптимизации производительности.

Что такое ForkJoinPool?

ForkJoinPool — это специализированный ExecutorService, разработанный для эффективной работы с алгоритмом "разделяй и властвуй" (divide and conquer). Он использует work-stealing алгоритм, который позволяет потокам, завершившим свою работу, брать задачи у других потоков из очереди.

Общий пул (Common Fork/Join Pool)

Для parallel Stream по умолчанию используется статический общий пул ForkJoinPool:

ForkJoinPool commonPool = ForkJoinPool.commonPool();
System.out.println(commonPool.getParallelism());  // Зависит от кол-ва ядер

Важные свойства:

• Это общий ресурс для всего приложения
• Количество потоков по умолчанию = количество процессоров - 1 (Runtime.getRuntime().availableProcessors() - 1)
• Пул переиспользуется для всех parallel Stream операций
• Lazy инициализация — создаётся при первом использовании

Как Parallel Stream использует ForkJoinPool

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

// Этот код использует common ForkJoinPool
int sum = numbers.parallelStream()
    .mapToInt(Integer::intValue)
    .sum();

System.out.println(sum);  // 36

Внутри Stream API происходит следующее:

// Примерно так это работает
ForkJoinTask<Integer> task = new SumTask(numbers, 0, numbers.size());
ForkJoinPool.commonPool().invoke(task);
int result = task.join();

Параллелизм и количество потоков

public static void main(String[] args) {
    int processors = Runtime.getRuntime().availableProcessors();
    System.out.println("Доступных процессоров: " + processors);
    
    ForkJoinPool pool = ForkJoinPool.commonPool();
    System.out.println("Параллелизм в common pool: " + pool.getParallelism());
    // Обычно: parallelism = processors - 1
    
    // На 8-ядерной системе:
    // Доступных процессоров: 8
    // Параллелизм в common pool: 7
}

Проблемы с общим пулом

1. Блокирование пула Если задачи в parallel Stream содержат блокирующие операции, это может замедлить весь пул:

// ПЛОХО: блокирующая операция в parallel Stream
List<String> urls = Arrays.asList("url1", "url2", "url3");
urls.parallelStream()
    .map(url -> fetchDataFromServer(url))  // Блокирует поток
    .collect(Collectors.toList());

2. Задачи могут взаимно блокировать друг друга Если параллельные задачи ждут друг друга, может возникнуть deadlock в пуле.

Использование собственного ForkJoinPool

Для избежания конкуренции за общий пул, можно создать собственный пул:

public class CustomPoolExample {
    public static void main(String[] args) throws Exception {
        // Создаём собственный ForkJoinPool с 4 потоками
        ForkJoinPool customPool = new ForkJoinPool(4);
        
        List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8);
        
        // Используем invoke() для запуска parallel Stream в custom pool
        int sum = customPool.invoke(() -> 
            numbers.parallelStream()
                .mapToInt(Integer::intValue)
                .sum()
        );
        
        System.out.println("Сумма: " + sum);
        customPool.shutdown();
    }
}

Варианты конфигурации параллелизма

Использование системных свойств для изменения параллелизма common pool:

# В JVM параметрах при запуске
java -Djava.util.concurrent.ForkJoinPool.common.parallelism=4 MyApp

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

Work-Stealing алгоритм

ForkJoinPool реализует work-stealing стратегию:

// Если поток T1 закончил свою работу раньше, чем T2,
// T1 может взять половину работы из очереди T2

Task<Integer> task = new RecursiveTask<Integer>() {
    protected Integer compute() {
        if (taskIsSmall()) {
            return solveDirectly();
        } else {
            Task<Integer> left = new Task(...);
            Task<Integer> right = new Task(...);
            
            left.fork();  // Отправить в пул
            right.fork(); // Отправить в пул
            
            return left.join() + right.join();  // Ждём результатов
        }
    }
};

Лучшие практики

✅ Используйте parallel Stream для:

• Больших коллекций (тысячи элементов и больше)
• CPU-intensive операций (вычисления, фильтрация)
• Операций без блокирования и побочных эффектов

❌ Не используйте parallel Stream для:

• Малых коллекций (overhead от потокообразования не стоит)
• I/O операций (блокирующие операции)
• Операций с shared состоянием

Мониторинг ForkJoinPool

ForkJoinPool pool = ForkJoinPool.commonPool();
System.out.println("Active threads: " + pool.getActiveThreadCount());
System.out.println("Parallelism: " + pool.getParallelism());
System.out.println("Queued tasks: " + pool.getQueuedTaskCount());
System.out.println("Steal count: " + pool.getStealCount());

Заключение

Parallel Stream в Java использует ForkJoinPool.commonPool(), который:

• ✅ Автоматически управляется JVM
• ✅ Использует work-stealing для балансировки нагрузки
• ⚠️ Может быть узким местом при конкурирующих parallel операциях
• ⚠️ Неэффективен для блокирующих операций

Для критичного контроля над параллелизмом рекомендуется создавать собственные ForkJoinPool экземпляры.