작업 훔치기와 병렬 Stream

기본 모드의 ForkJoinPool 작업자는 보통 자기 덱의 최근 작업을 처리하고, 일이 없으면 다른 작업자의 오래된 작업을 가져옵니다. asyncMode에서는 로컬 작업을 FIFO로 처리할 수 있습니다.

이러한 작업 훔치기(work stealing)는 하나의 중앙 큐에서 모든 작업자가 경쟁하는 비용을 줄이고 불균형한 분할을 완화합니다.

하지만 어느 작업자가 어느 원소를 언제 처리하는지는 기능 규칙이 아닙니다.

JDK 25의 병렬 Stream 구현도 Fork/Join 작업을 사용하지만, Stream API가 특정 풀이나 작업자 배치를 지정하는 것은 아닙니다. 처리 순서와 결과 순서를 구분해야 합니다.

병렬 콜백의 공유 누산기 경쟁

아래 예제는 두 콜백이 같은 int 값을 모두 읽은 뒤에야 기록할 수 있도록 배리어를 둡니다. 쓰기가 같은 순간에 일어나야 하는 것은 아닙니다.

원문의 읽기·증가·쓰기 조합은 value[0]++와 마찬가지로 원자적이지 않습니다.

두 콜백이 배리어를 정상 통과해 완료하면 모두 0을 읽고 1을 쓰므로 wrong-count=1이 됩니다. 배리어 대기는 추가 작업자가 실제로 실행돼야 진행하므로, 이 예제를 모든 병렬 실행 환경의 종료 보장으로 삼지는 않습니다.

lab/ParallelSharedCounterBug.java

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

public final class ParallelSharedCounterBug {
    public static void main(String[] args) throws Exception {
        int[] value = {0};
        CyclicBarrier bothRead = new CyclicBarrier(2);

        try (ForkJoinPool pool = new ForkJoinPool(2)) {
            pool.submit(
                            () ->
                                    List.of(10, 20).parallelStream()
                                            .forEach(
                                                    ignored -> {
                                                        int observed = value[0];
                                                        await(bothRead);
                                                        value[0] = observed + 1;
                                                    }))
                    .get();
        }

        System.out.println("wrong-count=" + value[0]);
    }

    private static void await(CyclicBarrier barrier) {
        try {
            barrier.await();
        } catch (InterruptedException exception) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException(exception);
        } catch (BrokenBarrierException exception) {
            throw new IllegalStateException(exception);
        }
    }
}

원칙은 콜백 밖의 가변 상태를 여러 작업자가 함께 갱신하지 않는 것입니다.

개수는 count(), 합계는 sum() 또는 결합 법칙을 만족하는 reduce, 여러 결과는 수집기로 모읍니다.

AtomicInteger로 데이터 경합만 막더라도 모든 원소가 하나의 원자적 변수에서 경쟁하므로 병렬 처리의 장점을 잃을 수 있습니다.

작업 스레드 덱과 훔치기 과정 관찰

작업 훔치기의 목적은 작업 분배이지 결과 순서 보장이 아닙니다.

다음 작업은 크기가 서로 다른 리프를 만들고 작업자 이름별 처리 개수를 기록합니다.

정확한 분포는 실행마다 달라도 합계는 항상 동일해야 합니다.

src/WorkStealingDistribution.java

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.atomic.LongAdder;

public final class WorkStealingDistribution {
    private static final class SumTask extends RecursiveTask<Long> {
        private final int from;
        private final int to;
        private final Map<String, LongAdder> counts;

        private SumTask(int from, int to, Map<String, LongAdder> counts) {
            this.from = from;
            this.to = to;
            this.counts = counts;
        }

        @Override
        protected Long compute() {
            if (to - from <= 4) {
                counts.computeIfAbsent(Thread.currentThread().getName(), key -> new LongAdder())
                        .increment();
                long sum = 0;
                for (int value = from; value < to; value++) {
                    sum += value;
                }
                return sum;
            }
            int middle = from + Math.max(1, (to - from) / 3);
            SumTask left = new SumTask(from, middle, counts);
            SumTask right = new SumTask(middle, to, counts);
            left.fork();
            return right.compute() + left.join();
        }
    }

    public static void main(String[] args) {
        Map<String, LongAdder> counts = new ConcurrentHashMap<>();
        try (ForkJoinPool pool = new ForkJoinPool(4)) {
            long sum = pool.invoke(new SumTask(1, 65, counts));
            System.out.println("sum=" + sum);
        }
        counts.forEach((worker, leaves) -> System.out.println(worker + "=" + leaves.sum()));
    }
}

