Что такое Stream API?

Интерфейс java.util.Stream представляет собой последовательность элементов, над которой можно производить различные операции.

Для чего нужны стримы?

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

IntStream.of(50, 60, 70, 80, 90, 100, 110, 120)

.filter(x -> x < 90)

.map(x -> x + 10)

.limit(3)

.forEach(System.out::print);

Создание экземпляра Stream

Stream.empty() //Пустой стрим

list.stream() //Стрим из List

map.entrySet().stream() //Стрим из Map

Arrays.stream(array) //Стрим из массива

Stream.of("1", "2", "3") //Стрим из указанных элементов

Стрим из BufferedReader с помощью метода lines(); нужно закрывать close().

подробнее

Типы операций

Промежуточные (“intermediate”, “lazy”) - обрабатывают поступающие элементы и возвращают стрим. Может быть много или ни одной.

Терминальные (“terminal”, ещё называют “eager”) - обрабатывают элементы и завершают работу стрима, может быть только один.

Важные моменты:

Обработка не начнётся до тех пор, пока не будет вызван терминальный оператор.

list.stream().filter(s - > s > 5) (не возьмёт ни единого элемента из списка);
Экземпляр стрима нельзя использовать более одного раза.

Коллекции	Streams
Конечны (хранят набор элементов)	Бесконечны
Индивидуальный доступ к элементам	Нет индивид. доступа к элементам
Можно менять (добавлять/удалять) элементы, в т.ч. через итератор	Если как то обрабатываем данные, то не влияет на источник

Почему Stream называют ленивым?

Ленивое программирование - технология, которая позволяет вам отсрочить вычисление кода до тех пор, пока не понадобится его результирующее значение.

Блок обработки – промежуточные операции не выполняются, пока не вызовется терминальная.

Способы создания стрима

Из коллекции:

Stream<String> fromCollection = Arrays.asList("x", "y", "z").stream();

Из набора значений:

Stream<String> fromValues = Stream.of("x", "y", "z");

Из массива:

Stream<String> fromArray = Arrays.stream(new String[]{"x", "y", "z"});

Из файла (каждая строка в файле будет отдельным элементом в стриме):

Stream<String> fromFile = Files.lines(Paths.get("input.txt"));

Из строки::

IntStream fromString = "0123456789".chars();

С помощью Stream.builder():

Stream<String> fromBuilder = Stream.builder().add("z").add("y").add("z").build();

С помощью Stream.iterate() (бесконечный):

Stream<Integer> fromIterate = Stream.iterate(1, n -> n + 1);

С помощью Stream.generate() (бесконечный):

Stream<String> fromGenerate = Stream.generate(() -> "0");

Промежуточные методы

Метод	Описание
peek()	Может видеть состояние данных в потоке (в основном для отладки) Stream<T> peek(Consumer<? super T> action);
map()	Позволяет задать функцию преобразования одного объекта в другой <R> Stream<R> map(Function<? super T, ? extends R> mapper);
flatMap()	Можно преобразовать один элемент в ноль, один или множество других <R> Stream<R> flatMap(Function<? super T, ? extends Stream<? extends R>> mapper);
filter()	Который фильтрует все элементы, возвращая только те, что соответствуют условию Stream<T> filter(Predicate<? super T> predicate);
limit()	Возвращает модифицированный поток, в котором не более n элементов Stream<T> limit(long maxSize);
skip()	Возвращает новый поток, в котором пропущены первые n элементов Stream<T> skip(long n);
sorted()	Для сортировки тех объектов, которые реализуют интерфейс Comparable Stream<T> sorted(); или Для реализации своей логики сортировки Stream<T> sorted(Comparator<? super T> comparator);
distinct()	Возвращает только уникальные элементы в виде потока Stream<T> distinct();

Терминальные методы

Метод	Описание
forEach()	Произведет переданное действие с каждым элементом стрима void forEach(Consumer<? super T> action);
findFirst()	Первый в порядке следования элемент из стрима Возвращает Optional т.к. элемента может не быть Optional<T> findFirst();
allMatch()	Позволяет удостовериться, удовлетворяют ли все элементы стрима определенному условию boolean allMatch(Predicate<? super T> predicate);
min()	Возвращает минимальный элемент из стрима Optional<T> min(Comparator<? super T> comparator);
max()	Возвращает максимальный элемент из стрима Optional<T> max(Comparator<? super T> comparator);
count()	Возвращает количество элементов, оставшееся в стриме long count();
collect()	Собирает элементы стрима в новое хранилище (использует Collectors) <R> R collect(Supplier<R> supplier, BiConsumer<R, ? super T> accumulator, BiConsumer<R, R> combiner); или <R, A> R collect(Collector<? super T, A, R> collector);
reduce()	Позволяет выполнять какое-то действие на всей коллекции и возвращать один результат T reduce(T identity, BinaryOperator<T> accumulator); или Optional<T> reduce(BinaryOperator<T> accumulator); или <U> U reduce(U identity, BiFunction<U, ? super T, U> accumulator, BinaryOperator<U> combiner);

