Параллельные потоки

Обычный поток данных обрабатывает элементы по одному, на одном ядре процессора. Параллельный поток делит элементы на части и обрабатывает их одновременно в нескольких потоках выполнения (threads). Включается он одним вызовом, но выгоден далеко не всегда.

parallelStream

У коллекции вместо stream() вызывают parallelStream(), а уже готовый поток переключают методом parallel(). Остальная цепочка операций не меняется:

void main() {
    List<Integer> prices = List.of(120, 45, 300, 80);
 
    int total = prices.parallelStream()
            .mapToInt(p -> p)
            .sum();
 
    IO.println("Итого: $" + total);
}

Результат:

Итого: $545

Результат тот же, что у обычного потока, — меняется только то, как идёт работа внутри.

Кто выполняет работу

Части параллельного потока обрабатывают потоки выполнения из общего пула ForkJoinPool.commonPool(), а также главный поток main. Их обычно столько, сколько ядер у процессора. Выведем, кто обработал каждый элемент:

void main() {
    List.of(1, 2, 3, 4, 5, 6).parallelStream()
            .forEach(n -> IO.println(n + " " + Thread.currentThread().getName()));
}

Результат (порядок и имена меняются от запуска к запуску):

6 ForkJoinPool.commonPool-worker-4
5 ForkJoinPool.commonPool-worker-5
1 ForkJoinPool.commonPool-worker-2
4 main
2 ForkJoinPool.commonPool-worker-1
3 ForkJoinPool.commonPool-worker-3

Порядок элементов

forEach в параллельном потоке выполняет действие в том порядке, в каком закончили части, — то есть в случайном. Если порядок важен, используйте forEachOrdered. А toList() и collect() сохраняют исходный порядок сами:

void main() {
    List<String> names = List.of("Анна", "Олег", "Ира", "Пётр");
 
    names.parallelStream().forEachOrdered(n -> IO.print(n + " "));
    IO.println("");
 
    IO.println(names.parallelStream().map(String::toUpperCase).toList());
}

Результат:

Анна Олег Ира Пётр 
[АННА, ОЛЕГ, ИРА, ПЁТР]

forEachOrdered заставляет потоки выполнения ждать друг друга, поэтому часть выигрыша от параллельности теряется.

Когда это выгодно

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

boolean isPrime(int n) {
    if (n < 2) {
        return false;
    }
    for (int d = 2; d * d <= n; d++) {
        if (n % d == 0) {
            return false;
        }
    }
    return true;
}
 
void main() {
    long start = System.currentTimeMillis();
    long count = IntStream.range(0, 3_000_000).filter(this::isPrime).count();
    IO.println("Последовательно: " + count + " за " + (System.currentTimeMillis() - start) + " мс");
 
    start = System.currentTimeMillis();
    count = IntStream.range(0, 3_000_000).parallel().filter(this::isPrime).count();
    IO.println("Параллельно: " + count + " за " + (System.currentTimeMillis() - start) + " мс");
}

Результат (время зависит от компьютера):

Последовательно: 216816 за 1585 мс
Параллельно: 216816 за 229 мс

А вот где параллельность вредит: маленькие коллекции (деление на части дороже самой работы), ожидание сети или диска, источники, которые плохо делятся (LinkedList, Stream.iterate), и операции, зависящие от порядка, — limit, findFirst, forEachOrdered. Без замера времени параллельный поток не включайте.

Частая ошибка

Самая опасная ошибка — менять из параллельного потока общую переменную или коллекцию. ArrayList не рассчитан на запись из нескольких потоков выполнения одновременно: часть элементов теряется, а иногда программа падает с исключением:

void main() {
    List<Integer> result = new ArrayList<>();
    IntStream.range(0, 10_000).parallel().forEach(n -> result.add(n));
    IO.println(result.size());
}

Результат (меняется от запуска к запуску):

6593

Вместо 10 000 элементов — меньше, и каждый раз по-разному. Не изменяйте внешние объекты из лямбды, а собирайте результат самим потоком: IntStream.range(0, 10_000).parallel().boxed().toList() всегда вернёт все 10 000 элементов по порядку.