계측 Map은 동시 갱신을 지원하지만 계산 결과는 작업 반환값으로 합칩니다.

작업자 분포를 단언으로 고정하면 스케줄러와 머신 차이에 따라 깨집니다.

확인할 결과는 1~64 합인 2080입니다. 인덱스가 겹치지 않는지는 분할 코드를 함께 확인해야 하며, 출력 합계 하나가 누락·중복의 부재까지 증명하지는 않습니다. 작업자별 카운터는 실행한 리프 수이고 작업 훔치기 횟수는 아닙니다.

처리 순서와 결과 순서

List는 만남 순서(encounter order)를 가진 원본이며, 그 원소들이 크기순으로 정렬돼 있어야 하는 것은 아닙니다.

src/ParallelEncounterOrder.java

import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;

public final class ParallelEncounterOrder {
    public static void main(String[] args) {
        List<Integer> source = List.of(5, 1, 4, 2, 3);

        List<Integer> collected = source.parallelStream().map(value -> value * 10).toList();

        ConcurrentLinkedQueue<Integer> completionOrder = new ConcurrentLinkedQueue<>();
        source.parallelStream().forEach(completionOrder::add);

        StringBuilder orderedText = new StringBuilder();
        source.parallelStream().forEachOrdered(value -> orderedText.append(value).append(' '));

        System.out.println("collected=" + collected);
        System.out.println("completion-elements=" + completionOrder.size());
        System.out.println("ordered=" + orderedText.toString().trim());
    }
}

입력 [5, 1, 4, 2, 3]의 만남 순서를 각 최종 연산이 어떻게 다루는지 비교합니다.

병렬 콜백과 최종 결과의 순서 계약
원문의 최종 연산	순서 계약	확인할 결과
toList()	만남 순서로 결과 목록 구성	[50, 10, 40, 20, 30]
forEach()	콜백 순서 보장 없음	큐 크기 5 · 삽입 순서는 고정하지 않음
forEachOrdered()	만남 순서로 콜백을 하나씩 실행	5 1 4 2 3

toList(): 순서 계약: 만남 순서로 결과 목록 구성
확인할 결과: [50, 10, 40, 20, 30]
forEach(): 순서 계약: 콜백 순서 보장 없음
확인할 결과: 큐 크기 5 · 삽입 순서는 고정하지 않음
forEachOrdered(): 순서 계약: 만남 순서로 콜백을 하나씩 실행
확인할 결과: 5 1 4 2 3

원문의 completionOrder는 큐에 값을 삽입한 순서를 담습니다. 이름과 달리 콜백 전체의 완료 시각을 계측한 기록은 아니며, 출력도 큐 크기만 보여 줍니다.

StringBuilder는 forEachOrdered가 앞 콜백의 동작을 다음 콜백보다 happens-before로 보장하는 경로에서만 사용했습니다. 모든 콜백이 같은 스레드에서 실행된다는 뜻은 아니며, 일반 병렬 콜백에서는 스레드 안전하지 않습니다.

unordered 적용 조건

unordered()는 원소를 임의로 섞는 연산이 아니라 이후 파이프라인에서 만남 순서를 유지해야 한다는 제약을 해제합니다.

distinct, limit, 수집기가 순서 제약에서 벗어나 최적화될 여지가 생깁니다.

결과 순서가 API 사용자에게 보이면 적용하지 않습니다.

Collector 특성과 병렬 축약

Collector의 공급자, 누산기, 결합 함수, 마무리 함수는 32장에서 살펴본 축약 규칙입니다.

병렬 실행에서는 수집기 특성이 추가 정보를 제공합니다.

CONCURRENT를 선언하려면 컨테이너와 누산기가 실제로 동시 갱신에 안전해야 합니다.

일반 HashMap에 선언만 붙이면 데이터 경합을 프레임워크에 숨기는 셈입니다.

다음 사용자 정의 수집기는 ConcurrentHashMap과 LongAdder를 사용하고 완성 시 불변 개수 Map으로 바꿉니다.

src/ConcurrentFrequencyCollector.java

import java.util.EnumSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.BiConsumer;
import java.util.function.BinaryOperator;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collector;

