beautifulsoups.dev
Published on

Java Stream 병렬 처리 느려짐? Spliterator·Collector 튜닝

Authors

서버 코드에서 parallelStream() 을 붙였는데 처리량이 줄고 지연이 늘어나는 경우가 흔합니다. 이유는 단순히 “병렬이니까 빠르겠지”라는 기대가 자바 스트림의 실행 모델과 충돌하기 때문입니다. 스트림 병렬화는 ForkJoinPool 위에서 동작하며, 분할(Spliterator)이 얼마나 잘 쪼개지는지, 중간/최종 연산이 병렬 친화적인지, Collector의 병합 비용이 얼마나 되는지에 따라 성능이 크게 갈립니다.

이 글에서는 병렬 스트림이 느려지는 대표 원인을 짚고, Spliterator/Collector 관점에서 병목을 제거하는 튜닝 방법을 코드로 설명합니다.

병렬 스트림이 느려지는 7가지 전형적 이유

1) 분할이 잘 안 되는 소스: Spliterator 특성이 빈약함

병렬 스트림은 입력을 여러 조각으로 쪼개 각 스레드가 처리합니다. 이때 Spliterator.trySplit() 이 잘게 분할하지 못하면 병렬화 이점이 거의 없습니다.

  • ArrayList/배열: 대체로 분할이 매우 잘 됨
  • LinkedList, Iterator 기반, I/O 스트림: 분할이 나쁨
  • 커스텀 컬렉션: SIZED, SUBSIZED, ORDERED 같은 특성을 제대로 제공하지 않으면 분할/스케줄링이 비효율적

2) 작업이 너무 작음: 오버헤드가 계산을 압도

병렬 스트림은 태스크 분할, 큐잉, 워크 스틸링, 병합 비용이 있습니다. 요소당 연산이 가벼우면 이 오버헤드가 더 커져 느려집니다.

3) 박싱/언박싱과 객체 할당 폭증

Stream<Integer> 같은 박싱 스트림은 요소 처리마다 오브젝트를 만지고 GC 압력을 높입니다. 가능하면 IntStream, LongStream 등 프리미티브 스트림을 사용해야 합니다.

4) 공유 상태/락/동기화로 인한 경합

병렬 스트림에서 forEach 로 공유 Map 에 put 하거나, synchronized 블록을 잡으면 스레드가 동시에 달려들며 락 경합이 생깁니다. 결과적으로 직렬보다 느려집니다.

5) Collector 병합 비용이 큼

병렬 수집은 기본적으로

  • 각 스레드가 로컬 컨테이너에 누적
  • 마지막에 컨테이너들을 병합

이라는 구조입니다. 컨테이너가 크거나 병합이 O(n) 으로 비싸면 병렬화가 손해가 됩니다.

6) 순서 보장 연산이 병렬성을 제한

sorted, distinct, limit, findFirst 같은 연산은 전역 정렬/전역 상태가 필요해 병렬 효율이 크게 떨어질 수 있습니다.

7) 포크조인 풀 공유로 인한 간섭

기본 parallelStream() 은 공용 ForkJoinPool.commonPool() 을 사용합니다. 같은 JVM에서 다른 병렬 작업(서블릿, 비동기 작업, 다른 병렬 스트림)이 common pool을 함께 쓰면 서로 간섭합니다.

먼저 확인할 것: 병렬화가 이득인 작업인가

병렬 스트림이 유효한 경우는 대체로 다음 조건을 만족합니다.

  • CPU 바운드(계산량 큼)
  • 요소 수가 충분히 많음
  • 요소당 연산이 꽤 무거움(예: 암호화, 복잡한 파싱, 큰 JSON 처리)
  • 공유 상태가 없음

반대로 DB 호출, HTTP 호출 같은 I/O 바운드에는 병렬 스트림이 “스레드만 늘리는” 결과가 되기 쉽습니다. I/O 병렬화는 보통 비동기 또는 별도 스레드풀/리액티브 모델이 더 적합합니다.

