Краткий ответ

Корректное объединение частичных контейнеров определяет функция combiner в составе Collector. Она принимает два промежуточных результата, объединяет их в один и возвращает итоговый контейнер того же типа.

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

Исторический контекст

Stream API отделяет описание обработки данных от способа её выполнения. Одна и та же pipeline может обрабатываться последовательно или разбиваться на независимые части для параллельного выполнения.

Чтобы поддержать такой подход, Collector описывает не только добавление отдельного элемента в результат, но и объединение уже накопленных частичных результатов. Это решает исходную проблему параллельной редукции: потоки не должны постоянно синхронно писать в один общий контейнер.

Постановка проблемы

При параллельном stream источник делится на части. Каждая часть обычно обрабатывается в собственном контейнере, после чего эти контейнеры нужно слить в один результат.

Если объединение реализовано неверно, возможны потеря элементов, дублирование, нарушение порядка, повреждение внутреннего состояния контейнера или результат, зависящий от числа потоков. Ошибка может не проявляться на маленьких наборах данных и становиться заметной только при изменении размера входа или степени параллелизма.

Подробное решение

Collector концептуально задаёт несколько операций:

supplier создаёт новый пустой промежуточный контейнер;
accumulator добавляет один элемент в контейнер;
combiner объединяет два частичных контейнера;
finisher при необходимости преобразует промежуточный результат в окончательный;
characteristics сообщают свойства коллектора, например отсутствие отдельной финальной трансформации или допустимость конкурентного накопления.

В параллельном выполнении разные задачи вызывают supplier и accumulator независимо. Затем runtime организует дерево объединений и вызывает combiner на парах частичных результатов. Поэтому combiner должен корректно работать не только для двух контейнеров, но и при многократном вложенном объединении.

Обычно требуется ассоциативность операции объединения: результат объединения контейнеров A и B, затем C, должен быть эквивалентен объединению A с результатом B и C. Если важен порядок элементов, combiner обязан сохранять его в соответствии с контрактом коллектора и характеристиками источника.

Combiner может изменить один из переданных контейнеров и вернуть его, если это безопасно и согласовано с контрактом. Он также может создать новый контейнер, но это обычно требует дополнительных аллокаций и копирования. Нельзя рассчитывать, что combiner будет вызван только один раз или что конкретный контейнер всегда окажется первым.

Минимальный пример коллектора, собирающего числа в список:

Collector<Integer, List<Integer>, List<Integer>> collector = Collector.of(
    ArrayList::new,
    List::add,
    (left, right) -> {
        left.addAll(right);
        return left;
    }
);

List<Integer> result = numbers.parallelStream().collect(collector);

Здесь accumulator добавляет отдельное число, а combiner переносит элементы из правого списка в левый. Такой combiner не делает общий список потокобезопасным: безопасность достигается тем, что частичные контейнеры используются независимо до этапа объединения.

Для коллектора с упорядоченным результатом одного корректного combiner недостаточно: порядок должен поддерживаться всей реализацией. Характеристика CONCURRENT означает возможность конкурентного накопления при выполнении условий контракта, но не превращает произвольный контейнер в потокобезопасный. Параллельный stream также не гарантирует ускорение: стоимость разбиения, синхронизации, объединения и передачи данных может превысить выгоду от нескольких потоков.

Ситуация из практики

Сервис обрабатывает большой поток событий и должен получить для каждого клиента набор уникальных кодов. Для каждого частичного результата можно создать собственную карту, где значение — множество кодов. Accumulator добавляет код в множество соответствующего клиента, а combiner объединяет карты и сливает множества по совпадающим ключам.

Рассматривались три варианта. Общая синхронизированная карта была простой, но создавала конкуренцию между потоками и фактически ограничивала параллелизм. Сбор всех событий в один список с последующей группировкой упрощал реализацию, но увеличивал память и откладывал основную работу до конца pipeline. Пользовательский collector с локальными картами и явным combiner требовал аккуратного тестирования, зато минимизировал совместное состояние.

Выбран третий вариант: каждый поток накапливал свою карту, а combiner объединял карты и множества. Важно было проверить ассоциативность объединения, отсутствие изменения уже недоступных контейнеров и корректность результата при разных размерах входа. Решение снизило конкуренцию, но параллельный режим оставили только для достаточно больших потоков событий, поскольку на малых наборах расходы на разбиение и слияние были выше выигрыша.

Что кандидаты часто упускают

1. Может ли combiner просто вернуть один из двух контейнеров, отбросив другой?

Нет, если отброшенный контейнер содержит значимые данные. Combiner обязан сохранить семантику последовательного накопления: каждый элемент входного stream должен быть представлен в итоговом результате согласно контракту коллектора. Исключение возможно только для специальной операции, где данные действительно намеренно отбрасываются, например при корректной реализации поиска экстремума, но и тогда выбранное значение должно учитывать оба частичных результата.

2. Должен ли combiner быть потокобезопасным?

Не обязательно в смысле одновременного доступа к одним и тем же контейнерам. Обычно разные частичные контейнеры принадлежат разным задачам, а runtime вызывает combiner так, чтобы конкретное объединение не выполнялось конкурентно над одним и тем же изменяемым контейнером.

Однако combiner всё равно должен быть безопасным относительно жизненного цикла коллектора: нельзя использовать внешний общий контейнер или состояние, которое меняется другими потоками без предусмотренной синхронизации. Потокобезопасность внутреннего контейнера и корректность алгоритма объединения — разные свойства.

3. Почему корректный combiner может сделать параллельный collector медленнее последовательного?

При каждом объединении нужно выполнить дополнительную работу: обойти элементы, слить структуры, иногда выделить память и скопировать данные. Для списка операция добавления всех элементов из правого контейнера может привести к большому объёму копирований на уровнях дерева объединения.

Кроме того, параллельный режим платит за разбиение источника, планирование задач и синхронизацию инфраструктуры. Поэтому параллельный collector оправдан, когда обработка элементов или объединение частичных результатов достаточно затратны, а источник эффективно делится. На небольших данных последовательный stream часто выигрывает именно из-за меньших накладных расходов.