5. Тип Stream<T> у Java 8
5.1. Приклад-01 — клас Stream
Операції з потоками Observable мають багато спільного з потоками Stream. Відмінність полягає в тому, що елемент потоку Stream не можна обробити, доки не буде отримано весь потік Stream, тоді як елемент потоку Observable можна обробити (спостерігається) одразу після його отримання, не чекаючи на отримання всього потоку Observable. Ще одна відмінність полягає в тому, що після отримання потоку Stream його значення використовуються шляхом їх почергового вилучення (pull) з потоку Stream. Для спостережуваного об’єкта все інакше. Як тільки він надсилає значення, воно передається (pushed) його підписнику.
Поняття Stream реалізується кількома класами. Тут ми розглянемо клас Stream<T>:
Клас Stream містить 39 методів. Ми розглянемо деякі з них. Розглянемо такий код:
 |
package dvp.java8.streams;
import java.util.List;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple01 {
public static void main(String[] args) {
// список осіб
List<Personne> personnes = Personnes.get();
// відображення 1
personnes.stream().forEach(p -> {
System.out.println(p);
});
System.out.println("----------------");
// відображення 2
personnes.stream().forEach(System.out::println);
}
}
- рядок 11: створюється екземпляр списку осіб;
- рядок 13: на основі цього списку створюється Stream. Усі колекції можна таким чином перетворити на потік Stream. Це дозволяє скористатися всіма методами цього класу, що дає змогу обробляти елементи колекції більш лаконічно, ніж за допомогою циклів. Це також дозволяє скористатися паралелізмом обробки елементів, коли це можливо;
- рядок 13: метод [Stream.forEach] має таку сигнатуру:
 |
Бачимо, що параметром методу є функціональний інтерфейс [Consumer<T>], представлений у розділі 4.4 — інтерфейс, єдиний метод якого використовує тип T і не повертає ніякого значення.
- У коді:
personnes.stream().forEach(p -> {
System.out.println(p);
});
- [personnes.stream()] генерує потік елементів типу [Personne], який подається в метод [forEach]. Параметр p має тип [Personne], а надана лямбда-функція відображає цю особу;
Попередній код можна спростити наступним чином (рядок 18):
personnes.stream().forEach(System.out::println);
Замість того, щоб передавати як параметр значення лямбда-функції, ми передаємо посилання на існуючий метод, у даному випадку метод println класу System.out. Звісно, цей метод повинен мати правильну сигнатуру, у даному випадку сигнатуру методу [Consumer.accept]: void accept(T t). Як зазначалося раніше, параметром методу [accept] буде тип [Personne];
Ми отримуємо такі результати:
Після використання потоку Stream його більше не можна використовувати. Щоб скористатися ним знову, його потрібно створити заново. Це ілюструє наступний код [Exemple01b]:
package dvp.java8.streams;
import java.util.stream.Stream;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple01b {
public static void main(String[] args) {
// потік осіб
Stream<Personne> personnes = Personnes.get().stream();
// перегляд 1
personnes.forEach(p -> {
System.out.println(p);
});
System.out.println("----------------");
// перегляд 2
personnes.forEach(System.out::println);
}
}
- рядок 11: для оптимізації коду вирішено побудувати Stream лише один раз. Отримані результати такі:
{"nom":"jean","age":20,"poids":70.0,"sexe":"HOMME"}
{"nom":"marie","age":10,"poids":30.0,"sexe":"FEMME"}
{"nom":"camille","age":30,"poids":55.0,"sexe":"FEMME"}
----------------
Exception in thread "main" java.lang.IllegalStateException: stream has already been operated upon or closed
at java.util.stream.AbstractPipeline.sourceStageSpliterator(Unknown Source)
at java.util.stream.ReferencePipeline$Head.forEach(Unknown Source)
at dvp.java8.streams.Exemple02b.main(Exemple02b.java:18)
Кожного разу, коли потрібно використати Stream, його доводиться створювати заново, навіть якщо він уже був створений раніше.
5.2. Приклад-02 — паралельна обробка елементів потоку
 |
Розглянемо такий код:
package dvp.java8.streams;
import java.util.List;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple02 {
public static void main(String[] args) {
// список осіб
List<Personne> personnes = Personnes.get();
// перегляд 1
personnes.stream().forEach(Exemple02::affiche);
System.out.println("-----------------");
// відображення 2
personnes.stream().parallel().forEach(Exemple02::affiche);
}
public static void affiche(Personne p) {
System.out.printf("Personne %s sur thread %s%n", p, Thread.currentThread().getName());
}
}
- рядки 19–21: метод [affiche] виводить на консоль рядок jSON, що відповідає певній особі, а також ім’я потоку виконання, в якому відбувається виведення;
- рядок 13: виводить список осіб. Зауважимо, що параметром методу [forEach] є посилання на попередній статичний метод;
- рядок 16: виконується те саме, але за допомогою методу [parallel] задається, щоб обробка елементів потоку здійснювалася паралельно у декількох потоках. Не кожна обробка може виконуватися паралельно. Тут слід припустити, що порядок відображення не має значення, оскільки при паралельній обробці порядок виконання потоків не гарантований. Крім того, зверніть увагу на синтаксис, який стане повсюдним як для Stream, так і для Observable:
- (продовження)
- flux генерує елементи e1, які подаються в метод m1;
- flux.m1, у свою чергу, є потоком елементів e2, які подаються в метод m2;
- flux.m1.m2 — це потік елементів e3, які подаються в метод m3;
Тип елементів e1, e2, e3 може змінюватися в процесі обробки вихідного потоку.
Виконання цього коду дає такий результат:
Бачимо, що паралельне виконання (рядки 5–7) відбувалося у трьох різних потоках і не дотримувалося порядку елементів, який відповідає рядкам 1–3. У цьому документі ми не будемо детально зупинятися на паралельній обробці елементів Stream, оскільки для цього потрібно розглянути умови, що роблять таку обробку можливою. Виявляється, що лише деякі операції можна виконувати паралельно. Однією з тих, що природно піддаються паралелізму, є підсумовування числових елементів потоку, яке ми зараз і розглянемо.
5.3. Приклад-03 — паралельна обробка елементів потоку
 |
