작업 훔치기와 병렬 Stream

ForkJoinPool 작업자는 보통 자기 덱의 최근 작업을 처리하고, 일이 없으면 다른 작업자의 오래된 작업을 가져옵니다. 이러한 작업 훔치기(work stealing)는 하나의 중앙 큐에서 모든 작업자가 경쟁하는 비용을 줄이고 불균형한 분할을 완화합니다. 하지만 어느 작업자가 어느 원소를 언제 처리하는지는 기능 규칙이 아닙니다. 병렬 Stream도 이 실행기 위에서 동작하므로 처리 순서와 결과 순서를 구분해야 합니다.

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

아래 예제는 두 작업이 같은 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는 처리 순서를 가집니다. 병렬 작업자의 처리 순서는 달라도 toList() 같은 정렬된 최종 연산은 결과 순서를 유지합니다. forEach는 처리 완료 순서로 부수 효과를 내보낼 수 있고 forEachOrdered는 처리 순서를 보존하는 대신 조정이 필요합니다.

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());
    }
}

collected=[50, 10, 40, 20, 30]과 ordered=5 1 4 2 3은 안정적인 결과입니다. completionOrder의 구체적인 순서는 일부러 검증하지 않습니다. StringBuilder는 forEachOrdered 콜백이 순서대로 호출되는 경로에서만 사용했으며 일반 병렬 콜백에서는 스레드 안전하지 않습니다.

unordered 적용 조건

unordered()는 원본의 순서를 임의로 섞는 연산이 아니라 하위 수집기가 처리 순서를 유지하지 않아도 된다고 알립니다. distinct, limit, 수집기가 순서 제약에서 벗어나 최적화될 여지가 생깁니다. 결과 순서가 API 사용자에게 보이면 적용하지 않습니다.

Collector 특성과 병렬 축약

Collector의 공급자, 누산기, 결합 함수, 마무리 함수는 32장에서 살펴본 축약 규칙입니다. 병렬 실행에서는 수집기 특성이 추가 정보를 제공합니다.

CONCURRENT: 여러 스레드가 같은 결과 컨테이너에 동시에 누적할 수 있다.
UNORDERED: 결과가 처리 순서에 의존하지 않는다.
IDENTITY_FINISH: 누산기 타입을 별도 변환 없이 결과로 반환할 수 있다.

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);
    }
}

마무리 함수가 새 Map을 만들기 때문에 IDENTITY_FINISH는 선언하지 않습니다. UNORDERED는 빈도 결과가 입력 순서와 무관하기 때문에 정직합니다. 표준 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이고 실행 순서나 작업자 이름을 기대값에 넣지 않는 것입니다. 결과가 화면에서 정렬돼야 한다면 병렬 집계가 끝난 뒤 항목을 정렬합니다.