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

Что такое Collector и какие есть встроенные Collectors

Утилитный класс java.util.stream.Collectors предоставляет богатый набор готовых фабричных методов: 4. Классификация и разбиение: groupingBy(...), partitioningBy(...). 5. Компози...

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

🟢 Junior Level

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

Collector<T, A, R> — это спецификация (рецепт) для терминальной операции collect(), определяющая, как накапливать элементы потока типа T в промежуточный аккумулятор типа A и преобразовать его в итоговый результат типа R.

Утилитный класс java.util.stream.Collectors предоставляет богатый набор готовых фабричных методов:

  1. Сбор в базовые коллекции: toList(), toSet(), toCollection(...), toMap(...), toUnmodifiableList().
  2. Агрегация и редукция: counting(), summingInt/Long/Double(), averagingInt(), summarizingInt(), minBy(), maxBy().
  3. Объединение строк: joining(), joining(", "), joining(", ", "[", "]").
  4. Классификация и разбиение: groupingBy(...), partitioningBy(...).
  5. Композиция: collectingAndThen(...), mapping(...), filtering(...), flatMapping(...), teeing(...) (Java 12+).

Базовые примеры на Java 21

import java.util.*;
import java.util.stream.Collectors;

public class BuiltInCollectorsDemo {

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

    public static void main(String[] args) {
        List<Product> products = List.of(
            new Product("Laptop", "Electronics", 1200.0, 5),
            new Product("Smartphone", "Electronics", 800.0, 15),
            new Product("Desk", "Furniture", 350.0, 3),
            new Product("Chair", "Furniture", 150.0, 0),
            new Product("Pen", "Stationery", 2.5, 100)
        );

        // 1. Сбор в неизменяемое множество уникальных категорий
        Set<String> categories = products.stream()
            .map(Product::category)
            .collect(Collectors.toUnmodifiableSet());

        // 2. Объединение названий товаров в форматированную строку
        String catalog = products.stream()
            .map(Product::name)
            .collect(Collectors.joining(", ", "Каталог: [", "]"));
        System.out.println(catalog);
        // Каталог: [Laptop, Smartphone, Desk, Chair, Pen]

        // 3. Подсчёт суммарной стоимости и количества через summarizingDouble
        DoubleSummaryStatistics stats = products.stream()
            .collect(Collectors.summarizingDouble(Product::price));
        System.out.printf("Товаров: %d, Средняя цена: %.2f, Макс: %.2f%n",
            stats.getCount(), stats.getAverage(), stats.getMax());

        // 4. Простое разделение (partitioning) по наличию на складе
        Map<Boolean, List<Product>> inStockPartition = products.stream()
            .collect(Collectors.partitioningBy(p -> p.stock() > 0));
        System.out.println("В наличии: " + inStockPartition.get(true).size());
        System.out.println("Нет на складе: " + inStockPartition.get(false).size());
    }
}

🟡 Middle Level

Анатомия интерфейса Collector<T, A, R>

Интерфейс Collector параметризован тремя типами:

  • T — тип входных элементов стрима;
  • A — промежуточный тип мутабельного контейнера (аккумулятора);
  • R — итоговый тип возвращаемого значения (может совпадать с A).

Он описывает четыре функции и набор характеристик:

public interface Collector<T, A, R> {
    Supplier<A> supplier();             // Создание нового пустого аккумулятора
    BiConsumer<A, T> accumulator();     // Добавление элемента в аккумулятор
    BinaryOperator<A> combiner();       // Слияние двух аккумуляторов (для параллельных стримов)
    Function<A, R> finisher();          // Финальное преобразование аккумулятора в результат
    Set<Characteristics> characteristics(); // Флаги оптимизации рантайма
}

Характеристики (Characteristics):

  • IDENTITY_FINISH: указывает, что функция finisher() является тождественной (Function.identity()). Рантайм стрима пропускает вызов finisher() и приводит тип (R) accumulator напрямую через unchecked cast, избегая лишних аллокаций.
  • UNORDERED: операция редукции не сохраняет порядок элементов исходного стрима (важно для оптимизации параллельного сбора).
  • CONCURRENT: функция accumulator() потокобезопасна и может вызываться одновременно из нескольких параллельных потоков на одном общем контейнере A. Требует отсутствия combiner() или его игнорирования. Должна комбинироваться с UNORDERED (если стрим не упорядочен).