Розглянемо такий код (Приклад 03а):
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
public class Exemple03a {
public static void main(String[] args) {
final long limite = 10_000_000L;
// кількість процесорів
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// список чисел
long début = new Date().getTime();
List<Long> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(i);
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// сума чисел — послідовний метод
début = new Date().getTime();
long somme = nombres.stream().reduce(0L, (s, i) -> s + i);
System.out.printf("somme séquentielle : somme=%s, durée (ms)=%s%n", somme, new Date().getTime() - début);
}
}
- у рядку 22 ми використовуємо метод [reduce], сигнатура якого така:
 |  |
- метод [reduce] працює з елементами типу T;
- метод [reduce] застосовує однакову обробку до всіх елементів потоку: початкове значення накопичувача вказується як перший параметр. Як другий параметр передається метод, що реалізує функціональний інтерфейс [BinaryOperator] [2]: на основі кожного елемента та накопичувача цей метод повертає нове значення накопичувача. Кінцеве значення накопичувача — це значення, яке повертає метод [reduce]. Код [3] ілюструє цей механізм. Метод [apply] є методом функціонального інтерфейсу [BinaryOperator] [2];
Повернемося до прикладу коду:
- рядок 12: виводиться кількість ядер, виявлених методом JVM;
- рядки 15–18: створюється список із 10 мільйонів чисел;
- рядок 22: сума цих чисел обчислюється послідовно за допомогою одного потоку;
Отримуємо такі результати:
Тепер замінимо рядок 22 коду на такий (Приклад03b):
long somme = nombres.stream().parallel().reduce(0L, (s, i) -> s + i);
Ми вимагаємо, щоб елементи потоку оброблялися паралельно за допомогою декількох потоків. Це можливо, оскільки порядок додавання чисел не має значення. Отже, можна привласнити n1 чисел потоку T1, n2 чисел — потоку T2, ... і в підсумку підсумувати суми, надані цими різними потоками. У результаті отримуємо такі результати:
Отже, приросту продуктивності практично немає. У наступних прикладах це буде часто траплятися. Управління потоками саме по собі є часомістким. Операція, що виконується кожним ядром, має бути достатньо складною, щоб проявився приріст продуктивності. Це ілюструє наступний приклад (Приклад03c):
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.function.BinaryOperator;
public class Exemple03c {
public static void main(String[] args) {
final long limite = 10_000L;
// кількість процесорів
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// список чисел
long début = new Date().getTime();
List<Long> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(i);
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// сума чисел — послідовний метод
début = new Date().getTime();
BinaryOperator<Long> bo = (s, i) -> {
try {
Thread.sleep(1);
} catch (InterruptedException e) {
}
return s + i;
};
long somme = nombres.stream().reduce(0L, bo);
System.out.printf("somme séquentielle : somme=%s, durée (ms)=%s%n", somme, new Date().getTime() - début);
}
}
- рядок 30: знову використовується метод [reduce], якому як параметр передається посилання на метод із рядків 23–29;
- рядок 28: метод [bo] обчислює суму двох своїх параметрів;
- рядки 24–27: штучно затримуємо потік на 1 мілісекунду, щоб імітувати інтенсивну роботу;
У результаті отримуємо такі результати:
Тепер, якщо замінити рядок 30 на такий:
long somme = nombres.stream().parallel().reduce(0L, bo);
отримуємо такі результати:
Чітко видно приріст продуктивності, який забезпечує паралельне виконання обчислення суми. Для обробки 8 чисел:
- послідовний потік очікує 8 разів по 1 мілісекунді, тобто 8 мс;
- 8 паралельних потоків одночасно чекають по 1 мілісекунді кожен (умовно, для спрощення), тобто загалом 1 мілісекунду для 8 чисел;
Отже, можна очікувати, що паралельне виконання буде у 8 разів швидшим за послідовне. У даному випадку це приблизно так і є.
5.4. Приклад-04 — фільтрування потоку
 |