public final class ConcurrentFrequencyCollector {
    private static final class Frequencies
            implements Collector<String, ConcurrentHashMap<String, LongAdder>, Map<String, Long>> {

        @Override
        public Supplier<ConcurrentHashMap<String, LongAdder>> supplier() {
            return ConcurrentHashMap::new;
        }

        @Override
        public BiConsumer<ConcurrentHashMap<String, LongAdder>, String> accumulator() {
            return (map, word) -> map.computeIfAbsent(word, key -> new LongAdder()).increment();
        }

        @Override
        public BinaryOperator<ConcurrentHashMap<String, LongAdder>> combiner() {
            return (left, right) -> {
                right.forEach(
                        (word, count) ->
                                left.computeIfAbsent(word, key -> new LongAdder())
                                        .add(count.sum()));
                return left;
            };
        }

        @Override
        public Function<ConcurrentHashMap<String, LongAdder>, Map<String, Long>> finisher() {
            return mutable ->
                    mutable.entrySet().stream()
                            .collect(
                                    java.util.stream.Collectors.toUnmodifiableMap(
                                            Map.Entry::getKey, entry -> entry.getValue().sum()));
        }

        @Override
        public Set<Characteristics> characteristics() {
            return EnumSet.of(Characteristics.CONCURRENT, Characteristics.UNORDERED);
        }
    }

    public static void main(String[] args) {
        Map<String, Long> counts =
                java.util.stream.Stream.of("java", "stream", "java", "fork", "stream", "java")
                        .parallel()
                        .collect(new Frequencies());
        System.out.println(counts);
    }
}

빈도 수집기의 특성과 구현 근거
특성의 의미	원문의 선언	구현 근거
같은 컨테이너에 동시 누적	CONCURRENT	ConcurrentHashMap + LongAdder
결과의 순서 무관	UNORDERED	같은 단어의 개수는 입력 순서와 무관
항등 마무리 함수	IDENTITY_FINISH 없음	누산 맵을 수정 불가 Map<String, Long>으로 변환

같은 컨테이너에 동시 누적: 원문의 선언: CONCURRENT
구현 근거: ConcurrentHashMap + LongAdder
결과의 순서 무관: 원문의 선언: UNORDERED
구현 근거: 같은 단어의 개수는 입력 순서와 무관
항등 마무리 함수: 원문의 선언: IDENTITY_FINISH 없음
구현 근거: 누산 맵을 수정 불가 Map<String, Long>으로 변환

이 수집기는 CONCURRENT와 UNORDERED를 함께 선언하므로 같은 컨테이너에 동시 누적할 수 있지만, 구현이 반드시 그 경로를 택해야 하는 것은 아닙니다. 출력은 java=3, stream=2, fork=1의 매핑으로 읽고 키의 인쇄 순서는 고정하지 않습니다.

표준 groupingByConcurrent가 요구를 충족하면 사용자 정의 수집기보다 먼저 선택합니다.

forEachOrdered를 쓰면 콜백을 병렬로 안전하게 바꿀 필요가 없나요?

순서를 보존하는 최종 연산이라는 사실만으로 외부 상태 변경이 좋은 설계가 되지는 않습니다.

콜백 이전의 중간 연산은 병렬로 실행될 수 있고 예외 뒤에는 일부 부수 효과가 남을 수 있습니다.

결과를 컬렉션으로 축약한 뒤 단일 스레드에서 외부 시스템에 반영하면 재시도와 롤백 경계가 더 명확합니다.

연습 문제

여러 문서의 태그를 병렬로 펼치고 태그별 개수를 세세요.

공유 HashMap을 직접 수정하지 말고 표준 동시성 수집기를 사용하며, 결과 키 순서를 요구하지 않습니다.

해설 보기

exercise/ParallelTagFrequencySolution.java

import java.util.List;
import java.util.concurrent.ConcurrentMap;
import java.util.function.Function;
import java.util.stream.Collectors;

public final class ParallelTagFrequencySolution {
    private record Document(String title, List<String> tags) {}

    public static void main(String[] args) {
        List<Document> documents =
                List.of(
                        new Document("stream", List.of("java", "api")),
                        new Document("pool", List.of("java", "concurrency")),
                        new Document("collector", List.of("api", "java")));

        ConcurrentMap<String, Long> counts =
                documents.parallelStream()
                        .flatMap(document -> document.tags().stream())
                        .unordered()
                        .collect(
                                Collectors.groupingByConcurrent(
                                        Function.identity(), Collectors.counting()));

        System.out.println("java=" + counts.get("java"));
        System.out.println("api=" + counts.get("api"));
        System.out.println("concurrency=" + counts.get("concurrency"));
    }
}

종료 기준은 각 태그의 개수가 3, 2, 1이고 실행 순서나 작업자 이름을 기대값에 넣지 않는 것입니다.

결과가 화면에서 정렬돼야 한다면 병렬 집계가 끝난 뒤 항목을 정렬합니다.