beautifulsoups.dev
Published on

Java Stream 병렬처리 성능폭망? Spliterator·Collector로 해결

Authors

자바에서 parallelStream()은 “버튼 한 번으로 병렬화”처럼 보이지만, 실무에서는 오히려 성능이 폭망하는 경우가 흔합니다. 이유는 단순히 “스레드가 많아서”가 아니라, 분할(splitting) 품질, 작업 단위의 크기, collector의 병렬 친화성, 공유 상태/동기화 비용, 데이터 소스의 특성이 맞물려서 발생합니다.

이 글에서는 병렬 스트림이 느려지는 메커니즘을 Spliterator와 Collector 관점에서 해부하고, “진짜 병렬화”가 되도록 고치는 방법을 코드 중심으로 정리합니다.

운영 장애에서 원인 추적이 중요한 것처럼, 성능 문제도 “어디서 병목이 생기는지”를 먼저 분해해야 합니다. 비슷한 관점의 글로 리눅스 OOMKilled 원인 추적 - cgroup·dmesg·ulimit도 참고할 만합니다.

왜 parallelStream()이 느려지는가

병렬 스트림은 내부적으로 ForkJoinPool.commonPool()에서 작업을 쪼개 실행합니다. 이때 성능은 크게 아래에 의해 좌우됩니다.

1) Spliterator가 “잘” 쪼개지지 않으면 병렬화가 실패한다

스트림의 데이터 소스는 Spliterator로 분할됩니다. 분할이 잘 되려면:

  • trySplit()이 빠르고 균등하게 쪼개야 함
  • SIZED, SUBSIZED 같은 특성이 있으면 분할/스케줄링이 유리함
  • 링크드 구조(LinkedList)나 I/O 기반 반복자처럼 분할이 어려우면 한 스레드가 대부분 일을 떠안음

즉, parallelStream()은 “데이터를 잘게 나눌 수 있을 때”만 이득이 납니다.

2) 작업 단위가 너무 작으면 오버헤드가 이긴다

병렬화는 공짜가 아닙니다.

  • 태스크 분할/큐잉
  • 워커 간 훔치기(work-stealing)
  • 결과 병합

이 비용이 실제 연산보다 크면, 병렬이 더 느립니다. 예를 들어 단순한 map(x -> x + 1) 정도는 대부분 병렬화 이득이 없습니다.

3) 공유 상태 + 동기화(락) + 경쟁이 병렬 이득을 갉아먹는다

병렬 스트림에서 아래 패턴은 특히 위험합니다.

  • forEach에서 synchronizedList.add() 같은 공유 컬렉션 갱신
  • AtomicLong.incrementAndGet() 같은 핫스팟 원자 연산
  • Collectors.toMap()에서 키 충돌이 잦아 병합 비용이 커짐

병렬은 “각 스레드가 독립적으로 계산하고 마지막에 합치는” 구조에 최적화되어 있습니다.

4) 블로킹 I/O나 외부 호출은 commonPool을 망가뜨린다

병렬 스트림은 기본적으로 ForkJoinPool.commonPool을 씁니다. 여기서 DB/HTTP 같은 블로킹 작업을 하면 워커가 묶여 전체 처리량이 떨어집니다.

이건 gRPC에서 데드라인/리트라이가 폭주하면 전체 시스템이 무너지는 것과 유사한 구조적 문제입니다. 필요하면 gRPC MSA에서 데드라인·리트라이 폭주 막는 법처럼 “리소스 격리” 관점으로 접근해야 합니다.

먼저 확인할 체크리스트

병렬 스트림을 쓰기 전에 아래를 점검하세요.

  1. 데이터 소스가 ArrayList, 배열, 범위(IntStream.range)처럼 잘 쪼개지는가
  2. 연산량이 충분히 큰가(보통 원소당 수십~수백 ns 수준이면 이득 없음)
  3. 공유 상태를 건드리지 않는가
  4. 결과 수집(collector)이 병렬 친화적인가
  5. 블로킹 작업을 parallelStream()으로 돌리고 있지 않은가

이제부터는 “고치는 방법”을 Spliterator와 Collector로 나눠 설명합니다.

Spliterator로 분할 품질을 개선하기

문제 예시: 커스텀 데이터가 한 덩어리로 처리됨

예를 들어 로그 파일을 줄 단위로 읽어 처리한다고 합시다. 단순히 BufferedReader.lines().parallel()을 하면 분할이 잘 안 되거나(소스 특성), I/O 자체가 병렬화에 부적합해서 성능이 나빠질 수 있습니다.

핵심은 병렬화 가능한 단위로 미리 쪼개서 스트림에 공급하는 것입니다.

해결 전략 1: “미리 청크로 분할된” Spliterator 만들기

아래는 List를 일정 크기 청크로 나눠 병렬 처리 효율을 높이는 예시입니다. (원소 단위가 너무 작을 때 특히 유용)

import java.util.*;
import java.util.function.Consumer;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

public final class ChunkedSpliterator<T> implements Spliterator<List<T>> {
    private final List<T> source;
    private final int chunkSize;
    private int index;

