beautifulsoups.dev
Published on

Java Stream 병렬처리 함정 - shared state 버그

Authors

서버 코드에서 parallelStream()을 켜고 벤치마크가 빨라졌다고 안심했다가, 운영에서만 간헐적으로 합계가 틀리거나 누락되는 이슈를 만나는 경우가 있습니다. 원인은 대부분 간단합니다. 스트림 파이프라인 안에서 외부의 공유 상태(shared state) 를 읽거나 수정했기 때문입니다.

이 글에서는 Java Stream 병렬처리에서 자주 터지는 shared state 버그 패턴을 짚고, 올바른 수집(collect) 방식과 설계 원칙을 코드로 정리합니다.

parallelStream이 위험해지는 순간

parallelStream()은 내부적으로 ForkJoinPool.commonPool()을 사용해 작업을 분할하고, 여러 스레드에서 동시에 연산합니다. 따라서 아래 조건 중 하나라도 만족하면 결과가 비결정적이거나(실행할 때마다 달라짐), 성능이 오히려 나빠질 수 있습니다.

  • 람다 내부에서 외부 변수를 변경(가변 객체, 컬렉션, 카운터 등)
  • forEach로 외부 리스트에 add 같은 사이드 이펙트 수행
  • 스레드 안전하지 않은 객체(SimpleDateFormat, StringBuilder, HashMap 등)를 공유
  • 락/동기화로 안전하게 만들었지만 락 경합으로 병렬화 이점이 사라짐

병렬 스트림의 핵심 규칙은 간단합니다.

  • map, filter, reduce는 순수 함수처럼 작성한다
  • 결과는 collect 같은 리덕션 연산 으로 모은다
  • 외부 상태 변경은 금지(또는 정말 필요하면 다른 도구를 쓴다)

대표 안티패턴 1: 공유 카운터 증가

아래 코드는 단골로 등장하는 버그입니다.

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

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

        int[] counter = new int[1]; // 가변 공유 상태

        nums.parallelStream().forEach(n -> {
            if ((n & 1) == 0) {
                counter[0]++; // 경쟁 상태(race condition)
            }
        });

        System.out.println("even count=" + counter[0]);
    }
}

counter[0]++는 원자적 연산이 아닙니다. 읽기-증가-쓰기 단계가 분리되므로 여러 스레드가 동시에 실행하면 업데이트가 유실됩니다.

올바른 대안

  1. 스트림 자체로 카운트하기
long evenCount = nums.parallelStream()
        .filter(n -> (n & 1) == 0)
        .count();
  1. 정말 카운터가 필요하면 LongAdder
import java.util.concurrent.atomic.*;

LongAdder adder = new LongAdder();
nums.parallelStream().forEach(n -> {
    if ((n & 1) == 0) adder.increment();
});
long evenCount = adder.sum();

다만 2번은 “스트림 철학”과는 거리가 있고, 카운트 목적이면 1번이 더 명확합니다.

대표 안티패턴 2: parallel forEach로 외부 리스트에 add

List<String> out = new ArrayList<>();

input.parallelStream()
     .map(String::trim)
     .forEach(out::add); // ArrayList는 스레드 안전하지 않음

이 코드는 다음 중 하나로 망가집니다.

  • ArrayIndexOutOfBoundsException 같은 런타임 예외
  • 예외는 없는데 결과 누락/중복
  • 운 좋게 통과하지만, JVM/부하/데이터 크기에 따라 간헐 재현

올바른 대안: collect로 모으기

List<String> out = input.parallelStream()
        .map(String::trim)
        .toList(); // Java 16+

Java 8 호환이 필요하면:

List<String> out = input.parallelStream()
        .map(String::trim)
        .collect(Collectors.toList());

collect는 병렬 환경에서 안전하게 부분 결과를 만들고 마지막에 합치는 방식으로 동작합니다(Collector의 특성에 따라). 핵심은 외부 리스트를 직접 건드리지 않는 것 입니다.

대표 안티패턴 3: shared mutable Map에 put