Розглянемо такий код:
package dvp.java8.streams;
import java.util.List;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple04 {
public static void main(String[] args) {
// список осіб
List<Personne> personnes = Personnes.get();
// перегляди
System.out.println("age < 28 ----------------------");
personnes.stream().filter(p -> p.getAge() < 28).forEach(p -> {
System.out.println(p);
});
System.out.println("poids < 50 ----------------------");
personnes.stream().filter(p -> p.getPoids() < 50).forEach(p -> {
System.out.println(p);
});
System.out.println("age < 28 ----------------------");
personnes.stream().filter(p -> p.getAge() < 28).forEach(System.out::println);
System.out.println("poids < 50 ----------------------");
personnes.stream().filter(p -> p.getPoids() < 50).forEach(System.out::println);
}
}
- рядок 14: метод [Stream.filter] має таку сигнатуру:
 |
- метод [filter] очікує як параметр екземпляр функціонального інтерфейсу [Predicate], представленого в розділі 4.2, єдиний метод якого, що підлягає реалізації, має такий вигляд: boolean test(T t);
- метод [filter] повертає елементи потоку, що відповідають Predicate. Отже, він слугує для фільтрування Stream;
Розглянемо такий код:
package dvp.java8.streams;
import java.util.List;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple04 {
public static void main(String[] args) {
// список осіб
List<Personne> personnes = Personnes.get();
// перегляди
System.out.println("age < 28 ----------------------");
personnes.stream().filter(p -> p.getAge() < 28).forEach(p -> {
System.out.println(p);
});
System.out.println("poids < 50 ----------------------");
personnes.stream().filter(p -> p.getPoids() < 50).forEach(p -> {
System.out.println(p);
});
System.out.println("age < 28 ----------------------");
personnes.stream().filter(p -> p.getAge() < 28).forEach(System.out::println);
System.out.println("poids < 50 ----------------------");
personnes.stream().filter(p -> p.getPoids() < 50).forEach(System.out::println);
}
}
- рядки 14–16: виводять осіб, вік яких <28;
- рядки 18–20: виводять осіб, вага яких <50;
- рядок 22: виконує те саме, що й рядки 14–16, але у більш стислому вигляді;
- рядок 24: виконує те саме, що й рядки 18–20, але більш лаконічно;
Результати виконання такі:
5.5. Приклад-05 — створення Stream<T2> на основі Stream<T1>
 |
Розглянемо такий код:
package dvp.java8.streams;
import java.util.List;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple05 {
public static void main(String[] args) {
// список осіб
List<Personne> personnes = Personnes.get();
// перегляди
System.out.println("Personne --> String ----------------------");
personnes.stream().map(p -> p.getNom()).forEach(System.out::println);
System.out.println("Personne --> Integer ----------------------");
personnes.stream().map(p -> p.getAge()).forEach(System.out::println);
}
}
- у рядку 13 метод [Stream.map] має таку сигнатуру:
 |
Параметром методу [Stream.map] є екземпляр функціонального інтерфейсу [Function], представленого в розділі 4.3, єдиним методом, який потрібно реалізувати, є: R apply(T t). Бачимо, що на основі типу T функція [apply] генерує тип R. Отже, метод [Stream.map] генеруватиме потік Stream типу R на основі потоку типу T (потік типу T тут означає, за неточною, але прийнятою термінологією, яку ми збережемо, потік елементів типу T).
Тепер розглянемо код прикладу:
- рядок 14: від особи p зберігається лише ім’я. Отже, отримуємо потік String;
- рядок 14: від особи p зберігається лише ім’я. Отже, отримуємо потік Integer;
Отримані результати такі:
5.6. Приклад-06 — інші методи класу Stream<T>
 |