Spliterator 튜닝: 잘 쪼개지게 만들기

Spliterator 의 핵심은 trySplit()

병렬 스트림은 내부적으로 trySplit() 을 반복 호출해 일을 나눕니다. 커스텀 데이터 소스를 스트림으로 만들 때 Spliterator 를 직접 구현하면 분할 전략을 최적화할 수 있습니다.

아래 예시는 큰 int[] 를 일정 청크 단위로 분할하는 Spliterator.OfInt 입니다. 포인트는

  • estimateSize() 를 정확히 제공
  • SIZED | SUBSIZED | ORDERED | IMMUTABLE 같은 특성을 가능한 한 제공
  • trySplit() 에서 반씩 쪼개되, 너무 작은 단위로 쪼개지지 않게 임계값을 둠

입니다.

import java.util.Spliterator;
import java.util.function.IntConsumer;

public final class ChunkedIntArraySpliterator implements Spliterator.OfInt {
    private final int[] data;
    private int index;
    private final int fence;
    private final int minChunk;

    public ChunkedIntArraySpliterator(int[] data, int origin, int fence, int minChunk) {
        this.data = data;
        this.index = origin;
        this.fence = fence;
        this.minChunk = minChunk;
    }

    @Override
    public OfInt trySplit() {
        int lo = index;
        int hi = fence;
        int remaining = hi - lo;
        if (remaining <= minChunk) return null;

        int mid = lo + remaining / 2;
        index = mid;
        return new ChunkedIntArraySpliterator(data, lo, mid, minChunk);
    }

    @Override
    public boolean tryAdvance(IntConsumer action) {
        if (index < fence) {
            action.accept(data[index++]);
            return true;
        }
        return false;
    }

    @Override
    public void forEachRemaining(IntConsumer action) {
        for (; index < fence; index++) {
            action.accept(data[index]);
        }
    }

    @Override
    public long estimateSize() {
        return fence - index;
    }

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

이 Spliterator 로 스트림을 만들면 분할이 예측 가능해지고, 너무 잘게 쪼개져 오버헤드가 늘어나는 상황도 완화할 수 있습니다.

import java.util.stream.IntStream;
import java.util.stream.StreamSupport;

int[] data = new int[10_000_000];
// ... fill

var sp = new ChunkedIntArraySpliterator(data, 0, data.length, 50_000);
IntStream s = StreamSupport.intStream(sp, true);

long sum = s.map(x -> x * 31L).sum();

LinkedList 를 병렬로 돌리고 있다면 구조를 바꿔라

LinkedList 는 분할이 비효율적입니다. 가능하면 ArrayList 로 변환하거나, 애초에 배열/프리미티브 배열로 들고 가는 편이 낫습니다.

// 안 좋은 예: LinkedList 병렬 처리
long c1 = linkedList.parallelStream().filter(this::heavy).count();

// 개선: ArrayList로 복사 후 병렬 처리
var arrayList = new java.util.ArrayList<>(linkedList);
long c2 = arrayList.parallelStream().filter(this::heavy).count();

복사 비용이 들지만, 요소 수가 크고 연산이 무거우면 전체적으로 이득이 되는 경우가 많습니다.

Collector 튜닝: 병합 비용과 경합 줄이기

Collectors.toList() 는 보통 안전하지만 “최적”은 아니다

병렬에서 toList() 는 스레드별 리스트를 만들고 마지막에 합치는 방식이라 공유 락 경합은 적습니다. 하지만 최종 병합 때 리스트 크기가 크면 병합 비용이 눈에 띌 수 있습니다.

특히 groupingBy 류는 병합이 비쌀 수 있습니다.

groupingBy 대신 groupingByConcurrent 가 항상 빠르진 않다

groupingByConcurrent 는 ConcurrentHashMap 을 사용해 동시에 업데이트합니다. 키 충돌이 많거나 값 컨테이너 업데이트가 잦으면 CAS/락 경합으로 느려질 수 있습니다.