Продвинутые встроенные коллекторы и композиция

Коллекторы поддерживают древовидную композицию с помощью так называемых downstream-коллекторов:

import java.util.*;
import java.util.stream.Collectors;

public class AdvancedCollectorsDemo {

    public record Employee(String name, String department, double salary, List<String> skills) {}

    public static void main(String[] args) {
        List<Employee> staff = List.of(
            new Employee("Alice", "Engineering", 120_000, List.of("Java", "Docker")),
            new Employee("Bob", "Engineering", 95_000, List.of("Java", "Kubernetes")),
            new Employee("Charlie", "HR", 70_000, List.of("Recruiting")),
            new Employee("Diana", "HR", 80_000, List.of("Talent", "Communication"))
        );

        // 1. Двухуровневая группировка: Отдел -> Средняя зарплата
        Map<String, Double> avgSalaryByDept = staff.stream()
            .collect(Collectors.groupingBy(
                Employee::department,
                Collectors.averagingDouble(Employee::salary)
            ));

        // 2. Группировка + маппинг + сбор во вложенное множество: Отдел -> Set<Имена>
        Map<String, Set<String>> namesByDept = staff.stream()
            .collect(Collectors.groupingBy(
                Employee::department,
                Collectors.mapping(Employee::name, Collectors.toSet())
            ));

        // 3. Java 9+: filtering и flatMapping внутри downstream
        Map<String, Set<String>> uniqueSkillsByDept = staff.stream()
            .collect(Collectors.groupingBy(
                Employee::department,
                Collectors.flatMapping(e -> e.skills().stream(), Collectors.toSet())
            ));

        // 4. Java 12+: Collectors.teeing() — объединение результатов двух независимых коллекторов
        record SalaryStats(double min, double max, double diff) {}
        SalaryStats salarySpread = staff.stream()
            .collect(Collectors.teeing(
                Collectors.minBy(Comparator.comparingDouble(Employee::salary)),
                Collectors.maxBy(Comparator.comparingDouble(Employee::salary)),
                (minOpt, maxOpt) -> new SalaryStats(
                    minOpt.map(Employee::salary).orElse(0.0),
                    maxOpt.map(Employee::salary).orElse(0.0),
                    maxOpt.map(Employee::salary).orElse(0.0) - minOpt.map(Employee::salary).orElse(0.0)
                )
            ));

        System.out.println("Разброс зарплат: " + salarySpread);
    }
}

🔴 Senior Level

Создание собственного Collector под капотом

Статический метод Collector.of(...) позволяет конструировать кастомные коллекторы без создания отдельных классов.

Рассмотрим практическую задачу: сбор элементов стрима в фиксированные пакеты (батчи) заданного размера List<List<T>>.

import java.util.*;
import java.util.stream.Collector;
import java.util.stream.IntStream;

public final class BatchCollector<T> implements Collector<T, List<List<T>>, List<List<T>>> {

    private final int batchSize;

    public BatchCollector(int batchSize) {
        if (batchSize <= 0) {
            throw new IllegalArgumentException("Размер пачки должен быть > 0");
        }
        this.batchSize = batchSize;
    }

    public static <T> Collector<T, ?, List<List<T>>> toBatches(int batchSize) {
        return new BatchCollector<>(batchSize);
    }

    @Override
    public java.util.function.Supplier<List<List<T>>> supplier() {
        return ArrayList::new;
    }

    @Override
    public java.util.function.BiConsumer<List<List<T>>, T> accumulator() {
        return (batches, item) -> {
            if (batches.isEmpty() || batches.get(batches.size() - 1).size() >= batchSize) {
                List<T> newBatch = new ArrayList<>(batchSize);
                newBatch.add(item);
                batches.add(newBatch);
            } else {
                batches.get(batches.size() - 1).add(item);
            }
        };
    }