Деякі з 39 методів класу Stream проілюструємо за допомогою такого коду:
package dvp.java8.streams;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import dvp.data.Personne;
import dvp.data.Personnes;
import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.DoubleStream;
import java.util.stream.IntStream;
import java.util.stream.Stream;
public class Exemple06 {
// маппер jSON
static private ObjectMapper jsonMapper = new ObjectMapper();
public static void main(String[] args) throws JsonProcessingException {
// список осіб
List<Personne> personnes = Personnes.get();
// усі особи
affiche("all", personnes);
// перша особа
affiche("findFirst", personnes.stream().findFirst().get());
// будь-яка особа
affiche("findAny", personnes.stream().findAny().get());
// особи, крім першої
affiche("skip 1", personnes.stream().skip(1L).collect(Collectors.toList()));
// перші 2 особи
affiche("limit 2", personnes.stream().limit(2L).collect(Collectors.toList()));
// кількість осіб
affiche("count", personnes.stream().count());
// найстарша особа
affiche("age max", personnes.stream().max(Comparator.comparingInt(Personne::getAge)).get());
// найлегша людина
affiche("poids min", personnes.stream().min(Comparator.comparingDouble(Personne::getPoids)).get());
// остання особа за алфавітним порядком імен
affiche("nom max", personnes.stream().max((p1, p2) -> p1.getNom().compareToIgnoreCase(p2.getNom())).get());
// сумарний вік усіх осіб
affiche("âge total (reduce)", personnes.stream().map(p -> p.getAge()).reduce(0, (a1, a2) -> a1 + a2));
// особи за віком у порядку зростання
affiche("personnes par âge croissant",
personnes.stream().sorted(Comparator.comparingInt(Personne::getAge)).collect(Collectors.toList()));
// чи є особи віком понад 100 років?
affiche("des personnes de + de 100 ans (anyMatch)", personnes.stream().anyMatch(p -> p.getAge() > 100));
// чи всім особам не більше 100 років?
affiche("des personnes de + de 100 ans (noneMatch)", personnes.stream().noneMatch(p -> p.getAge() > 100));
// чи всі особи старші за 8 років
affiche("des personnes de + de 8 ans (allMatch)", personnes.stream().allMatch(p -> p.getAge() > 8));
// людей групують за статтю
affiche("personnes regroupées par sexe", personnes.stream().collect(Collectors.groupingBy(p -> p.getSexe())));
// видалення дублікатів елементів зі списку
affiche("distinct", Stream.of(1, 2, 1).distinct().collect(Collectors.toList()));
// з Stream<Stream<T>> створюємо Stream<T>
affiche("flatMap", Stream.of(1, 2, 3).flatMap(i -> Stream.of(i, i + 10)).collect(Collectors.toList()));
// з Stream<Stream<Integer>> утворюємо IntStream, для якого обчислюємо суму
affiche("flatMapToInt", Stream.of(1, 2, 3).flatMapToInt(i -> IntStream.of(i, i + 10)).sum());
// з Stream<Stream<Integer>> утворюють DoubleStream, а потім масив
affiche("flatMapToDouble", Stream.of(1, 2, 3).flatMapToDouble(i -> DoubleStream.of(i, i * 1.2)).toArray());
// max з потоку цілих чисел
affiche("reduce Integer::max", Stream.of(1, 10, 8).reduce(Integer::max).get());
// min потоку чисел типу Double
affiche("reduce Integer::min", Stream.of(1.5, 10.4, 8.9).reduce(Double::min).get());
// середнє значення потоку цілих чисел
affiche("IntStream average", IntStream.of(1, 10, 8).average().getAsDouble());
// статистика потоку цілих чисел
affiche("IntStream summaryStatistics", IntStream.of(1, 10, 8).summaryStatistics());
}
public static <T> void affiche(String message, T value) throws JsonProcessingException {
System.out.println(String.format("%s ----", message));
System.out.println(jsonMapper.writeValueAsString(value));
}
}
- рядки 72, 75: виводять рядок jSON з другого параметра методу;
- рядок 24: виводить рядок jSON для всіх осіб. Отримуємо такий результат:
5.6.1. [findFirst]
// перша особа
affiche("findFirst", personnes.stream().findFirst().get());
Метод [findFirst] повертає перший елемент потоку, якщо він існує. Його сигнатура така:
 |
Результат має тип Optional<T>, який було введено в Java 8:
 |
Клас Optional<T> дозволяє по-різному обробляти покажчики null. Метод, який повинен повертати тип T, що може мати значення null, може вирішити повернути тип Optional<T>. Метод [Optional<T>.isPresent()] дозволяє дізнатися, чи повернув метод значення чи ні. Наступний код [Exemple06b] ілюструє частину принципу роботи Optional<T>:
package dvp.java8.streams;
import java.util.Optional;
import com.fasterxml.jackson.core.JsonProcessingException;
public class Exemple06b {
public static void main(String[] args) throws JsonProcessingException {
// опціональний без значення
Optional<Integer> o1 = m1();
System.out.println(o1.isPresent());
affiche(o1);
// опціональний параметр із значенням
Optional<Integer> o2 = m2();
System.out.println(o2.isPresent());
affiche(o2);
}
private static void affiche(Optional<Integer> o1) {
try {
// отримуємо значення опціонального параметра
// викликає 1 виняток, якщо значення відсутнє
System.out.println(o1.get());
} catch (Throwable th) {
System.out.printf("%s : %s%n", th.getClass().getName(), th.getMessage());
}
}
public static Optional<Integer> m1() {
// немає значення
return Optional.empty();
}
public static Optional<Integer> m2() {
// значення
return Optional.of(10);
}
}
Отримано такі результати:
false
java.util.NoSuchElementException : No value present
true
10
Повернемося до коду, що ілюструє метод [findFirst]:
// перша особа
affiche("findFirst", personnes.stream().findFirst().get());
- рядок 2: щоб спростити код, ми використовуємо метод [get] на Optional<Personne>, згенерованому методом [findFirst]. Згідно з принципами чистого коду, слід спочатку викликати метод [Optional<Personne>.isPresent()], а потім — метод [get];
Отримано такий результат:
5.6.2. [findAny]
// будь-яка особа
affiche("findAny", personnes.stream().findAny().get());
Метод [findAny] має такий підпис:
 |