  • 키 분포가 넓고 충돌이 적다: 유리할 수 있음
  • 핫키가 존재한다: 경합으로 불리

예시:

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

// 충돌이 적을 때 유리할 수 있음
var m1 = items.parallelStream()
    .collect(groupingByConcurrent(Item::category, counting()));

// 핫키가 많으면 오히려 groupingBy가 나을 때도 있음
var m2 = items.parallelStream()
    .collect(groupingBy(Item::category, counting()));

결론은 “측정”입니다. 데이터 분포가 성능을 결정합니다.

커스텀 Collector로 병합을 더 싸게 만들기

문자열을 합치는 작업을 예로 들면 Collectors.joining() 은 내부적으로 StringBuilder 를 사용하지만, 병렬에서 병합이 잦으면 비용이 커질 수 있습니다. 커스텀 Collector 로

  • 누적 컨테이너를 가볍게
  • 병합을 최소화

하는 전략을 취할 수 있습니다.

아래는 StringBuilder 를 누적 컨테이너로 쓰는 커스텀 Collector 예시입니다.

import java.util.Set;
import java.util.stream.Collector;

public final class StringBuilderCollector {
    public static Collector<CharSequence, StringBuilder, String> joining(CharSequence delimiter) {
        return Collector.of(
            StringBuilder::new,
            (sb, s) -> {
                if (sb.length() > 0) sb.append(delimiter);
                sb.append(s);
            },
            (left, right) -> {
                if (left.length() == 0) return right;
                if (right.length() == 0) return left;
                left.append(delimiter).append(right);
                return left;
            },
            StringBuilder::toString,
            Set.of(Collector.Characteristics.UNORDERED)
        );
    }
}

사용:

String out = items.parallelStream()
    .map(Item::name)
    .collect(StringBuilderCollector.joining(","));

주의할 점은 UNORDERED 특성을 주면 순서가 바뀔 수 있다는 것입니다. 순서가 중요하면 UNORDERED 를 주지 말고, 대신 병렬화 이득이 줄 수 있음을 받아들여야 합니다.

reduce 는 병렬 친화적이지만 조건이 있다

병렬 reduce 는 결합법칙을 만족해야 합니다. 예를 들어 덧셈은 안전하지만, 문자열 덧붙이기를 reduce 로 하면 병합 비용이 커지고 느려집니다.

// 좋은 예: 결합법칙 성립, 비용 작음
long sum = nums.parallelStream().reduce(0L, Long::sum);

// 나쁜 예: 문자열 + 는 병합 비용이 큼
String s = words.parallelStream().reduce("", (a, b) -> a + b);

문자열은 위에서 본 것처럼 StringBuilder 기반 Collector로 처리하는 편이 낫습니다.

스트림 파이프라인 자체 튜닝 포인트

프리미티브 스트림으로 바꾸기

// 박싱: Integer 객체가 왔다갔다
long s1 = list.parallelStream().mapToLong(Integer::longValue).sum();

// 처음부터 IntStream이면 더 낫다
long s2 = java.util.stream.IntStream.of(array).parallel().asLongStream().sum();

forEachOrdered 를 피하기

병렬에서 순서를 강제하면 병렬성이 크게 제한됩니다.

// 병렬인데 사실상 발목 잡을 수 있음
items.parallelStream().forEachOrdered(this::process);

// 순서가 필요 없다면
items.parallelStream().forEach(this::process);

sorted() 를 병렬로 붙이면 이득이 적을 수 있음

정렬은 전역 연산이라 비용이 큽니다. 가능하면 정렬을 줄이거나, 정렬 범위를 축소하거나, 병렬화 대상에서 제외하세요.

commonPool 간섭 피하기: 전용 ForkJoinPool 사용

서비스에서 parallelStream() 을 무분별하게 쓰면 common pool이 포화되어 전체 애플리케이션 지연이 증가할 수 있습니다. 특정 작업만 병렬화하고 싶다면 전용 ForkJoinPool 에서 실행하세요.

import java.util.concurrent.ForkJoinPool;

ForkJoinPool pool = new ForkJoinPool(Math.min(8, Runtime.getRuntime().availableProcessors()));

long result = pool.submit(() ->
    items.parallelStream()
         .mapToLong(this::heavyCpuWork)
         .sum()
).join();

pool.shutdown();

이 방식은 common pool 간섭을 줄이지만, 풀을 너무 많이 만들면 스레드가 늘어 역효과가 날 수 있습니다. “작업 단위별로 1개” 정도로 제한하는 것이 보통 안전합니다.

측정이 먼저: JMH로 병렬화 손익분기점 찾기

병렬 스트림 튜닝은 추측으로 하면 실패합니다. 최소한 JMH로