예를 들어 사용자별 합계를 만들려고 아래처럼 작성하면 쉽게 깨집니다.

Map<String, Integer> sumByUser = new HashMap<>();

orders.parallelStream().forEach(o -> {
    sumByUser.merge(o.userId(), o.amount(), Integer::sum); // HashMap은 안전하지 않음
});

merge 자체는 편하지만, HashMap이 병렬 업데이트를 견디지 못합니다.

올바른 대안 1: groupingBy + summingInt

Map<String, Integer> sumByUser = orders.parallelStream()
        .collect(Collectors.groupingBy(
                Order::userId,
                Collectors.summingInt(Order::amount)
        ));

올바른 대안 2: groupingByConcurrent

키가 많고 병렬 수집을 적극 활용하려면:

import java.util.concurrent.*;

ConcurrentMap<String, Integer> sumByUser = orders.parallelStream()
        .collect(Collectors.groupingByConcurrent(
                Order::userId,
                Collectors.summingInt(Order::amount)
        ));

단, groupingByConcurrent는 결과 타입이 ConcurrentMap이고, downstream collector가 동시 업데이트에 적합한지(또는 프레임워크가 적절히 분리/병합하는지) 이해하고 써야 합니다.

peek로 디버깅하다가 사이드 이펙트 심기

peek는 “중간 연산 디버깅” 용도로 소개되지만, 병렬에서는 출력 순서가 뒤섞이고, 더 나쁘게는 peek 안에서 상태 변경을 하며 버그를 심는 경우가 많습니다.

List<String> seen = new ArrayList<>();

long cnt = input.parallelStream()
        .peek(seen::add) // 위험: shared state
        .filter(s -> s.length() > 3)
        .count();

디버깅이 목적이라면, 병렬을 잠시 끄고 stream()으로 확인하거나, 로깅을 하더라도 외부 상태 변경은 피하세요.

synchronized로 감싸면 해결일까?

공유 리스트를 Collections.synchronizedList로 감싸면 예외는 줄어듭니다.

List<String> out = Collections.synchronizedList(new ArrayList<>());
input.parallelStream().forEach(out::add);

하지만 이 방식은 다음 문제가 남습니다.

  • 락 경합으로 병렬 이점이 사라짐(사실상 직렬화)
  • 호출자가 락 범위를 잘못 쓰면 여전히 위험
  • 설계 의도가 불명확(스트림을 왜 쓰는지 애매)

대부분은 collect(toList())로 끝납니다. 락으로 “고쳐 쓰기”는 최후의 수단으로 두는 편이 좋습니다.

병렬 스트림이 특히 취약한 객체들

병렬 스트림에서 공유하면 위험한 대표 예시입니다.

  • SimpleDateFormat: 스레드 안전하지 않음
  • StringBuilder: 스레드 안전하지 않음(대신 StringBuffer는 동기화지만 성능 비용)
  • HashMap, ArrayList: 스레드 안전하지 않음
  • JPA EntityManager 같은 요청/트랜잭션 스코프 객체: 스레드 안전하지 않음

예를 들어 날짜 파싱을 병렬화하고 싶다면 DateTimeFormatter를 사용하세요.

import java.time.*;
import java.time.format.*;

DateTimeFormatter fmt = DateTimeFormatter.ISO_LOCAL_DATE;

List<LocalDate> dates = strings.parallelStream()
        .map(s -> LocalDate.parse(s, fmt))
        .toList();

reduce를 잘못 쓰면 결과가 틀어진다

병렬에서 reduce는 결합 법칙(associative)과 항등원(identity)이 중요합니다. 예를 들어 문자열을 이어붙이는 것을 reduce로 하면 성능도 나쁘고(중간 문자열 폭증), 결합 방식에 따라 결과가 예상과 달라질 수 있습니다.

나쁜 예:

String joined = input.parallelStream()
        .reduce("", (a, b) -> a + b); // 비효율 + 병렬에 부적합

대안:

String joined = input.parallelStream()
        .collect(Collectors.joining());

언제 parallelStream을 쓰면 좋은가