    @Override
    public java.util.function.BinaryOperator<List<List<T>>> combiner() {
        return (left, right) -> {
            if (left.isEmpty()) return right;
            if (right.isEmpty()) return left;

            List<List<T>> merged = new ArrayList<>(left);
            for (List<T> batch : right) {
                for (T item : batch) {
                    if (merged.isEmpty() || merged.get(merged.size() - 1).size() >= batchSize) {
                        List<T> newBatch = new ArrayList<>(batchSize);
                        newBatch.add(item);
                        merged.add(newBatch);
                    } else {
                        merged.get(merged.size() - 1).add(item);
                    }
                }
            }
            return merged;
        };
    }

    @Override
    public java.util.function.Function<List<List<T>>, List<List<T>>> finisher() {
        return Collections::unmodifiableList;
    }

    @Override
    public Set<Characteristics> characteristics() {
        return Collections.emptySet(); // Требует порядка, finisher не identity, не concurrent
    }

    public static void main(String[] args) {
        List<List<Integer>> batches = IntStream.rangeClosed(1, 10)
            .boxed()
            .collect(BatchCollector.toBatches(3));

        System.out.println(batches); 
        // [[1, 2, 3], [4, 5, 6], [7, 8, 9], [10]]
    }
}

Механика параллельного сбора: Concurrent vs Tree-Fold Merge

При параллельном исполнении стрима (parallelStream()) рантайм ReduceOps выбирает одну из двух фундаментальных стратегий:

1. Стандартная редукция (Divide and Conquer Merge):
   Поток 1 (Chunk 1) ---> Accumulator 1 \
                                          ---> Combiner(Acc1, Acc2) \
   Поток 2 (Chunk 2) ---> Accumulator 2 /                            ---> Finisher ---> Result
                                                                    /
   Поток 3 (Chunk 3) ---> Accumulator 3 \                          /
                                          ---> Combiner(Acc3, Acc4)
   Поток 4 (Chunk 4) ---> Accumulator 4 /

2. Конкурентная редукция (CONCURRENT + UNORDERED):
   Поток 1 ----\
   Поток 2 -------> Единый потокобезопасный Accumulator (ConcurrentMap) ---> Finisher ---> Result
   Поток 3 ----/
   Поток 4 --/
   (Combiner не вызывается вовсе; блокировки минимизированы CAS-операциями)
  1. Редукция методом слияния поддеревьев (Fold-Tree):
    • Каждый поток в ForkJoinPool создает собственный независимый экземпляр аккумулятора через supplier().
    • Накапливает элементы через accumulator() без всякой синхронизации.
    • По завершении веток задачи объединяются попарно вызовами combiner().
    • Плюсы: Нет конкуренции за память, нет блокировок, гарантируется детерминированный порядок сборки (если источник упорядочен).
    • Минусы: Дополнительные аллокации памяти на создание промежуточных коллекций и $O(N)$ накладные расходы на копирование при их слиянии в combiner().
  2. Конкурентная редукция (Concurrent Reduction):
    • Активируется только если коллектор имеет флаг Characteristics.CONCURRENT И (стрим неупорядочен UNORDERED или коллектор имеет флаг UNORDERED).
    • supplier() вызывается ровно один раз для всего параллельного стрима.
    • Все рабочие потоки одновременно вызывают accumulator() над одним общим экземпляром (например, ConcurrentHashMap).
    • Метод combiner() вообще не вызывается.
    • Пример: Collectors.toConcurrentMap() или Collectors.groupingByConcurrent().

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

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

«Collector — это спецификация процесса мутабельной редукции для метода collect(). Он инкапсулирует 4 функции: создание контейнера (supplier), добавление элемента (accumulator), слияние контейнеров при параллельном вычислении (combiner) и финальное преобразование (finisher), а также флаги characteristics (CONCURRENT, UNORDERED, IDENTITY_FINISH). Основные встроенные коллекторы сосредоточены в классе Collectors (toList, toSet, toMap, groupingBy, partitioningBy, joining, teeing). Для параллельных стримов критично различать сборку через слияние деревьев (по умолчанию) и конкурентную сборку в единый потокобезопасный контейнер при флаге CONCURRENT.»

