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

Кроме последовательных потоков Stream API поддерживает параллельные потоки. Распараллеливание потоков позволяет задействовать несколько ядер процессора (если целевая машина многоядерная) и тем самым может повысить производительность и ускорить вычисления. В то же время говорить, что применение параллельных потоков на многоядерных машинах однозначно повысит производительность - не совсем корректно. В каждом конкретном случае надо проверять и тестировать.

Чтобы сделать обычный последовательный поток параллельным, надо вызвать у объекта Stream метод parallel. Кроме того, можно также использовать метод parallelStream() интерфейса Collection для создания параллельного потока из коллекции.

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

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

import java.util.Optional;
import java.util.stream.Stream;

public class Program {
 
    public static void main(String[] args) {
		
		Stream<Integer> numbersStream = Stream.of(1, 2, 3, 4, 5, 6);
		Optional<Integer> result = numbersStream.parallel().reduce((x,y)-> x*y);
		System.out.println(result.get()); // 720
	}
}

Еще один пример:

import java.util.Arrays;
import java.util.List;


public class Program {

    public static void main(String[] args) {

        List<String> people = Arrays.asList("Tom","Bob", "Sam", "Kate", "Tim");

        System.out.println("Последовательный поток");
        people.stream().filter(p->p.length()==3).forEach(System.out::println);

        System.out.println("\nПараллельный поток");
        people.parallelStream().filter(p->p.length()==3).forEach(System.out::println);
    }
}

В данном случае сначала для списка people создаем поток и выполняем над ним ряд операций в последовательном режиме. В частности, находим в списке строки, длина которых равна 3 и выводим их на консоль. В этом случае все операции с потоком будут производиться над элементами списка в том порядке, в котоом элементы идут в списке.

Затем с помощью метода people.parallelStream() для списка создается параллельный поток. Причем применяются те же операции, однако теперь порядок, в котором над элементами списка будут производиться операции, не детерминирован.

Консольный вывод:

Последовательный поток
Tom
Bob
Sam
Tim

Параллельный поток
Sam
Tim
Bob
Tom

В случае с параллельным потоком вывод недетерминирован и может отличаться.

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

Stream<Integer> numbersStream = Stream.of(1, 2, 3, 4, 5, 6);
Integer result = numbersStream.parallel().reduce(1, (x,y)->x * y);
System.out.println(result);

Фактически здесь происходит перемножение чисел. При этом нет разницы между 1 * 2 * 3 * 4 * (5 * 6) или 5 * 6 * 1 * (2 * 3) * 4. Мы можем расставить скобки любым образом, разместить последовательность чисел в любом порядке, и все равно мы получим один и тот же результат. То есть данная операция является ассоциативной и поэтому может быть распараллелена.

Вопросы производительности в параллельных операциях

Фактически применение параллельных потоков сводится к тому, что данные в потоке будут разделены на части, каждая часть обрабатывается на отдельном ядре процессора, и в конце эти части соединяются, и над ними выполняются финальные операции. Рассмотрим некоторые критерии, которые могут повлиять на производительность в параллельных потоках:

• Размер данных. Чем больше данных, тем сложнее сначала разделять данные, а потом их соединять.

• Количество ядер процессора. Теоретически, чем больше ядер в компьютере, тем быстрее программа будет работать. Если на машине одно ядро, нет смысла применять параллельные потоки.

• Чем проще структура данных, с которой работает поток, тем быстрее будут происходить операции. Например, данные из ArrayList легко использовать, так как структура данной коллекции предполагает последовательность несвязанных данных. А вот коллекция типа LinkedList - не лучший вариант, так как в последовательном списке все элементы связаны с предыдущими/последующими. И такие данные трудно распараллелить.

• Над данными примитивных типов операции будут производиться быстрее, чем над объектами классов

Упорядоченность в параллельных потоках

Как правило, элементы передаются в поток в том же порядке, в котором они определены в источнике данных. При работе с параллельными потоками система сохраняет порядок следования элементов. Исключение составляет метод forEach(), который может выводить элементы в произвольном порядке. И чтобы сохранить порядок следования, необходимо применять метод forEachOrdered:

phones.parallelStream()
    .sorted()
    .forEachOrdered(s->System.out.println(s));

Сохранение порядка в параллельных потоках увеличивает издержки при выполнении. Но если нам порядок не важен, то мы можем отключить его сохранение и тем самым увеличить производительность, использовав метод unordered:

phones.parallelStream()
    .sorted()
	.unordered()
    .forEach(s->System.out.println(s));