🌊 Раздел 8 · Вопрос #23

Что делают операции distinct(), sorted(), limit(), skip()

Все четыре метода являются промежуточными (intermediate) операциями с состоянием (Stateful), которые изменяют состав, порядок или количество элементов в потоке:

🌊 Stream API 📄 Исходный Markdown

🟢 Junior Level

Краткий ответ для собеседования (30 секунд)

Все четыре метода являются промежуточными (intermediate) операциями с состоянием (Stateful), которые изменяют состав, порядок или количество элементов в потоке:

  • distinct() — отфильтровывает дубликаты, оставляя только уникальные элементы (сравнение идет по контракту equals() и hashCode()).
  • sorted() — сортирует элементы в естественном порядке (Comparable) или с помощью переданного Comparator.
  • limit(long maxSize) — усекает поток, пропуская дальше не более первых maxSize элементов (операция короткого замыкания).
  • skip(long n) — отбрасывает первые n элементов потока, пропуская вниз по конвейеру оставшиеся.

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

import java.util.Comparator;
import java.util.List;

public class StatefulOpsDemo {

    public record Product(String name, double price) {}

    public static void main(String[] args) {
        List<Integer> numbers = List.of(5, 2, 8, 2, 9, 1, 5, 8, 3);

        // 1. Комбинирование distinct, sorted, skip и limit
        List<Integer> result = numbers.stream()
            .distinct()            // [5, 2, 8, 9, 1, 3] (удалены дубликаты)
            .sorted()              // [1, 2, 3, 5, 8, 9] (отсортированы по возрастанию)
            .skip(1)               // [2, 3, 5, 8, 9]    (пропустили минимальный элемент)
            .limit(3)              // [2, 3, 5]          (взяли следующие 3 элемента)
            .toList();

        System.out.println("Результат обработки чисел: " + result);

        // 2. Сортировка объектов с Comparator
        List<Product> products = List.of(
            new Product("Клавиатура", 3500.0),
            new Product("Мышь", 1800.0),
            new Product("Монитор", 24000.0)
        );

        List<Product> cheapToExpensive = products.stream()
            .sorted(Comparator.comparingDouble(Product::price))
            .toList();

        System.out.println("Сортировка по цене: " + cheapToExpensive);
    }
}

🟡 Middle Level

Разделение на Stateless и Stateful операции

В архитектуре Stream API промежуточные операции фундаментально делятся на два типа:

Stateless (map, filter):
Элемент 1 ──► [Filter] ──► [Map] ──► Downstream (потоковая передача без буферов)

Stateful (sorted, distinct):
Элемент 1 ──┐
Элемент 2 ──┼──► [ Внутренний буфер памяти ] ──► (Барьер: ожидание конца стрима)
Элемент 3 ──┘
  1. Stateless (без состояния): filter, map, flatMap, peek. Для каждого элемента решение принимается на лету. Память не расходуется ($O(1)$).
  2. Stateful (с состоянием):
    • distinct(): сохраняет состояние уже увиденных элементов во внутреннем хеш-множестве (LinkedHashSet / ConcurrentHashMap). Требует $O(N)$ дополнительной памяти.
    • sorted(): выступает полным барьером конвейера (Pipeline Barrier). Метод не может отдать ни одного элемента downstream, пока не вычитает абсолютно все элементы источника в массив и не отсортирует его (через TimSort / Dual-Pivot Quicksort).
    • limit(n) / skip(n): сохраняют счетчик обработанных элементов и поддерживают короткое замыкание.

Оптимизации Stream Flags (StreamOpFlag)

Если источник данных уже обладает необходимыми флагами, рантайм Stream API пропускает операции без выполнения (no-op):

import java.util.List;
import java.util.TreeSet;

public class StreamFlagsOptimizationDemo {

    public static void main(String[] args) {
        TreeSet<String> sortedSet = new TreeSet<>(List.of("C", "A", "B"));

        // TreeSet изначально обладает флагами DISTINCT и SORTED в Spliterator:
        List<String> list = sortedSet.stream()
            .distinct() // NO-OP: JVM пропускает операцию (коллекция уже Set)
            .sorted()   // NO-OP: JVM пропускает операцию (коллекция уже отсортирована)
            .toList();

        System.out.println(list); // [A, B, C]
    }
}

🔴 Senior Level

3 критические ловушки производительности в Production

1. Антипаттерн «Top-N через sorted().limit(K)»

Классическая ошибка — сортировка миллиона объектов ради топ-5:

// ❌ ПЛОХО: Сортирует ВСЕ 1 000 000 элементов со сложностью $O(N \log N)$
List<User> top5 = users.stream()
    .sorted(Comparator.comparing(User::getRating).reversed())
    .limit(5)
    .toList();

Senior-решение: Для малых $K$ ($K \ll N$) использовать PriorityQueue (Min-Heap) фиксированного размера $K$. Сложность снижается с $O(N \log N)$ до $O(N \log K)$, а потребление памяти — с $O(N)$ до $O(K)$.