Метод [findAny] може повертати будь-який елемент потоку. Під час тестування було помічено, що послідовне виконання повертає перший елемент потоку, тоді як паралельне виконання може повертати будь-який елемент. Це ілюструє наступний код [Exemple06c]:
package dvp.java8.streams;
import java.util.List;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple06c {
// мапер jSON
static private ObjectMapper jsonMapper = new ObjectMapper();
public static void main(String[] args) throws JsonProcessingException {
// список осіб
List<Personne> personnes = Personnes.get();
// усі особи
affiche("all", personnes);
// будь-яка особа
affiche("findAny parallèle", personnes.stream().parallel().findAny().get());
// будь-яка особа
affiche("findAny séquentiel", personnes.stream().findAny().get());
}
public static <T> void affiche(String message, T value) throws JsonProcessingException {
System.out.println(String.format("%s ----", message));
System.out.println(jsonMapper.writeValueAsString(value));
}
}
- рядок 22: findAny, виконаний паралельно;
- рядок 24: findAny виконується послідовно;
Отримано такі результати:
- рядок 4: паралельне виконання повернуло елемент 2 зі списку осіб. Це міг бути й інший;
- рядок 6: послідовне виконання повернуло перший елемент списку осіб;
Використання методу [findAny] має сенс лише при паралельній обробці потоку.
5.6.3. [skip]
// особи, крім першої
affiche("skip 1", personnes.stream().skip(1L).collect(Collectors.toList()));
Метод [skip] має таку сигнатуру:
 |
Метод [skip] пропускає перші n елементів потоку. Як зазначено у документації вище, паралельне виконання цього методу дає незначний приріст продуктивності, а іноді навіть може призвести до її зниження. Справа в тому, що для пропускання перших n елементів потоки змушені координувати свої дії, що нівелює виграш у продуктивності, отриманий завдяки паралелізму.
Метод [skip] повертає потік Stream<Personne>, який перетворюється на тип List<Personne> за допомогою методу [collect], сигнатура якого така:
 |
Метод [collect] приймає як параметр екземпляр типу [Collector], сигнатура якого є складною. Існують попередньо визначені реалізації типу [Collector], які найчастіше дозволяють уникнути необхідності його самостійної реалізації. У даному випадку використовується реалізація [Collectors.toList()]. [Collectors] — це клас, що містить численні статичні методи, які реалізують тип [Collector<T,A,R>]. Саме тут слід шукати, коли потрібно перетворити Stream на стандартну колекцію Java:
 |
Деякі з цих методів ми використаємо пізніше.
Виконання дає такий результат:
Перший елемент списку (jean) було пропущено.
5.6.4. [limit]
// перші 2 особи
affiche("limit 2", personnes.stream().limit(2L).collect(Collectors.toList()));
Метод [limit] має таку сигнатуру:
 |
Метод [limit] дозволяє зберегти лише перші n елементів потоку. Він не пристосований для паралельної обробки.
Виконання дає такий результат:
5.6.5. [count]
// кількість осіб
affiche("count", personnes.stream().count());
Метод [count] має таку сигнатуру:
 |
Метод [count] повертає кількість елементів Stream. Паралельне виконання методу не приносить приросту продуктивності, як показано в наступному коді (Приклад06d1):
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.stream.Stream;
public class Exemple06d1 {
public static void main(String[] args) {
final long limite = 10_000_000L;
// кількість процесорів
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// список чисел
long début = new Date().getTime();
List<Long> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(i);
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// підрахунок чисел — послідовний метод
Stream<Long> sNombres = nombres.stream();
début = new Date().getTime();
long count = sNombres.count();
System.out.printf("comptage séquentiel : compteur=%s, durée (ms)=%s%n", count, new Date().getTime() - début);
}
}
- рядки 11–22: створюється Stream із 10 мільйонами чисел;
- рядки 22–24: підрахунок Stream;
Виконання дає такий результат:
Якщо замінити рядок 22 коду на наступний (Приклад06d2):
Stream<Long> sNombres = nombres.stream().parallel();
отримуємо такі результати:
5.6.6. [max, min]
// найстарша людина
affiche("age max", personnes.stream().max(Comparator.comparingInt(Personne::getAge)).get());
Метод [max] має такий підпис:
 |
Метод [max] повертає максимальне значення потоку, використовуючи компаратор, який передається йому як параметр. Comparator — це функціональний інтерфейс, єдиний метод якого, що підлягає реалізації, має сигнатуру: int compare (T o1, T o2). Цей метод повинен повертати -1, якщо o1 < o2, 0, якщо o1.equals(o2), +1, якщо o1 > o2. Функціональний інтерфейс Comparator має багато статичних методів за замовчуванням, які реалізують інтерфейс Comparator для найпоширеніших випадків. Отже, у коді:
affiche("age max", personnes.stream().max(Comparator.comparingInt(Personne::getAge)).get());
ми використовуємо статичний метод [Comparator.comparingInt], сигнатура якого така:
 |
Тип ToIntFunction є функціональним інтерфейсом:
 |