  • 입력 크기
  • 스레드 수
  • 분할 임계값
  • Collector 선택

을 바꿔가며 측정해야 합니다.

import org.openjdk.jmh.annotations.*;

import java.util.*;
import java.util.concurrent.TimeUnit;

@BenchmarkMode(Mode.Throughput)
@OutputTimeUnit(TimeUnit.SECONDS)
@State(Scope.Thread)
public class ParallelStreamBench {

    @Param({"100000", "1000000"})
    int n;

    List<Integer> data;

    @Setup
    public void setup() {
        data = new ArrayList<>(n);
        for (int i = 0; i < n; i++) data.add(i);
    }

    private long heavy(int x) {
        long v = x;
        for (int i = 0; i < 200; i++) v = v * 1664525L + 1013904223L;
        return v;
    }

    @Benchmark
    public long sequential() {
        return data.stream().mapToLong(this::heavy).sum();
    }

    @Benchmark
    public long parallel() {
        return data.parallelStream().mapToLong(this::heavy).sum();
    }
}

여기서 중요한 건 “내 서버/내 데이터/내 파이프라인”으로 측정하는 것입니다. 병렬 스트림 성능은 CPU 코어 수, GC, 데이터 분포, 다른 워크로드에 따라 크게 달라집니다.

실전 체크리스트

  • 입력 소스가 잘 분할되는가: 배열/ArrayList 선호, LinkedList 지양
  • 요소당 연산이 충분히 무거운가: 가벼우면 직렬이 더 빠를 가능성 큼
  • 프리미티브 스트림으로 박싱 제거했는가
  • 공유 상태를 없앴는가: Atomic/락/공유 Map 업데이트는 경합 유발
  • Collector 병합 비용을 점검했는가: groupingByConcurrent 는 만능이 아님
  • sorted, distinct, findFirst, forEachOrdered 같은 전역/순서 연산을 재검토했는가
  • common pool 간섭이 있는가: 전용 ForkJoinPool 고려
  • JMH로 손익분기점을 확인했는가

마무리: 병렬 스트림은 “자동 튜닝”이 아니라 “분할·병합 설계”다

parallelStream() 은 버튼 하나로 병렬화되는 것처럼 보이지만, 실제 성능은 Spliterator 의 분할 품질과 Collector 의 병합 비용이 좌우합니다. 특히 커스텀 데이터 소스나 복잡한 집계 로직에서는 기본 구현이 최선이 아닐 수 있습니다.

병렬 스트림이 느려졌다면 “스레드를 늘려서”가 아니라,

  • 얼마나 잘 쪼개지는지(Spliterator)
  • 얼마나 싸게 모으는지(Collector)

부터 의심하고, JMH로 데이터 기반 결정을 내리는 것이 가장 빠른 해결책입니다.

추가로, 성능 문제를 원인별로 분해해 진단하는 접근은 다른 영역에서도 동일하게 통합니다. 예를 들어 데이터 파이프라인에서 병목을 찾는 방식은 MongoDB $lookup 느림? 인덱스·pipeline 튜닝 같은 글에서도 유사한 관점으로 확장해볼 수 있고, CI 빌드 캐시가 느릴 때도 GitHub Actions 캐시 미스 - 키·경로 디버깅 실전 처럼 “원인 분해 후 측정”이 핵심입니다.

Discuss on Twitter • View on GitHub