    public ChunkedSpliterator(List<T> source, int chunkSize) {
        this.source = Objects.requireNonNull(source);
        this.chunkSize = Math.max(1, chunkSize);
        this.index = 0;
    }

    @Override
    public boolean tryAdvance(Consumer<? super List<T>> action) {
        if (index >= source.size()) return false;
        int end = Math.min(source.size(), index + chunkSize);
        action.accept(source.subList(index, end));
        index = end;
        return true;
    }

    @Override
    public Spliterator<List<T>> trySplit() {
        int remaining = source.size() - index;
        if (remaining <= chunkSize * 2) return null; // 너무 작으면 분할하지 않음

        int mid = index + remaining / 2;
        // mid를 chunkSize 경계로 맞추면 병합/캐시 효율이 좋아지는 경우가 많음
        mid = mid - (mid % chunkSize);
        if (mid <= index) return null;

        List<T> left = source.subList(index, mid);
        index = mid;
        return new ChunkedSpliterator<>(left, chunkSize);
    }

    @Override
    public long estimateSize() {
        int remaining = source.size() - index;
        return (remaining + chunkSize - 1L) / chunkSize;
    }

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

    public static <T> Stream<List<T>> chunkedStream(List<T> list, int chunkSize, boolean parallel) {
        Spliterator<List<T>> sp = new ChunkedSpliterator<>(list, chunkSize);
        return StreamSupport.stream(sp, parallel);
    }
}

이제 원소 하나씩 처리하는 대신 “청크 단위”로 병렬 처리합니다.

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

class Demo {
    public static void main(String[] args) {
        List<Integer> data = IntStream.range(0, 10_000_000).boxed().toList();

        long sum = ChunkedSpliterator
                .chunkedStream(data, 10_000, true)
                .mapToLong(chunk -> {
                    long s = 0;
                    for (int x : chunk) s += x;
                    return s;
                })
                .sum();

        System.out.println(sum);
    }
}

포인트는 다음입니다.

  • 태스크 개수가 “원소 수”가 아니라 “청크 수”로 줄어 오버헤드 감소
  • 각 태스크가 충분히 무거워져 병렬화 이득 증가
  • trySplit()이 균등 분할을 제공

해결 전략 2: 데이터 구조 자체를 바꿔라

LinkedList의 spliterator()는 분할 효율이 좋지 않습니다. 가능하면:

  • ArrayList로 변환 후 병렬 처리
  • primitive stream(IntStream, LongStream) 사용
List<MyObj> linked = new LinkedList<>();
// ... fill

List<MyObj> array = new ArrayList<>(linked);
array.parallelStream()
     .map(MyObj::heavy)
     .toList();

변환 비용이 있더라도 전체 파이프라인 비용 대비 이득이면 충분히 가치가 있습니다.

Collector로 병렬 수집 병목 제거하기

병렬 스트림의 성능 폭망은 종종 “연산”이 아니라 “수집”에서 터집니다.

흔한 안티패턴 1: 공유 컬렉션에 forEach로 넣기

List<Result> out = Collections.synchronizedList(new ArrayList<>());
input.parallelStream().forEach(x -> out.add(work(x))); // 락 경합

병렬화했는데 마지막에 락 하나로 줄 세우는 꼴입니다.

해결: collect를 사용하고, 병렬 친화적인 Collector를 선택

List<Result> out = input.parallelStream()
        .map(this::work)
        .collect(Collectors.toList());

toList()/Collectors.toList()는 내부적으로 스레드별 컨테이너에 누적 후 병합하는 방식이라 공유 락보다 훨씬 낫습니다.

흔한 안티패턴 2: Collectors.groupingBy로 병렬 그룹핑

Collectors.groupingBy()는 기본적으로 HashMap 기반이며 병합 비용이 커질 수 있습니다. 병렬 환경에서는 groupingByConcurrent()를 고려할 수 있습니다.

Map<String, List<Event>> byType = events.parallelStream()
        .collect(Collectors.groupingByConcurrent(Event::type));

다만 groupingByConcurrent는 동시성 맵을 쓰기 때문에:

  • 키가 매우 핫하면 내부 경쟁이 생김
  • 값 리스트에 대한 동시 append 비용이 생길 수 있음

이럴 때는 “값을 리스트로 모으는 것” 자체가 병목이므로, 아예 다른 수집 전략이 필요합니다(예: 카운팅, 요약 통계).

커스텀 Collector로 “락 없는 병렬 수집” 만들기

예를 들어 문자열을 정규화한 뒤 빈도를 세는 작업을 해봅시다. ConcurrentHashMap.compute를 병렬로 돌리면 키 경쟁이 심할 수 있습니다. 대신 스레드 로컬 맵에 누적하고 마지막에 병합하는 Collector를 만들 수 있습니다.

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

