07. 병렬 데이터 처리와 성능

간단 정리#

내부 반복을 이용하면 병렬로 처리하기 쉬움
항상 병렬처리가 빠른건 아님
병렬처리 사용 시 성능은 직접 측정을 권장
데이터가 아주 많거나, 요소당 처리시간이 길면 병렬성으로 이점을 누릴 수 있음
기본형 특화 스트림, 오바른 자료구조 선택이 연산 병렬화 보다 성능적으로 더 큰 영향을 미칠 수 있음
포크/조인 프레임워크는
- 병렬화 할 수 있는 태스크를 작은 태스크로 분할 후
- 각 태스크를 개별 스레드에 실행하며
- 마지막에 각 서브태스크의 결과를 합쳐 최종결과를 생산함
Spliterator는 스트림을 어떻게 병렬화할건지 정의함

병렬 스트림으로 데이터를 병렬 처리하기#

stream.parallel() 은 내부적으로 청크단위로 나누어 계산
- 스레드는 ForkJoinPool 을 이용해 관리
- Runtime.getRumtime().availableProcessors() 가 반환하는 값으로 세팅
병렬화는 완전 공짜가 아님
- 스트림을 재귀적으로 분할
- 각 서브스트림을 서로 다른 스레드의 리듀싱 연산으로 할당
- 이들 결과를 하나의 값으로 합쳐야 함
멑리코어간의 데이터 이동은 비싸다
병렬 스트림이 올바로 동작하려면 공유된 가변 상태를 피해야 함

병렬 스트림 사용시 주의 사항#

확신이 서지 않으면 직접 측정
박싱을 주의
순차 스트림보다 병렬 스트림 성능이 떨어지느 연산이 있음
- limit, findFirst (unordered limit 은 괜찮음)
스트림에서 수행하는 전체 파이프라인 연산 비용을 고려하라
소량의 데이터에선 병렬스트림이 도움 안됨
자료구조 적절한지 확인
- LikedList 보단 ArrayList 가 랜덤 엑세스 가능하므로 좋음
중간 연산이 어떻게 스트림 요소를 바꾸는지 주의
최종연산의 병합과정 비용을 살펴보라

분해성 좋은 스트림 소스#

ArrayList (best)
IntStream.range (best)
HashSet (good)
TreeSet (good)

병렬 스트림의 성능 분석#

public long parallelSum(long n) {
    return Stream.iterate(1L, i -> i + 1)
            .limit(n)
            .parallel()
            .reduce(0L, Long::sum);
}

public long sequentialSum(long n) {
    return Stream.iterate(1L, i -> i + 1)
            .limit(n)
            .reduce(0L, Long::sum);
}

public long iterativeSum(long n) {
    long result = 0;
    for (long i = 1; i <= n; i++)
        result += i;
    return result;
}

그런데 iterative가 가장 빠르다...?
위의 예제 문제점
- 반복 결과로 박싱된 객체가 만들어지므로 숫자를 더하려면 언박싱을 해야함
- 반복 작업은 병렬로 수행할 수 있는 독립 단위로 나누기가 어려움
해결방안: 청크단위로 분할할 수 있고 primitive type을 사용해서 언박싱을 회피할 수 있도록 LongStream.rangeClosed() 를 사용

public long rangedParallelSum(long n) {
    return LongStream.rangeClosed(1L, n)
            .parallel()
            .sum();
}

public long rangedSequentialSum(long n) {
    return LongStream.rangeClosed(1L, n).sum();
}

100회 반복, n = 10_000_000 입력시 경과 시간
- iterativeSum: 660ms
- sequentialSum: 16s 701ms
- parallelSum: 51s 882ms
- rangedSequentialSum: 886ms
- rangedParallelSum: 287ms

포크/조인 프레임워크#

Java 7
병렬화 할 수 있는 작업을 재귀적으로 작은 작업으로 분할
이후 서브 태스크 각각의 결과를 합쳐 전체 결과를 만들도록 설계
자세한 구현은 예제에서 확인
forJoinSum 경과시간 (100회 반복, n = 10_000_000): 3s 4445ms
느려졌다?
- 전체 스트림을 long[] 으로 변환하는 과정이 있었기 때문

포크/조인 사용 시 주의사항#

join은 서브태스크가 종료될때 까지 호출자를 블럭시킴
RecursiveTask 내에서는 ForJoinPool의 invoke 메서드 사용 X
서버태스크에 fork 메서드를 호출해 ForJoinPool의 일정을 조절할 수 있음
- 예제에서 left, right 둘다 fork하는 것 보단 한족에선 compute 로 재활용이 좋음
병렬계산은 디버깅이 어려움
순차 처리보다 무조건 빠르지 않음

작업 훔치기#

실제로는 코어 개수와 상관없이 적절한 크기로 분할된 많은 태스크를 forking 하는 것이 바람직
작업 훔치기 (work stealing)
- ForJoinPool의 모든 스레드를 거의 공정하게 분할 함
- 어떤 스레드가 작업이 끝나면 작업 대기 큐에서 바로 꺼내 작업함
- 태스크 분할 시 작업 대기 큐로 바로 전송

Spliterator로 스트림 데이터 쪼개기#

Java 8
스트림을 자동으로 분할하는 기법 Spliterator
분할할 수 있는 자동 반복자
병렬 작업에 특화

// Spliterator 인터페이스
public interface Spliterator<T> {
    // 요소를 소비하며 남아있는 요소가 있는지 boolean 반환
    boolean tryAdvance(Consumer<? super T> action);

    // 일부 요소를 분할해서 두 번째 Spliterator 를 생성
    Spliterator<T> trySplit();

    // 탐색해야할 요소 수
    long estimateSize();

    // trySplit 해서 나눠질 때의 특징
    int characteristics();
}