Метод [applyAsInt] функціонального інтерфейсу ToIntFunction створює тип int на основі типу T. Повернемося до нашого коду:
affiche("age max", personnes.stream().max(Comparator.comparingInt(Personne::getAge)).get());
Фактичним параметром методу [Comparator.comparingInt] тут має бути лямбда-вираз Personne --> int. Ми передаємо посилання на метод [Personne.getAge], який саме має таку сигнатуру. У підсумку отримаємо особу з найбільшим віком. Отримуємо тип Optional<Personne>, з якого витягуємо значення за допомогою методу [Optional.get]. Отримуємо такий результат:
Паралельний розрахунок max не приносить приросту продуктивності, як показує наступний приклад: (Приклад06e1):
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.Random;
import java.util.stream.Stream;
public class Exemple06e1 {
public static void main(String[] args) {
// дані
// final long limit = 100L;
// final boolean verbose = true;
final long limite = 10_000_000L;
final boolean verbose = false;
// кількість процесорів
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// список чисел
long début = new Date().getTime();
List<Long> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(new Random().nextLong());
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// максимальне значення чисел — послідовний метод
Stream<Long> sNombres = nombres.stream();
Comparator<Long> compLong = (l1, l2) -> {
if (verbose) {
// потік
System.out.printf("[%s]", Thread.currentThread().getName());
}
// порівняння
long v1 = l1.longValue();
long v2 = l2.longValue();
if (v1 < v2) {
return -1;
} else {
if (v1 == v2) {
return 0;
} else {
return +1;
}
}
};
début = new Date().getTime();
// максимальна довжина = sNombres.max(Comparator.naturalOrder()).get();
long max = sNombres.max(compLong).get();
System.out.printf("%nmax séquentiel : max=%s, durée (ms)=%s%n", max, new Date().getTime() - début);
}
}
- рядок 29: маємо потік limite випадкових чисел типу Long;
- рядки 30–47: лямбда-змінна compLong реалізує інтерфейс Comparator<Long>. Цей інтерфейс зазвичай реалізується методом [Comparator.naturalOrder()] у рядку 49. Але тут ми хочемо відобразити потік виконання (рядки 31–33). Тому ми реалізуємо інтерфейс самостійно;
- рядок 50: пошук max;
Отримуємо такі результати:
 |
Якщо тепер замінити рядок 27 на такий (Приклад06e2):
Stream<Long> sNombres = nombres.stream().parallel();
отримаємо такі результати:
 |
Отже, паралельне виконання було повільнішим. Якщо перейти до 10 мільйонів чисел із verbose=false, отримаємо такі результати:
для послідовного виконання:
для паралельного виконання, яке, отже, залишається повільнішим.
Метод [Stream.min] використовується аналогічно:
// найлегша людина
affiche("poids min", personnes.stream().min(Comparator.comparingDouble(Personne::getPoids)).get());
5.6.7. [reduce]
// сумарний вік усіх осіб
affiche("âge total (reduce)", personnes.stream().map(p -> p.getAge()).reduce(0, (a1, a2) -> a1 + a2));
Метод [reduce] було представлено в розділі 5.3. У рядку 2 вище підсумовуються віки всіх осіб. Результат такий:
5.6.8. [sorted]
// особи за віком у порядку зростання
affiche("personnes par âge croissant",
personnes.stream().sorted(Comparator.comparingInt(Personne::getAge)).collect(Collectors.toList()));
// люди в алфавітному порядку за прізвищами
List<Personne> lPersonnes=personnes.stream().sorted((p1, p2) -> p1.getNom().compareTo(p2.getNom())).collect(Collectors.toList());
affiche("personnes par ordre alphabétique des noms", lPersonnes);
Метод [sorted] (рядки 3 і 5) має такий підпис:
 |
Метод [sorted] приймає як параметр тип [Comparator], описаний у розділі 5.6.6 для методів min та max. Він дозволяє сортувати об’єкт типу Stream за порядком, визначеним порівняльником, який передається йому як параметр. Ми бачили, що інтерфейс [Comparator] за замовчуванням надає кілька статичних методів, що реалізують поширені порівняльники, зокрема для чисел та символьних рядків. Тут ми використовуємо метод [Comparator.comparingInt], який приймає як параметр тип ToIntFunction — це функціональний інтерфейс методу [applyAsInt] із таким сигнатурою: int applyAsInt(T t). Тут фактичним параметром, переданим методу [Comparator.comparingInt] у рядку 3, є посилання на метод [Personne.age], який повертає вік людини.
Інтерфейс [Comparator] не надає статичних методів для порівняння рядків. У рядку 5 ми самостійно створюємо лямбду, що реалізує єдиний метод цього інтерфейсу: int compare(T t1, T t2)
(p1, p2) -> p1.getNom().compareTo(p2.getNom())
Ця лямбда порівнює імена осіб. Отримано такі результати:
Паралельне виконання сортування, здається, неможливе, як показує наступний код (Приклад06f1):
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.Random;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
public class Exemple06f1 {
// маппер jSON
static ObjectMapper jsonMapper = new ObjectMapper();
public static void main(String[] args) throws JsonProcessingException {
// дані
final long limite = 100L;
final boolean verbose = true;
// остаточний довгий ліміт = 10_000_000L;
// final boolean verbose = false;
// кількість процесорів
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// список чисел
long début = new Date().getTime();
List<Integer> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(new Random().nextInt(1000));
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// сортування чисел — послідовний метод
Stream<Integer> sNombres = nombres.stream();
début = new Date().getTime();
Comparator<Integer> compInt = (i1, i2) -> {
if (verbose) {
// потік
System.out.printf("[%s]", Thread.currentThread().getName());
}
// порівняння
int v1 = i1.intValue();
int v2 = i2.intValue();
if (v1 < v2) {
return +1;
} else {
if (v1 == v2) {
return 0;
} else {
return -1;
}
}
};
if (verbose) {
affiche("nombres", sNombres.sorted(compInt).collect(Collectors.toList()));
}
System.out.printf("tri séquentiel : durée (ms)=%s%n", new Date().getTime() - début);
}
public static <T> void affiche(String message, T value) throws JsonProcessingException {
System.out.println(String.format("%s ----", message));
System.out.println(jsonMapper.writeValueAsString(value));
}
}
- рядки 30–36: створюється потік випадкових чисел limite;
- у рядку 32 передаємо лямбду compInt (рядки 38–55) методу [sorted]. Ця лямбда сортує числа в порядку спадання та виводить на екран потік, який її виконує.
Отримано такі результати:
 |