Шпаргалка характеристик Collector

Характеристика Что означает Поведение рантайма Пример коллектора
IDENTITY_FINISH A == R, финальное преобразование не требуется finisher() не вызывается, делается каст (R) a toList(), toSet(), toCollection()
UNORDERED Результат не зависит от порядка обхода элементов Стриму разрешается игнорировать порядок в Spliterator toSet(), groupingByConcurrent()
CONCURRENT accumulator потокобезопасен для многопоточного вызова Один общий аккумулятор на весь параллельный стрим, без combiner toConcurrentMap(), groupingByConcurrent()

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

1. В чём разница между Collectors.toList(), Collectors.toUnmodifiableList() и Stream.toList() (Java 16+)?

  • Collectors.toList(): возвращает реализацию List без гарантий изменяемости, типа или потокобезопасности (исторически обычно ArrayList, но спецификация позволяет JDK вернуть любой List). Допускает null.
  • Collectors.toUnmodifiableList() (Java 10+): возвращает неизменяемый List (через Collections.unmodifiableList или компактную внутреннюю реализацию). Запрещает элементы null (выбрасывает NullPointerException).
  • Stream.toList() (Java 16+): компактный терминальный метод стрима. Не аллоцирует объект Collector, собирает элементы напрямую во внутренний компактный неизменяемый массив java.util.ImmutableCollections.ListN/List12. В отличие от toUnmodifiableList(), разрешает элементы null и потребляет меньше памяти на аллокацию промежуточных структур.

2. Как работает Collectors.teeing() (Java 12+) под капотом: делает ли он два прохода по стриму?

  • Нет, стрим проходит ровно один раз.
  • teeing(c1, c2, merger) создает составной аккумулятор, хранящий внутри состояния аккумуляторов обоих коллекторов A1 и A2.
  • Для каждого элемента стрима T последовательно вызываются c1.accumulator().accept(a1, item) и c2.accumulator().accept(a2, item).
  • В конце вызываются оба finisher(), и их результаты объединяются с помощью предоставленного BiFunction merger.

3. Что произойдёт при дублировании ключей в Collectors.toMap(keyMapper, valueMapper)?

  • Будет выброшено исключение IllegalStateException: Duplicate key <key> (attempted merging values <v1> and <v2>).
  • Чтобы предотвратить исключение, необходимо передавать перегрузку с тремя аргументами: Collectors.toMap(keyMapper, valueMapper, (existing, replacement) -> replacement). Если требуется конкретная реализация Map (например, TreeMap), используется перегрузка с четырьмя аргументами Collectors.toMap(keyMapper, valueMapper, mergeFunction, TreeMap::new).

4. Почему Collectors.groupingByConcurrent() в parallelStream может работать медленнее, чем обычный Collectors.groupingBy()?

  • groupingByConcurrent() использует единый общий ConcurrentHashMap. При огромном количестве элементов и высокой конкуренции потоков запись в бакеты мапы натыкается на конкуренцию кэш-линий процессора (False Sharing) и блокировки узлов/CAS-циклы.
  • Обычный groupingBy() создает независимые HashMap в каждом потоке (вообще без синхронизации и конфликтов кэшей), а в конце быстро мерджит их в дереве Fork/Join. Если размер групп невелик или число потоков велико, раздельная сборка без конкуренции часто выигрывает у общего ConcurrentHashMap.

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

  • 🚩 Использование Collectors.toList() с расчетом на его неизменяемость (или наоборот, предположение, что он всегда возвращает изменяемый ArrayList).
  • 🚩 Забывание merge-функции в Collectors.toMap() при обработке входных данных из внешних API/БД, что приводит к внезапным IllegalStateException на проде.
  • 🚩 Мутация внешних переменных внутри кастомного коллектора вместо использования собственного аккумулятора.
  • 🚩 Реализация флага Characteristics.CONCURRENT на аккумуляторе, который не является потокобезопасным (например, на обычном ArrayList). Это приведёт к гонкам данных, потере элементов и ArrayIndexOutOfBoundsException в параллельных стримах.

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