병렬 Stream 결합 법칙

병렬 Stream은 원본을 Spliterator로 나누고 Fork/Join 작업으로 처리한 뒤 부분 결과를 결합합니다. 문법은 간단하지만 정확성에는 더 강한 조건이 붙습니다. 연산은 공유 상태를 변경하지 않아야 하고, 축약은 분할 위치와 결합 순서가 달라도 같은 결과를 만들어야 합니다. 성능은 데이터 크기, 원소당 비용, 분할 가능성, 순서 제약, 풀 경쟁을 포함한 측정 문제입니다.

뺄셈 `reduce`의 결합 법칙 위반

뺄셈은 결합 법칙을 만족하지 않습니다. 순차 계산은 왼쪽부터 누적하지만 병렬 계산은 부분값을 만든 뒤 서로 다른 순서로 결합합니다. 아래 실행은 같은 입력에서 wrong-same=false를 출력합니다.

lab/NonAssociativeParallelReduceBug.java

import java.util.concurrent.ForkJoinPool;
import java.util.stream.IntStream;

public final class NonAssociativeParallelReduceBug {
    public static void main(String[] args) throws Exception {
        int sequential = IntStream.rangeClosed(1, 8).reduce(0, (left, right) -> left - right);

        int parallel;
        try (ForkJoinPool pool = new ForkJoinPool(4)) {
            parallel =
                    pool.submit(
                                    () ->
                                            IntStream.rangeClosed(1, 8)
                                                    .parallel()
                                                    .reduce(0, (left, right) -> left - right))
                            .get();
        }

        System.out.println("sequential=" + sequential);
        System.out.println("parallel=" + parallel);
        System.out.println("wrong-same=" + (sequential == parallel));
    }
}

원칙은 항등값과 누산기·결합 함수가 하나의 일관된 대수 구조를 이루는 것입니다. 덧셈의 0, 곱셈의 1처럼 항등값은 어느 부분 결과와 결합해도 값을 바꾸지 않아야 합니다. a + b, min, max, 불변 컬렉션 결합처럼 결합 법칙을 만족하는 연산도 오버플로, 부동소수점 반올림, 순서 요구 사항을 별도로 고려합니다.

순차·병렬 결과 속성 검사

몇 개 예제의 결과가 같다고 법칙이 증명되는 것은 아니지만 다양한 분할 크기와 입력을 비교하면 결합자 오류를 일찍 찾을 수 있습니다. 다음 예제는 합계와 최댓값을 순차·병렬로 계산해 동등성을 검사합니다.

src/AssociativeReductionContract.java

import java.util.List;
import java.util.concurrent.ForkJoinPool;

public final class AssociativeReductionContract {
    private static int sequentialSum(List<Integer> values) {
        return values.stream().reduce(0, Integer::sum);
    }

    private static int parallelSum(List<Integer> values, ForkJoinPool pool) throws Exception {
        return pool.submit(() -> values.parallelStream().reduce(0, Integer::sum)).get();
    }

    public static void main(String[] args) throws Exception {
        List<List<Integer>> cases =
                List.of(
                        List.of(),
                        List.of(7),
                        List.of(3, -2, 8, 1),
                        java.util.stream.IntStream.rangeClosed(1, 100).boxed().toList());

        try (ForkJoinPool pool = new ForkJoinPool(4)) {
            for (List<Integer> values : cases) {
                int expected = sequentialSum(values);
                int actual = parallelSum(values, pool);
                if (expected != actual) {
                    throw new AssertionError("mismatch: " + values.size());
                }
                System.out.println("size=" + values.size() + ", sum=" + actual);
            }
        }
    }
}

운영 입력 검사에서는 난수 입력 시드를 기록하고 오버플로 정책까지 정합니다. 합계가 int 범위를 넘을 수 있으면 long, Math.addExact, BigInteger 중 요구에 맞는 선택을 합니다.

공용 풀의 공유 범위

parallelStream()은 일반적으로 ForkJoinPool.commonPool()을 사용합니다. 병렬성은 대개 사용 가능한 프로세서 수를 바탕으로 정하지만 실행 환경 설정과 컨테이너의 CPU 할당량에 영향을 받습니다. 요청 하나만 빨라지는지보다 애플리케이션의 다른 병렬 Stream, CompletableFuture의 기본 비동기 단계, 라이브러리 작업과 같은 풀에서 경쟁하는지 살펴봐야 합니다.

src/CommonPoolInventory.java

import java.util.concurrent.ForkJoinPool;
import java.util.stream.IntStream;

public final class CommonPoolInventory {
    public static void main(String[] args) {
        ForkJoinPool common = ForkJoinPool.commonPool();
        int result = IntStream.rangeClosed(1, 1_000).parallel().map(value -> value * 2).sum();

        System.out.println("parallelism=" + common.getParallelism());
        System.out.println("pool-size=" + common.getPoolSize());
        System.out.println("queued-submissions=" + common.getQueuedSubmissionCount());
        System.out.println("checksum=" + result);
    }
}

이 숫자는 관찰값이지 고정된 기대값이 아닙니다. main 스레드 같은 호출자도 계산에 참여할 수 있어 작업자 개수와 실제 참여 스레드 수가 같지 않습니다. JVM 시스템 속성 하나로 공용 풀의 병렬성을 바꾸면 프로세스 전체에 영향을 주므로 지역적인 최적화처럼 적용하지 않습니다.

전용 풀의 자원 소유권