Класс Collectors и его методы

В Java 8 в классе Collectors реализовано несколько распространённых коллекторов:

Метод	Описание
toList() toCollection() toSet()	представляют стрим в виде списка, коллекции или множества
toConcurrentMap() toMap()	позволяют преобразовать стрим в Map
averagingInt() averagingDouble() averagingLong()	возвращают среднее значение
summingInt() summingDouble() ummingLong()	возвращает сумму
summarizingInt() summarizingDouble() summarizingLong()	возвращают SummaryStatistics с разными агрегатными значениями
partitioningBy()	разделяет коллекцию на две части по соответствию условию и возвращает их как Map<Boolean, List>
groupingBy()	разделяет коллекцию на несколько частей и возвращает Map<N, List <T>>
mapping()	дополнительные преобразования значений для сложных Collector-ов

Так же существует возможность создания собственного коллектора через Collector.of():

Collector <String, List<String>, List<String>> toList = Collector.of(

ArrayList::new,

List::add,

(l1, l2) -> { l1.addAll(l2); return l1; }

);

Параллельная обработке в Java 8.

Стримы могут быть последовательными и параллельными.

Операции над последовательными стримами выполняются в одном потоке процессора, над параллельными — используя несколько потоков процессора.

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

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

Для создания параллельного потока из коллекции можно также использовать метод parallelStream() интерфейса Collection.

Чтобы сделать обычный последовательный стрим параллельным, надо вызвать у объекта Stream метод parallel().

Метод isParallel() позволяет узнать является ли стрим параллельным.

С помощью, методов parallel() и sequential() можно определять какие операции могут быть параллельными, а какие только последовательными.

Так же из любого последовательного стрима можно сделать параллельный и наоборот:

collection

.stream()

.peek(...) // операция последовательна

.parallel()

.map(...) // операция может выполняться параллельно

.sequential()

.reduce(...) // операция снова последовательна

Как правило, элементы передаются в стрим в том же порядке, в котором они определены в источнике данных.

При работе с параллельными стримами система сохраняет порядок следования элементов.

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

Критерии, которые могут повлиять на производительность в параллельных стримах:

Размер данных - чем больше данных, тем сложнее сначала разделять данные, а потом их соединять.
Количество ядер процессора. Теоретически, чем больше ядер в компьютере, тем быстрее программа будет работать. Если на машине одно ядро, нет смысла применять параллельные потоки.
Чем проще структура данных, с которой работает поток, тем быстрее будут происходить операции.

Например, данные из ArrayList легко использовать, так как структура данной коллекции предполагает последовательность несвязанных данных. А вот коллекция типа LinkedList - не лучший вариант, так как в последовательном списке все элементы связаны с предыдущими/последующими. И такие данные трудно распараллелить.
Над данными примитивных типов операции будут производиться быстрее, чем над объектами классов.
Крайне не рекомендуется использовать параллельные стримы для скольких-нибудь долгих операций (например, сетевых соединений), так как все параллельные стримы работают c одним ForkJoinPool, то такие долгие операции могут остановить работу всех параллельных стримов в JVM из-за отсутствия доступных потоков в пуле, т.е. параллельные стримы стоит использовать лишь для коротких операций, где счет идет на миллисекунды, но не для тех где счет может идти на секунды и минуты;
Сохранение порядка в параллельных стримах увеличивает издержки при выполнении и если порядок не важен, то имеется возможность отключить его сохранение и тем самым увеличить производительность, использовав промежуточную операцию unordered():

collection.parallelStream()

.sorted()

.unordered()

.collect(Collectors.toList());

Стримы для примитивов

Кроме универсальных объектных существуют особые виды стримов для работы с примитивными типами данных int, long и double: IntStream, LongStream и DoubleStream.

Эти примитивные стримы работают так же, как и обычные объектные, но со следующими отличиями:

используют специализированные лямбда-выражения;

например, IntFunction или IntPredicate вместо Function и Predicate
поддерживают дополнительные конечные операции: sum(), average(), mapToObj().

Stream API

Вопросы: