Stream Api

Stream Api - способ работать со структурами данных в функциональном стиле. Чаще всего с помощью stream в Java 8 работают с коллекциями, но на самом деле этот механизм может использоваться и для других задач. Не работает с примитивами за исключением int, long, double для которых есть отдельные Stream.

Виды операторов:

Intermediate(конвейерные, промежуточные) операторы

Конвейерные операторы - возвращают новый Stream и не выполняются без терминального оператора.

1. map(Function mapper) - преобразует каждый элемент стрима. Одно выходное значение из каждого входного.
2. flatMap(Function<T, Stream <R>> mapper) - создаёт произвольное количество выходных значений(0 или больше) для каждого входного. Под капотом из каждого элемента создается новый поток и в конце все потоки создают исходный поток.
3. filter(Predicate predicate) - фильтрует стрим, пропуская только те элементы, что проходят по условию.
4. dropWhile(Predicate predicate) — фильтрует стрим, пропуская только те элементы, что не проходят по условию.
5. takeWhile(Predicate predicate) — пропускает далее элементы пока условие выполняется и при первом не выполнении, больше никого не пропускает.
6. limit(long maxSize) – ограничивает стрим по количеству элементов.
7. skip(long n) – пропускаем n элементов.
8. sorted() – сортировка.
9. sorted(Comparator comparator) – сортирует стрим (сортировка как у TreeMap).
10. distinct() - убирает повторы элементов.

Terminal операторы

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

1. forEach(Consumer action) - аналог for each.
2. count() – возвращает количество элементов стрима.
3. reduce - позволяет выполнять агрегатные функции на всей коллекцией и возвращать один результат.
4. collect(Collector collector) – метод собирает все элементы в список, множество или другую коллекцию, сгруппировывает элементы по какому-нибудь критерию, объединяет всё в строку и т.д.
5. min(Comparator comparator)
6. max(Comparator comparator)
7. findFirst() - получить первый элемент стрима.
8. allMatch(Predicate predicate) - возвращает true, если все элементы стрима удовлетворяют условию или false.
9. anyMatch(Predicate predicate) - вернет true, если хотя бы один элемент стрима удовлетворяет условию predicate.
10. noneMatch(Predicate predicate) - вернёт true, если, пройдя все элементы стрима, ни один не удовлетворил условию.

Примеры создания Stream

• list.stream()
• map.entrySet().stream()
• Arrays.stream(array)
• Stream.of("1", "2", "3")

Примеры использования:

Проверка

var list = List.of("A", "B", "C");
var matchingElements = Stream.of("B", "C", "D")
        .anyMatch(list::contains);
        

Поиск совпадений элементов двух списков

var list = List.of("A", "B", "C");
var matchingElements = List.of("B", "C", "D")
    .stream()
    .filter(list::contains)
    .toList();
        

Поиск не совпадений элементов двух списков

var list = List.of("A", "B", "C");
var matchingElements = List.of("B", "C", "D")
    .stream()
    .filter(element -> !list.contains(element))
    .toList();
        

Нахождения процентиля(персентиль)

var latencies = Stream.of(10, 3, 6, 7, 8, 8, 9, 13, 15, 16, 20).toList();
System.out.println(percentile(latencies, 25));
System.out.println(percentile(latencies, 50));
System.out.println(percentile(latencies, 75));
System.out.println(percentile(latencies, 100));

public static long percentile(List<Integer> latencies, double percentile) {
    int index = (int) Math.ceil(percentile / 100.0 * latencies.size());
    return latencies.get(index-1);
}
        

Нахождения процентиля(персентиль)

var latencies = Stream.of(10, 3, 6, 7, 8, 8, 9, 13, 15, 16, 20).toList();
System.out.println(percentile(latencies, 25));
System.out.println(percentile(latencies, 50));
System.out.println(percentile(latencies, 75));
System.out.println(percentile(latencies, 100));

public static long percentile(List<Integer> latencies, double percentile) {
    int index = (int) Math.ceil(percentile / 100.0 * latencies.size());
    return latencies.get(index-1);
}
        

List to Map

var list = Stream.of("1", "2", "3").toList();
var map = list.stream().collect(Collectors.toMap(i -> i, i -> "sample"));
        

Parallel stream

Parallel stream - потоки данных которые обрабатываются параллельно на нескольких потоках, чтобы ускорить выполнение коллекций.

Выполняется в ForkJoinPool потоках.

Как создать

List<Integer> numbers = List.of(1, 2, 3, 4, 5, 6);

// Последовательный поток
int sum1 = numbers.stream().mapToInt(Integer::intValue).sum();

// Параллельный поток
int sum2 = numbers.parallelStream().mapToInt(Integer::intValue).sum();
        

Как это работает:

1. Коллекция разбивается на подзадачи.
2. Каждая подзадача исполняется в потоках ForkJoinPool.commonPool() - по умолчанию количество потоков равное, количеству ядер процессора.
3. Результаты собираются и объединяются.

Преимущество

• Ускоряет обработку больших коллекций на многоядерных CPU.
• Легко применяется к обычным операциям Stream API(map, filter, reduce).

Ограничения и риски:

• Не всегда быстро - маленькие коллекции могут выполняться быстрее не параллельно, из-за затрат на парализацию.
• Не используйте если порядок обработки важен.
• Нельзя безопасно обновлять переменные вне потоков без атомарных структур или синхронизации.