병렬 스트림은 만능이 아닙니다. 아래 조건에서만 “검토할 가치”가 큽니다.

  • 요소 수가 충분히 많고(수만~수백만)
  • 각 요소 처리 비용이 크고 CPU 바운드이며
  • 처리 함수가 순수 함수에 가깝고(공유 상태 없음)
  • 스레드 풀 공용 사용(commonPool)이 다른 작업과 충돌하지 않음

특히 서버 애플리케이션에서는 commonPool을 다른 비동기 작업도 같이 쓰는 경우가 많아, 병렬 스트림이 예기치 않게 전체 지연을 늘릴 수 있습니다. 이럴 땐 병렬 스트림 대신 명시적인 ExecutorService 또는 CompletableFuture로 격리된 풀을 쓰는 편이 운영 안정성에 유리합니다.

운영 환경에서 “공유 자원 때문에 꼬이는 문제”는 병렬 처리 전반에서 반복됩니다. 예를 들어 분산 환경에서 재시도/중복 실행을 다루는 관점은 Saga 패턴에서 보상 트랜잭션 중복 실행 막는 법에서도 유사한 결로 등장합니다.

안전한 체크리스트

아래 질문에 하나라도 예라면 병렬 스트림을 재검토하세요.

  1. 람다 내부에서 외부 컬렉션에 add, put을 하는가
  2. 외부 카운터/플래그/배열 값을 변경하는가
  3. 스레드 안전하지 않은 객체를 공유하는가
  4. synchronized로 감싸서 “동작만” 맞추려 하는가
  5. 결과 순서가 중요한데 forEach를 쓰는가(필요하면 forEachOrdered지만 성능 감소)

그리고 가능하면 다음 형태로 바꾸는 것을 우선합니다.

  • forEach로 누적하지 말고 collect로 수집
  • count, sum, max 같은 내장 리덕션 활용
  • 상태가 필요하면 groupingBy, groupingByConcurrent, toMap 같은 Collector 사용

재현용 미니 테스트: 왜 간헐적으로 터지나

공유 리스트 add 버그는 데이터가 작으면 잘 안 터져서 더 위험합니다. 아래처럼 반복 실행하면 재현 확률이 올라갑니다.

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

public class FlakyParallelBug {
    public static void main(String[] args) {
        List<Integer> input = IntStream.range(0, 200_000).boxed().toList();

        for (int i = 0; i < 50; i++) {
            List<Integer> out = new ArrayList<>();
            try {
                input.parallelStream().forEach(out::add);
                if (out.size() != input.size()) {
                    System.out.println("Mismatch at run " + i + ": " + out.size());
                    break;
                }
            } catch (Exception e) {
                System.out.println("Exception at run " + i + ": " + e);
                break;
            }
        }
    }
}

이런 류의 버그는 “가끔” 발생하기 때문에 로그/모니터링에서 이상 징후만 보이고 원인 파악이 늦어집니다. 운영에서 간헐 장애를 줄이기 위한 관점은 K8s CrashLoopBackOff 원인별 진단·해결 체크리스트처럼 체크리스트로 체계화하는 것이 효과적입니다.

결론: 병렬화보다 먼저 순수함수와 수집을 지켜라

parallelStream()의 함정은 “병렬이라서 어렵다”가 아니라, 공유 상태를 섞는 순간 프로그램 모델이 바뀐다는 데 있습니다. 스트림 파이프라인은 가능한 한 순수 함수로 유지하고, 결과는 collect/리덕션으로 모으세요.

  • 외부 상태 변경이 보이면 일단 리팩터링 대상으로 본다
  • 병렬 수집은 groupingByConcurrent 같은 도구를 사용한다
  • 서버에서는 commonPool 경쟁까지 고려해 병렬 스트림 사용을 결정한다

성능은 중요하지만, 운영에서 “가끔 틀리는 결과”는 그 어떤 성능 향상보다 비용이 큽니다. 병렬 스트림을 적용하기 전후로 정확성 테스트와 부하 테스트를 반드시 같이 돌리는 것을 권장합니다.

Discuss on Twitter • View on GitHub