🌊 Розділ 8 · Питання #6

Що таке Collector і які є вбудовані Collectors

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

🌊 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 у паралельних стрімах.

Пов’язані теми