사용자 정의 ForkJoinPool의 작업자 안에서 병렬 Stream을 실행하면 현재 JDK 구현에서는 그 풀을 사용하는 방식이 널리 활용됩니다. 그러나 API는 풀을 매개변수로 직접 받도록 설계되지 않았습니다. 실행기를 명시적으로 제어하는 것이 핵심이라면 CompletableFuture, 직접 만든 ForkJoinTask, 실행기에 제출한 분할 작업이 더 분명합니다. 전용 풀은 생성 위치, 수명, shutdown, 요청 간 공유 정책을 문서화합니다.

성능 측정의 예열·결과 소비

한 번 nanoTime으로 잰 값은 JIT 컴파일, 클래스 로딩, CPU 주파수, GC에 크게 흔들립니다. 아래의 작은 측정 틀은 벤치마크 도구를 대신하지 않지만 최소한 예열, 반복 측정, 체크섬 소비, 순차·병렬 경로의 동일 입력이라는 조건을 보여 줍니다.

src/ParallelMeasurementPrerequisites.java

import java.util.Arrays;

public final class ParallelMeasurementPrerequisites {
    private static long work(int value) {
        long result = value;
        for (int round = 0; round < 20; round++) {
            result = (result * 1_664_525L + 1_013_904_223L) & 0x7fff_ffffL;
        }
        return result;
    }

    private static long sequential(int[] values) {
        return Arrays.stream(values).mapToLong(ParallelMeasurementPrerequisites::work).sum();
    }

    private static long parallel(int[] values) {
        return Arrays.stream(values)
                .parallel()
                .mapToLong(ParallelMeasurementPrerequisites::work)
                .sum();
    }

    private static long measure(int[] values, boolean useParallel) {
        long start = System.nanoTime();
        long checksum = useParallel ? parallel(values) : sequential(values);
        long elapsed = System.nanoTime() - start;
        System.out.println(
                (useParallel ? "parallel" : "sequential")
                        + " ns="
                        + elapsed
                        + ", checksum="
                        + checksum);
        return checksum;
    }

    public static void main(String[] args) {
        int[] values = java.util.stream.IntStream.range(0, 50_000).toArray();
        for (int warmup = 0; warmup < 3; warmup++) {
            sequential(values);
            parallel(values);
        }

        long first = 0;
        long second = 0;
        for (int round = 0; round < 3; round++) {
            first = measure(values, false);
            second = measure(values, true);
        }
        if (first != second) {
            throw new AssertionError("different checksum");
        }
    }
}

실제 의사결정에는 JMH처럼 죽은 코드 제거 방지, 별도 JVM 프로세스, 예열과 측정 반복을 체계적으로 다루는 도구를 사용합니다. 평균만 보지 말고 백분위수와 분산을 확인하며, 운영 환경과 같은 코어 제한과 동시 부하에서 측정합니다.

병렬화하기 좋은 작업과 나쁜 작업

큰 기본형 배열처럼 균등하게 분할되고 원소당 CPU 계산이 충분하면 후보입니다.
연결 구조, 작은 컬렉션, 값싼 접근자는 분할·결합 비용이 이득을 넘기 쉽습니다.
정렬된 limit, findFirst, 상태 보유 sorted는 조정과 버퍼링을 요구합니다.
블로킹 I/O는 공용 풀 작업자를 묶어 다른 CPU 작업까지 지연시킬 수 있습니다.
외부 API 호출, 데이터베이스 쓰기, 로그 추가 같은 부수 효과에는 재시도·순서·부분 실패 규칙이 필요합니다.

I/O 동시성은 가상 스레드와 제한된 세마포어, 비동기 클라이언트, 전용 실행기처럼 자원 종류에 맞는 도구를 검토합니다. CPU 병렬성과 블로킹 동시성을 같은 병렬 Stream 설정으로 해결하지 않습니다.

병렬이 한 번 더 빨랐으면 바로 적용해도 되나요?

아닙니다. 결과 동등성을 먼저 확인하고 별도 JVM 프로세스, 충분한 예열, 반복 측정, 대표 데이터 크기, 동시 요청 부하를 포함해야 합니다. 적용 뒤에는 처리량뿐 아니라 꼬리 지연 시간, CPU 사용률, 공용 풀 큐와 다른 기능의 지연도 관찰합니다.

연습 문제

입력이 10,000개 미만이면 순차, 그 이상이면 병렬로 부호 없음 변환 합계를 계산하세요. 두 경로가 같은 함수를 사용하고 결과가 동일해야 합니다.

해설 보기

exercise/ThresholdParallelChecksumSolution.java

import java.util.Arrays;
import java.util.stream.IntStream;

public final class ThresholdParallelChecksumSolution {
    private static final int PARALLEL_THRESHOLD = 10_000;

    private static long normalize(int value) {
        return Integer.toUnsignedLong(value * 31 + 17);
    }

    private static long checksum(int[] values) {
        IntStream stream = Arrays.stream(values);
        if (values.length >= PARALLEL_THRESHOLD) {
            stream = stream.parallel();
        }
        return stream.mapToLong(ThresholdParallelChecksumSolution::normalize).sum();
    }

    public static void main(String[] args) {
        int[] small = IntStream.range(0, 100).toArray();
        int[] large = IntStream.range(0, 20_000).toArray();
        System.out.println("small=" + checksum(small));
        System.out.println("large=" + checksum(large));
    }
}

임계값은 고정된 상수가 아니라 측정으로 얻은 정책입니다. 작업 부하와 하드웨어가 달라지면 다시 측정하고 설정으로 조정할 수 있게 합니다.

뺄셈 reduce의 결합 법칙 위반