Якщо в попередньому коді замінити рядок 36 на наступний (Приклад06f2):
Stream<Integer> sNombres = nombres.stream().parallel();
отримаємо такі результати:
 |
Виявляється, що, на диво, сортування потоку чисел відбулося за допомогою єдиного потоку. Паралелізму не було. Або ж я щось не врахував?
5.6.9. [anyMatch, noneMatch, allMatch]
// чи є люди віком понад 100 років?
affiche("des personnes de + de 100 ans (anyMatch)", personnes.stream().anyMatch(p -> p.getAge() > 100));
// чи всім людям не більше 100 років?
affiche("des personnes de + de 100 ans (noneMatch)", personnes.stream().noneMatch(p -> p.getAge() > 100));
// чи всі люди старші за 8 років
affiche("des personnes de + de 8 ans (allMatch)", personnes.stream().allMatch(p -> p.getAge() > 8));
У рядках 2, 4 та 6 методи [anyMatch, noneMatch, allMatch] мають як параметр тип Predicate, описаний у розділі 4.2. Отже, вони виконують фільтрацію. Усі три повертають булеве значення:
- anyMatch повертає true, якщо існує хоча б один елемент типу Stream, який відповідає критеріям фільтра;
- noneMatch перетворюється на true, якщо в Stream немає жодного елемента, що відповідає фільтру;
- allMatch повертає true, якщо всі елементи Stream відповідають фільтру;
Отримано такі результати:
5.6.10. [collect(Collectors.groupingBy)]
// людей групують за статтю
affiche("personnes regroupées par sexe", personnes.stream().collect(Collectors.groupingBy(p -> p.getSexe())));
Метод [collect] було представлено в розділі 5.6.3. Його параметром є реалізація інтерфейсу [Collector]. Клас [Collectors] надає низку статичних методів, що реалізують інтерфейс [Collector]. До цього моменту ми використовували метод [Collectors.toList()]. Тут ми використовуємо статичний метод [Collectors.groupingBy], який створює словник на основі Stream. Його сигнатура така:
 |
Метод [groupingBy] створює на основі типу Stream<T> тип Map<K,List<T>>. Ключ K надається параметром методу [groupingBy] типу Function<T,K>, єдиний метод якого має сигнатуру: K apply(T t). Якщо потрібно створити словник, індексований за статтю осіб, слід надати функцію, яка генерує стать на основі даних про особу. Тут ми передаємо як фактичний параметр методу [groupingBy] посилання на метод [Personne.getSexe]. Отримані результати такі:
У рядку 2 ми маємо рядок jSON із словника, індексованого двома ключами: HOMME та FEMME.
Паралельні обчислення не дають приросту продуктивності, як показує наступний приклад (Приклад06g1):
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
public class Exemple06g1 {
// mppeur jSON
static ObjectMapper jsonMapper = new ObjectMapper();
public static void main(String[] args) throws JsonProcessingException {
// дані
final long limite = 100L;
final boolean verbose = true;
// кінцевий довгий ліміт = 10_000_000L;
// final boolean verbose = false;
// кількість процесорів
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// список чисел
long début = new Date().getTime();
List<Integer> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(new Random().nextInt(1000));
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// групування чисел по сотнях — послідовний метод
Stream<Integer> sNombres = nombres.stream();
Function<Integer, Integer> groupByCent = n -> {
if (verbose) {
System.out.printf("[%s]", Thread.currentThread().getName());
}
return n / 100;
};
début = new Date().getTime();
// Map<Integer, List<Integer>> lNombres = sNombres.collect(Collectors.groupingBy(число -> число / 100));
Map<Integer, List<Integer>> lNombres = sNombres.collect(Collectors.groupingBy(groupByCent));
System.out.printf("%nregroupement séquentiel : durée (ms)=%s%n", new Date().getTime() - début);
// результати
if (verbose) {
affiche("nombres regroupés", lNombres);
}
}
public static <T> void affiche(String message, T value) throws JsonProcessingException {
System.out.println(String.format("%s ----", message));
System.out.println(jsonMapper.writeValueAsString(value));
}
}
- рядки 23–38: побудова потоку чисел limite;
- у рядку 47 числа згруповано по сотнях. Функцію лямбда у рядках 39–44 використовують для відображення потоку виконання;
Результати виконання такі:
 |
Якщо в коді замінити рядок 38 на наступний (Приклад06g2):
Stream<Integer> sNombres = nombres.stream().parallel();
отримуємо такі результати:
 |
Видно, що паралельне виконання групування погіршило продуктивність.
5.6.11. [distinct]
// видалення дублікатів елементів зі списку
affiche("distinct", Stream.of(1, 2, 1).distinct().collect(Collectors.toList()));
Метод [distinct] має такий сигнатур:
 |