public final class FrequencyCollector {
    public static Collector<String, Map<String, Long>, Map<String, Long>> toFrequencyMap() {
        Supplier<Map<String, Long>> supplier = HashMap::new;

        BiConsumer<Map<String, Long>, String> accumulator = (m, s) ->
                m.merge(s, 1L, Long::sum);

        BinaryOperator<Map<String, Long>> combiner = (a, b) -> {
            // b를 a에 병합
            for (var e : b.entrySet()) {
                a.merge(e.getKey(), e.getValue(), Long::sum);
            }
            return a;
        };

        // IDENTITY_FINISH: 누적 컨테이너가 최종 결과와 동일
        return Collector.of(supplier, accumulator, combiner,
                Collector.Characteristics.IDENTITY_FINISH);
    }
}

사용 예:

Map<String, Long> freq = lines.parallelStream()
        .map(String::trim)
        .filter(s -> !s.isEmpty())
        .map(String::toLowerCase)
        .collect(FrequencyCollector.toFrequencyMap());

이 패턴의 장점:

  • 누적(accumulate)은 스레드 로컬 HashMap에서 진행되어 락이 없음
  • 병합(combine)은 태스크 병합 시점에만 발생
  • 키 충돌이 많아도 ConcurrentHashMap의 핫스팟 경쟁보다 유리한 경우가 많음

주의할 점은 메모리 사용량입니다. 스레드 수만큼 맵이 생기므로, 키 종류가 매우 많으면 메모리가 증가합니다. (이런 경우는 병렬화 자체가 메모리 압박을 만들 수 있으니, 필요하면 배치 크기/청크 크기 조절이 필요합니다.)

Spliterator와 Collector를 함께 써서 “진짜 병렬 파이프라인” 만들기

아래는 “작업 단위가 작은 원소”를 청크로 묶고, 각 청크에서 로컬 집계를 한 뒤, 마지막에 병합하는 예시입니다.

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

public class ParallelPipeline {

    static String normalize(String s) {
        return s.trim().toLowerCase();
    }

    public static Map<String, Long> parallelFrequency(List<String> input) {
        return ChunkedSpliterator.chunkedStream(input, 20_000, true)
                .map(chunk -> {
                    Map<String, Long> local = new HashMap<>();
                    for (String s : chunk) {
                        String n = normalize(s);
                        if (!n.isEmpty()) local.merge(n, 1L, Long::sum);
                    }
                    return local;
                })
                .reduce(new HashMap<>(), (a, b) -> {
                    for (var e : b.entrySet()) a.merge(e.getKey(), e.getValue(), Long::sum);
                    return a;
                });
    }

    public static void main(String[] args) {
        List<String> data = IntStream.range(0, 5_000_000)
                .mapToObj(i -> "Key" + (i % 10_000))
                .toList();

        Map<String, Long> freq = parallelFrequency(data);
        System.out.println(freq.get("key1"));
    }
}

이 접근은 스트림 API를 유지하면서도:

  • 분할 단위를 제어하고
  • 수집을 병렬 친화적으로 바꾸며
  • 공유 상태를 제거

하는 효과가 있습니다.

병렬 스트림을 쓸 때의 실무 팁

1) 블로킹 작업이면 parallelStream() 대신 격리된 실행기를 고려

병렬 스트림은 기본적으로 공용 풀을 쓰므로, 블로킹 작업을 섞으면 애플리케이션 전체에 악영향을 줄 수 있습니다. 이런 경우는:

  • CompletableFuture + 전용 ExecutorService
  • 또는 가상 스레드(자바 21+) 기반 구조

처럼 “풀을 격리”하는 설계가 안전합니다.

2) unordered()로 병합 비용을 줄일 수 있는지 확인

순서가 필요 없다면:

List<Result> out = input.parallelStream()
        .unordered()
        .map(this::work)
        .toList();

ORDERED 제약이 풀리면 병합/스케줄링이 더 유리해질 수 있습니다.

3) 측정은 반드시 JMH로

System.nanoTime()으로 재면 워밍업, 인라이닝, GC, OS 스케줄링에 흔들립니다. JMH로:

  • 워밍업/측정 분리
  • 스레드 수/데이터 크기별 비교
  • GC 프로파일링

을 해야 결론이 납니다.

결론: 병렬 스트림 성능은 “분할”과 “수집”이 결정한다

parallelStream()의 성능이 폭망하는 핵심 원인은 대개 다음 둘 중 하나(혹은 둘 다)입니다.

  • Spliterator 분할이 비효율적이라 워커가 놀거나 한쪽으로 쏠림
  • Collector/수집 단계가 공유 상태나 병합 비용으로 병목

해결책은 의외로 명확합니다.

  1. 분할 단위를 제어하라: 청크 기반 Spliterator 또는 데이터 구조 변경
  2. 락 없는 수집으로 바꿔라: 스레드 로컬 누적 후 병합하는 Collector/reduce 패턴
  3. 블로킹 작업은 병렬 스트림에 넣지 말고 격리하라

이 3가지만 지켜도 “병렬화했더니 더 느려짐”에서 벗어나, 병렬 스트림을 안정적으로 성능 도구로 쓸 수 있습니다.

Discuss on Twitter • View on GitHub