2. Зависание на бесконечных стримах (Infinite Stream Hang)

// 💥 Катастрофа: OutOfMemoryError / зависание потока!
Stream.iterate(0, i -> i + 1)
    .sorted()  // Барьер зависнет навсегда, пытаясь буферизовать бесконечность!
    .limit(10)
    .toList();

// ✅ ПРАВИЛЬНО: limit() ОБЯЗАН стоять ДО барьера sorted():
Stream.iterate(0, i -> i + 1)
    .limit(10) // Делает стрим конечным!
    .sorted()
    .toList();

3. Пагинация через skip(N).limit(M)

Использование stream.skip(100_000).limit(20) для пагинации в коде вынуждает стрим поштучно протянуть 100 000 элементов через все фильтры. Пагинация должна выполняться на уровне СУБД (LIMIT/OFFSET или Keyset Pagination).

Поведение в parallelStream()

В параллельном режиме stateful-операции теряют эффективность:

  • distinct() в parallel: требует синхронизации через ConcurrentHashMap или дорогого слияния частичных сетов в combiner.
  • sorted() в parallel: использует Arrays.parallelSort(), который дает выигрыш только на объемах от $100\,000$ элементов.
  • limit() и skip() в parallel: если источник упорядочен (ORDERED), потоки вынуждены буферизовать результаты и координироваться для соблюдения порядка. Вызов .unordered() перед limit() снимает эту блокировку, позволяя вернуть любые первые $N$ готовых элементов.

🎯 Шпаргалка для интервью

30-секундный ответ

«distinct(), sorted(), limit() и skip() — это промежуточные stateful-операции. В отличие от stateless-операций (map, filter), они требуют сохранения внутреннего состояния. distinct() использует equals/hashCode и требует $O(N)$ памяти под хеш-таблицу. sorted() — полный барьер конвейера, накапливающий все элементы перед сортировкой; вызов sorted() на бесконечном стриме без предшествующего limit() приводит к OutOfMemoryError. В parallelStream для ускорения limit() на упорядоченных коллекциях рекомендуется вызывать .unordered().»

Матрица характеристик операций

Операция Сложность по времени Память Является барьером? Влияние на бесконечный стрим
distinct() $O(N)$ $O(N)$ (хеш-таблица) Нет (отдает по мере нахождения) Может зависнуть при поиске уникальных
sorted() $O(N \log N)$ $O(N)$ (массив элементов) Да (полный барьер) Зависает / OOM (нельзя вызывать)
limit(n) $O(1)$ per item $O(1)$ Нет (короткое замыкание) Делает бесконечный стрим конечным
skip(n) $O(1)$ per item $O(1)$ Нет Пропускает первые $n$ элементов

Топ-4 каверзных вопроса на интервью

1. Почему sorted() называют барьером конвейера (pipeline barrier)?

  • Потому что алгоритм сортировки не может определить, какой элемент должен быть первым, пока не сравнит его со всеми остальными элементами стрима. Поэтому sorted() полностью останавливает вертикальное прохождение элементов (Loop Fusion), накапливает все элементы источника во внутренний массив, сортирует его и только затем начинает передавать элементы следующему узлу.

2. Что произойдет при вызове Stream.generate(Math::random).sorted().limit(5)?

  • Программа зависнет на 100% загрузке CPU и завершится с OutOfMemoryError. Операция sorted() попытается вычитать бесконечный стрим до конца перед тем, как отдать хоть один элемент в limit(5). Если же поменять порядок на .limit(5).sorted(), код отработает мгновенно, так как limit сделает стрим конечным до попадания в барьер сортировки.

3. Как distinct() определяет равенство элементов и какие требования предъявляются к объектам?

  • distinct() использует стандартные методы equals() и hashCode(), сохраняя уникальные элементы во внутреннем наборе. Если у класса не переопределены equals и hashCode, сравнение будет происходить по ссылке (==), и логически одинаковые объекты не будут отфильтрованы как дубликаты.

4. Зачем вызывать .unordered() перед .limit() в параллельном стриме?

  • Если стрим упорядочен (ORDERED), limit(n) обязан вернуть ровно те элементы, которые шли первыми в исходной коллекции. Это заставляет параллельные ветви ForkJoinPool синхронизироваться и сбрасывать результаты правых ветвей. Вызов .unordered() разрешает стриму вернуть любые $n$ элементов, которые первыми успели обработать параллельные потоки, что в разы ускоряет выполнение.

Типичные ошибки и Red Flags

  • 🚩 Вызов sorted() на бесконечном стриме до вызова limit().
  • 🚩 Использование sorted().limit(K) для нахождения топ-3 элементов на миллионных коллекциях вместо PriorityQueue.
  • 🚩 Использование distinct() для классов с некорректным или отсутствующим hashCode() и equals().
  • 🚩 Попытка использовать stream.skip(offset).limit(pageSize) для пагинации больших таблиц БД на уровне бэкенда вместо SQL LIMIT/OFFSET.

Связанные темы