사용자 정의 Collector와 Spliterator

1. 사용자 정의 Collector: 언제, 어떻게 작성할까

Java Stream API에서는 스트림을 컬렉션이나 집계 결과로 변환할 때 Collector 인터페이스를 사용합니다. 보통은 Collectors 클래스의 준비된 컬렉터(toList(), toMap(), groupingBy() 등)를 사용하지만, 가끔은 특별한 요구가 있을 수 있고 — 그럴 때는 직접 Collector를 작성할 수 있습니다.

Collector는 스트림의 요소로부터 최종 결과를 어떻게 수집할지 정의하는 객체입니다. 네 가지(사실은 다섯 가지) 핵심 구성 요소를 규정합니다:

• supplier — 요소를 수집할 새 컨테이너를 생성합니다(예: 새 리스트나 맵).
• accumulator — 다음 요소를 컨테이너에 추가합니다.
• combiner — 두 컨테이너를 병합합니다(병렬 스트림에서 중요!).
• finisher — 컨테이너를 최종 결과로 변환합니다(예: 불변으로 만들거나 다른 타입으로 변환).
• characteristics — 컬렉터의 속성을 나타내는 플래그 집합입니다(예: 병렬 처리 지원 여부, 결과 타입 변경 여부 등).

시그니처:

Collector<T, A, R>

• T — 스트림 요소의 타입,
• A — 중간 누산기 타입,
• R — 결과 타입.

2. 예시: MultiMap을 위한 Collector (Map<K, List<V>>)

예를 들어, Pair<K, V>의 스트림을 Map<K, List<V>>(멀티맵)으로 수집하여 각 키에 값 리스트를 매핑하고 싶다고 합시다.

구현 예:

public static <K, V> Collector<Pair<K, V>, ?, Map<K, List<V>>> toMultiMap() {
    return Collector.of(
        HashMap::new, // supplier
        (map, pair) -> map.computeIfAbsent(pair.key(), k -> new ArrayList<>()).add(pair.value()), // accumulator
        (map1, map2) -> { // combiner
            map2.forEach((k, vList) -> map1.merge(k, vList, (l1, l2) -> { l1.addAll(l2); return l1; }));
            return map1;
        },
        Function.identity(), // finisher
        Collector.Characteristics.UNORDERED
    );
}

사용 예:

List<Pair<String, Integer>> pairs = List.of(
    new Pair<>("a", 1), new Pair<>("b", 2), new Pair<>("a", 3)
);

Map<String, List<Integer>> multiMap = pairs.stream().collect(toMultiMap());
// multiMap: {a=[1, 3], b=[2]}

3. 예시: 상위 N개 요소를 위한 Collector

예를 들어, 스트림에서 가장 큰 요소 N개를 리스트로 수집하고 싶습니다(예: 내림차순 상위 5개).

구현:

public static <T> Collector<T, ?, List<T>> topN(int n, Comparator<? super T> comparator) {
    return Collector.of(
        () -> new PriorityQueue<>(n, comparator), // supplier
        (pq, t) -> {
            pq.offer(t);
            if (pq.size() > n) pq.poll(); // 가장 작은 값을 제거
        },
        (pq1, pq2) -> {
            pq2.forEach(t -> {
                pq1.offer(t);
                if (pq1.size() > n) pq1.poll();
            });
            return pq1;
        },
        pq -> {
            List<T> result = new ArrayList<>(pq);
            result.sort(comparator.reversed()); // 내림차순
            return result;
        },
        Collector.Characteristics.UNORDERED
    );
}

사용 예:

List<Integer> top3 = Stream.of(5, 1, 9, 3, 7, 2).collect(topN(3, Comparator.naturalOrder()));
// top3: [9, 7, 5]

4. 언제 직접 Collector를 작성하지 말아야 하는가

• 표준 컬렉터와 다운스트림 연산(groupingBy, mapping, flatMapping, collectingAndThen 등)의 조합으로 문제를 표현할 수 있다면, 그것들을 사용하는 것이 더 낫습니다.
• 자체 Collector는 특수한 데이터 구조, 복잡한 집계, 상위 N, 멀티맵 등과 같은 정말 비정형적인 시나리오에만 필요합니다.
• Collector를 위한 Collector를 만들지 마세요 — 유지보수와 테스트만 복잡해집니다.

예시:

// Map<K, Set<V>>를 위한 직접 Collector 대신:
.collect(Collectors.groupingBy(
    Pair::key,
    Collectors.mapping(Pair::value, Collectors.toSet())
))

5. 사용자 정의 Spliterator: 왜 그리고 어떻게

Spliterator는 컬렉션(또는 다른 데이터 소스)을 효율적으로 순회하고 분할하기 위한 특별한 인터페이스로, 특히 병렬 처리에 유용합니다. 일반적인 이터레이터와 달리, Spliterator는 병렬 처리를 위해 컬렉션을 서로 독립적인 조각으로 ‘분할’(split)할 수 있습니다.

핵심 메서드:

• tryAdvance(Consumer<? super T> action) — 다음 요소를 처리합니다.
• trySplit() — 컬렉션을 두 부분으로 나누려고 시도합니다(한 부분에 대한 새 Spliterator를 반환).
• estimateSize() — 남은 요소 수의 추정값.
• characteristics() — 특성 비트 마스크(ORDERED, SIZED, SUBSIZED 등).