Він дозволяє видалити дублікати з потоку. Метод [Stream.of] (рядок 2) має таку сигнатуру:
 |
Вона дозволяє створити Stream на основі явно вказаних значень. Результати виконання такі:
5.6.12. [flatMap]
// з Stream<Stream<T>> отримуємо Stream<T>
affiche("flatMap", Stream.of(1, 2, 3).flatMap(i -> Stream.of(i, i + 10)).collect(Collectors.toList()));
Метод [flatMap] має такий сигнатур:
  |
Метод [flatMap] приймає як параметр функцію, яка:
- приймає як параметр елемент типу T з Stream;
- повертає в результаті потік Stream<R>;
Якби замість методу [flatMap] було б використано метод [map], описаний у розділі 5.5, результатом був би тип Stream<Stream<R>>, де кожен елемент типу T початкового потоку породив би елемент Stream<R>. Чи дає метод [flatMap] тип Stream<R>? Він об’єднує (flatten) різні потоки Stream<R> в єдиний потік. Саме це показують результати виконання попереднього коду:
Існують спеціалізовані варіанти [flatMap]:
// з Stream<IntStream> створюємо IntStream, для якого обчислюємо суму
affiche("flatMapToInt", Stream.of(1, 2, 3).flatMapToInt(i -> IntStream.of(i, i + 10)).sum());
Метод [flatMapToInt] має такий сигнатур:
 |
Метод [flatMapToInt] приймає як параметр функцію, яка повертає наступний тип IntStream:
 |
IntStream є похідним від int. Цей тип є кращим за тип Stream<Integer>, оскільки його обробка дозволяє уникнути операцій boxing/unboxing між типами Integer та int. Цей інтерфейс успадковує багато методів типу Stream<T> та додає інші, зокрема згаданий вище метод [sum], який підсумовує елементи IntStream.
Наступний код ілюструє використання аналогічного методу [flatMapToDouble]:
// з Stream<DoubleStream> створюється DoubleStream, а потім масив
affiche("flatMapToDouble", Stream.of(1, 2, 3).flatMapToDouble(i -> DoubleStream.of(i, i * 1.2)).toArray());
Метод [DoubleStream.toArray] дозволяє перейти від типу DoubleStream до типу double[].
Результати для цих двох прикладів такі:
Наступний приклад демонструє підвищення продуктивності, досягнуте завдяки переходу від типу Stream<Long> до типу LongStream (Приклад06i1):
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
public class Exemple06i1 {
public static void main(String[] args) {
final long limite = 10_000_000L;
// кількість процесорів
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// список чисел
long début = new Date().getTime();
List<Long> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(i);
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// сума чисел — послідовний метод
début = new Date().getTime();
long somme = nombres.stream().reduce(0L, (s, i) -> s + i);
System.out.printf("somme séquentielle du Stream<Integer> : somme=%s, durée (ms)=%s%n", somme, new Date().getTime() - début);
}
}
- рядок 22: обчислення суми потоку чисел типу Long;
Отримуємо такі результати:
Тепер замінимо рядок 22 на такий (Приклад06i2):
long somme = nombres.stream().mapToLong(n -> n.longValue()).sum();
Метод Stream<Integer>.mapToLong дозволяє отримати потік типу LongStream з елементами примітивного типу long, який потім підсумовується за допомогою функції sum. У результаті отримуємо такі результати:
Приріст продуктивності є очевидним.
5.6.13. Методи потоку примітивних чисел
// максимальне значення потоку цілих чисел
affiche("IntStream max", IntStream.of(1, 10, 8).max());
// мінімальне значення з потоку double
affiche("DoubleStream min", DoubleStream.of(1.5, 10.4, 8.9).min());
// середнє значення потоку цілих чисел
affiche("IntStream average", IntStream.of(1, 10, 8).average().getAsDouble());
// статистика потоку цілих чисел
affiche("IntStream summaryStatistics", IntStream.of(1, 10, 8).summaryStatistics());
Потоки примітивних значень (int, long, double) надають методи, пристосовані до цих типів. Результат виконання попереднього коду такий:
- Результатом виконання рядка 2 коду є тип OptionalInt, аналогічний типу Optional<Integer>. Значення, що зберігається в цьому об’єкті, можна отримати за допомогою методу [getAsInt()]. Наявність значення можна перевірити за допомогою методу [isPresent()]. Рядок 2 результатів не означає, що клас [OptionalInt] має поля з іменами [asInt, present]. За замовчуванням бібліотека jSON використовує всі публічні методи getX та isY об’єкта, що серіалізується, у jSON. І тут дійсно є метод [getAsInt] та інший метод [isPresent], хоча самі поля [asInt, present] не існують;
- результатом виконання рядка 4 коду є тип OptionalDouble, аналогічний типу Optional<Double>;
- результатом 6-го рядка коду є тип OptionalDouble, значення якого можна отримати за допомогою методу [getAsDouble()]. Метод [average] обчислює середнє значення потоку чисел;
- результатом 8-го рядка коду є тип IntSummaryStatistics, який визначається наступним чином:
 |
Бачимо, що отриманий об’єкт IntSummaryStatistics надає різну інформацію про потік чисел, таку як кількість значень, суму, максимальне, мінімальне значення та середнє.