trySplit: 분할 전략

균형 잡힌 분할 — 병렬 스트림에서 중요합니다: trySplit은 작업이 고르게 분배되도록 크기가 비슷한 부분을 반환해야 합니다.

분할할 수 없다면(예: 요소가 너무 적음) — null을 반환합니다.

예시: 파일을 청크 단위로 읽는 Spliterator
큰 파일을 메모리에 모두 올리지 않고 한 번에 1000줄씩(청크) 처리하고 싶다고 합시다.

public class ChunkedLineSpliterator implements Spliterator<List<String>> {
    private final BufferedReader reader;
    private final int chunkSize;

    public ChunkedLineSpliterator(BufferedReader reader, int chunkSize) {
        this.reader = reader;
        this.chunkSize = chunkSize;
    }

    @Override
    public boolean tryAdvance(Consumer<? super List<String>> action) {
        List<String> chunk = new ArrayList<>(chunkSize);
        try {
            String line;
            for (int i = 0; i < chunkSize && (line = reader.readLine()) != null; i++) {
                chunk.add(line);
            }
            if (chunk.isEmpty()) return false;
            action.accept(chunk);
            return true;
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    }

    @Override
    public Spliterator<List<String>> trySplit() {
        // 파일의 스트리밍 읽기에서는 분할이 의미 없으므로 null 반환
        return null;
    }

    @Override
    public long estimateSize() {
        return Long.MAX_VALUE; // 사전에 알 수 없음
    }

    @Override
    public int characteristics() {
        return ORDERED | NONNULL;
    }
}

사용 예:

try (BufferedReader reader = Files.newBufferedReader(Path.of("big.txt"))) {
    StreamSupport.stream(new ChunkedLineSpliterator(reader, 1000), false)
        .forEach(chunk -> processChunk(chunk));
}

Spliterator 특성

• ORDERED — 요소가 특정 순서를 가집니다(예: 리스트).
• SIZED — 요소의 정확한 개수를 알고 있습니다.
• SUBSIZED — trySplit으로 생성된 모든 Spliterator도 SIZED입니다.
• IMMUTABLE — 순회 중 소스가 변경되지 않습니다.
• CONCURRENT — 소스가 안전한 병렬 수정을 지원합니다.
• DISTINCT, SORTED, NONNULL — 추가 속성.

중요: 특성을 올바르게 지정하는 것은 스트림 최적화에 영향을 줍니다.

6. 예시

• 파일을 청크(덩어리)로 읽기 — 큰 파일을 전체 메모리에 올리지 않고 부분적으로 처리할 수 있습니다.
• 불필요한 할당 없이 파싱 — 바이트/문자 스트림을 파싱하면서 임시 객체 생성을 최소화하려면, 원본 배열의 “윈도” 또는 “슬라이스”를 반환하는 Spliterator를 구현할 수 있습니다.

예시: CSV를 행 단위로 파싱하는 Spliterator

public class CsvLineSpliterator implements Spliterator<String[]> {
    private final BufferedReader reader;

    public CsvLineSpliterator(BufferedReader reader) {
        this.reader = reader;
    }

    @Override
    public boolean tryAdvance(Consumer<? super String[]> action) {
        try {
            String line = reader.readLine();
            if (line == null) return false;
            action.accept(line.split(","));
            return true;
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    }

    @Override
    public Spliterator<String[]> trySplit() {
        return null; // 순차 파싱
    }

    @Override
    public long estimateSize() {
        return Long.MAX_VALUE;
    }

    @Override
    public int characteristics() {
        return ORDERED | NONNULL;
    }
}

7. parallel()과의 통합 — 안전하게 구현하는 법

• Spliterator가 병렬 분할을 지원하고(trySplit이 null이 아님), 특성에 SIZED/SUBSIZED가 포함되어 있으면 Stream API가 효율적으로 병렬화할 수 있습니다.
• 스트리밍 소스(파일, 소켓)에서는 보통 분할을 지원하지 않습니다 — 순차 스트림을 사용하세요.
• 컬렉션과 배열의 경우 — 균형 잡힌 분할을 구현하세요(예: 배열을 반으로 나누기).

예시: 배열용 Spliterator

public class ArraySpliterator<T> implements Spliterator<T> {
    private final T[] array;
    private int start, end;

    public ArraySpliterator(T[] array, int start, int end) {
        this.array = array;
        this.start = start;
        this.end = end;
    }

    @Override
    public boolean tryAdvance(Consumer<? super T> action) {
        if (start < end) {
            action.accept(array[start++]);
            return true;
        }
        return false;
    }

    @Override
    public Spliterator<T> trySplit() {
        int mid = (start + end) >>> 1;
        if (mid == start) return null;
        ArraySpliterator<T> split = new ArraySpliterator<>(array, start, mid);
        start = mid;
        return split;
    }

    @Override
    public long estimateSize() {
        return end - start;
    }

    @Override
    public int characteristics() {
        return ORDERED | SIZED | SUBSIZED | IMMUTABLE;
    }
}

사용 예:

String[] arr = {"a", "b", "c", "d"};
StreamSupport.stream(new ArraySpliterator<>(arr, 0, arr.length), true)
    .forEach(System.out::println);