[모던 자바 인 액션] 7장 병렬 데이터 처리와 성능

✏️[모던 자바 인 액션, 전문가를 위한 자바 8, 9, 10 기법 가이드] 스터디 관련 책 내용을 정리한 글입니다.

이 장의 내용

---

7.1 병렬 스트림

병렬 스트림이란, 각각의 스레드에서 처리할 수 있도록 스트림 요소를 여러 청크로 분할한 스트림입니다.

따라서, 병렬 스트림을 이용하면 모든 멀티코어 프로세서가 각각의 청크를 처리하도록 할당할 수 있습니다.

예를 들어, 숫자 n을 인수로 1부터 n까지의 모든 숫자의 합계를 반환하는 메서드를 다음처럼 구현할 수 있습니다.

public long sequentialSum(long n) {
    return Stream.iterate(1L, i -> i + 1) // 무한 자연수 스트림을 생성합니다.
                 .limit(n) // n개 이하로 제한
                 .reduce(0L, Long::sum); // 모든 숫자를 더하는 스트림 리듀싱 연산
}

전통적인(?!) 자바에서는 다음처럼 반복문으로 구현할 수 있습니다.

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

n이 커진다면 위 연산을 병렬로 처리하는 것이 좋습니다. 그렇다고 하면 무엇부터 건드려야 하는지, 결과 변수는 어떻게 동기화할지? 그리고 몇 개의 스레드를 사용하는 것이 좋고, 숫자는 어떻게 생성하고, 생성된 숫자는 누가 더할지? 생각이 듭니다.

하지만, 병렬 스트림을 이용하면 걱정없이 위 모든 문제를 쉽게 해결할 수 있습니다.

7.1.1 순차 스트림을 병렬 스트림으로 변환하기

순차 스트림에 parallel 메서드를 호출하면 기존의 함수형 리듀싱 연산이 병렬로 처리됩니다.

public long parallelSum(long n) {
    return Stream.iterate(1L, i -> i + 1)
                 .limit(n)
                 .parallel() // 스트림을 병렬 스트림으로 변환합니다.
                 .reduce(0L, Long::sum);

이전 코드와 달리 스트림이 여러 청크로 분할되어 각각 리듀싱 연산을 수행한 후 다시 리듀싱 연산으로 합쳐져 전체 스트림의 리듀싱 결과를 도출합니다. 

순차 스트림에 parallel을 호출해도 스트림 자체에서는 아무 변화도 일어나지 않습니다. 내부적으로는 parallel을 호출하면 이후 연산이 병렬로 수행해야 함을 의미하는 불리언 플래그가 설정되고, 반대로 sequential로 병렬 스트림을 순차 스트림으로 바꿀 수 있습니다. 이 두 메서드를 이용해서 어떤 연산을 병렬로 실행하고 어떤 연산을 순차로 실행할 지 제어할 수 있습니다.

stream.parallel()
      .filter(...)
      .sequential()
      .map(...)
      .parallel()
      .reduce();

parallel과 sequential 두 메서드 중 최종적으로 호출된 메서드가 전체 파이프라인에 영향을 미칩니다.

병렬 스트림에서 사용하는 스레드 풀 설정
병렬 스트림은 내부적으로 ForkJoinPool을 사용합니다. 기본적으로 ForkJoinpool은 프로세서 수, 즉  Runtime.getRuntime().availableProcessors()가 반환하는 값에 상응하는 스레드를 갖습니다.

System.setProperty("java.util.concurrent.ForkJoinPool.common.parallism", "12");
위 예제는 전역 설정 코드로 이후의 모든 병렬 스트림 연산에 영향을 줍니다.
현재는 하나의 병렬 스트림에 사용할 수 있는 특정한 값을 지정할 수 없습니다. 일반적으로 기기의 프로세서 수와 같으므로 특별한 이유가 없다면 ForkJoinPool의 기본값을 그대로 사용할 것을 권장합니다.

7.1.2 스트림 성능 측정

병렬화를 이용하면 순차나 반복 형식에 비해 성능이 더 좋아질 거라고 추측하지만, 소프트웨어 공학에서 추측은 굉장히 위험한 방법입니다. 특히, 성능을 최적화할 때는 세 가지 중요한 규칙이 있습니다. 첫째도 측정, 둘째도 측정, 셋째도 측정입니다. 따라서 자바 마이크로벤치마크 하니스(JMH)라는 라이브러리를 이용해서 작은 벤치마크를 구현해야 합니다.

JMH를 이용하면 간단하게 애노테이션 기반 방식을 지원하고, 안정적으로 자바 프로그램이나 자바 가상 머신(JVM)을 대상으로 하는 다른 언어용 벤치마크를 구현할 수 있습니다.

@BenchmarkMode(Mode.AverageTime) //벤치마크 대상 메서드를 실행하는데 걸린 평균 시간 측정
@OutputTimeUtil(TimeUnit.MILLISECONDS) //벤치마크 결과를 ms 단위로 출력
@Fork(2, jvmArgs = {"-Xms4G", "-Xmx4G"}) //4GB의 힙 공간을 제공한 환경에서 2번의 벤치마크를 수행
public class ParallelStreamBenchmark {
    private static final long N = 10_000_000L;
  
    @Benchmark
    public long sequentialSum() {
    	return Stream.iterate(1L, i -> i + 1).limit(N)
                                             .reduce(0L, Long::sum);
    }
    
    @TearDown(Level.Invocation) //매 벤치마크 실행한 후에 가비지 컬렉터 동작 시도
    public void tearDown() {
    	System.gc();
    }
}

벤치마크가 가능한 가비지 컬렉터의 영향을 받지 않도록 힙의 크기를 충분히 설정하며 끝날 때까지 가비지 컬렉터가 실행되도록 강제합니다. 이렇게 주의를 기울여도 결과를 정확하지 않을 수 있습니다. 왜냐하면, 기계가 지원하는 코어의 갯수 등이 실행 시간에 영향을 미칠 수 있기 때문입니다.

위 코드를 실행하면 JMH 명령은 핫스팟이 코드를 최적화할 수 있도록 20번을 실행하며 벤치 마크를 준비한 다음 20번을 더 실행해 최종 결과를 계산합니다. 즉, JMH는 기본적으로 20 + 20회 프로그램을 반복 실행합니다.

JMH의 특정 애노테이션이나 -w, -i 플래그를 명령행에 추가해서 이 기본 동작 횟수를 조절할 수 있습니다.

전통적인 for 루프를 사용해서 반복하는 방법이 더 저수준으로 동작할 뿐 아니라 특히 기본값을 박싱하거나 언박싱할 필요가 없으므로 더 빠를 거라고 예상할 수 있습니다.

다음 

7.1.3 병렬 스트림의 올바른 사용법

병렬 스트림을 잘못 사용하면서 발생하는 많은 문제는 공유된 상태를 바꾸는 알고리즘을 사용하기 때문에 일어납니다.

다음은 n까지의 자연수를 더하면서 공유된 누적자를 바꾸는 프로그램을 구현한 코드입니다.

public long sideEffectSum(long n) {
    Accumulator accumulator = new Accumulator();
    LongStream.rangeClosed(1, n).forEach(accumulator::add);
    return accumulator.total;
}

public class Accumulator {
    public long total = 0;
    public void add(long value) {
        total += value;
    }
}

본질적으로 순차 실행할 수 있도록 구현되어 있으므로 병렬로 실행하면 안됩니다. 특히, total을 접근할 때마다 (다수의 스레드에서 동시에 데이터에 접근하는) 데이터 레이스 문제가 발생합니다. 동기화로 문제를 해결하다보면 결국 병렬화라는 특성이 없어집니다. 스트림을 병렬로 만들어서 어떤 문제가 일어나는지 확인하면 다음과 같습니다.

public long sideEffectParallelSum(long n) {
    Accumulator accumulator = new Accumulator();
    LongStream.rangeClosed(1, n).parallel().forEach(accumulator::add);
    return accumulator.total;
}

메서드의 성능은 둘째 치고, 올바른 값이 안나옵니다. 여러 스레드에서 동시에 누적자를 실행하면서 문제가 발생합니다.

결국 여러 스레드에서 공유하는 객체의 상태를 바꾸는 forEach 블록 내부에서 add 메서드를 호출하면서 이 같은 문제가 발생합니다. 이 예제처럼 병렬 스트림을 사용했을 때 올바르게 동작하기 위해서는 공유된 가변 상태를 피해야 합니다.

7.1.4 병렬 스트림 효과적으로 사용하기

• 확신이 서지 않으면 순차 스트림과 병렬 스트림 구현 시 성능을 직접 측정하는 것이 좋습니다.
• 자동 박싱과 언박싱은 성능을 크게 저하시킬 수 있는 요소이므로 주의해서 사용해야 합니다. 기본형 특화 스트림(IntStream, LongStream, DoubleStream)을 사용하는 것이 좋습니다.
• limit이나 findFirst처럼 요소의 순서에 의존하는 연산은 병렬 스트림이 순차 스트림보다 성능이 더 떨어집니다. 예를 들어, findAny는 요소의 순서와 상관없이 연산하므로 findFirst보다 좋고, 스트림에 N개 요소가 있을 때 요소의 순서가 상관없다면 비정렬된 스트림에 limit을 호출하는 것이 더 좋습니다.
• 스트림에서 수행하는 전체 파이프라인 연산 비용을 고려해야합니다. 처리할 요소 수(N)가 많고 하나의 요소를 처리하는 비용을 Q라고 할 때 전체 스트림 파이프라인 처리 비용을 N*Q라고 볼 수 있습니다. Q가 높아진다는 것은 병렬 스트림으로 성능을 개선할 수 있는 가능성이 있다고 볼 수 있습니다.
• 병렬 스트림은 소량의 데이터를 처리하는 상황에서는 병렬화 과정에서 생기는 부가 비용을 상쇄할 만큼의 이득을 얻지 못합니다.
• 스트림을 구성하는 자료구조가 적절하지 확인해야 합니다. 예를 들어 LinkedList 를 분할하려면 모든 요소를 탐색해야 하지만 ArrayList는 모든 요소를 탐색하지 않고도 리스트를 분할할 수 있기 때문입니다. 또한, range 팩토리 메서드나 커스텀 Spliterator를 구현하면 쉽게 분해할 수 있습니다.
• 스트림의 특성과 파이프라인 중간 연산이 스트림의 특성을 어떻게 바꾸는지에 따라 분해 과정의 성능이 달라질 수 있습니다. 예를 들어, SIZED 스트림은 정확히 같은 크기의 두 스트림으로 분할할 수 있으므로 효과적으로 스트림을 병렬 처리할 수 있습니다. 반면, 필터 연산이 있으면 스트림의 길이를 예측할 수 없어 효과적으로 스트림을 병렬 처리할 수 있을 지 알 수 없게 됩니다.
• 최종 연산의 병합 과정 비용이 비싸다면 병렬 스트림으로 얻은 성능의 이익이 서브 스트림의 부분 결과를 합치는 과정에서 상쇄될 수 있습니다.

스트림 소스의 분해성

소스	분해성
ArrayList	훌륭함
LinkedList	나쁨
IntStream.range	훌륭함
Stream.iterate	나쁨
HashSet	좋음
TreeSet	좋음

7.2 포크/조인 프레임워크

포크/조인 프레임워크는 병렬화할 수 있는 작업을 재귀적으로 작은 작업으로 분할하여 서브 태스크로 처리한 뒤, 각각의 결과를 합쳐서 전체 결과로 만드는 방식입니다. 포크/조인 프레임워크에서는 서브 태스크를 스레드 풀의 작업자 스레드에 분산 할당하는 ExecutorService 인터페이스를 구현합니다.

7.2.1 RecursiveTask 활용

스레드 풀을 이용하려면 RecursiveTask<R>의 서브 클래스를 만들어야 합니다. R은 병렬화된 태스크가 생성하는 결과 형식 또는 결과가 없을 때(결과가 없더라도 다른 비지역 구조를 바꿀 수 있다)는 RecursiveAction 형식입니다.

RecursiveTask를 정의하려면 추상 메서드 compute를 구현해야 합니다.

protected abstract R compute();

compute 메서드는 태스크를 서브 태스크로 분할하는 로직과 더 이상 분할할 수 없을 때 개별 서브 태스크의 결과를 생산할 알고리즘을 정의합니다. 따라서 대부분 compute 메서드 구현은 다음과 같은 의사 코드 형식을 유지합니다.

if(태스크가 충분히 작거나 더이상 분할할 수 없으면) {
    순차적으로 태스크 계산
} else {
    태스크를두 서브태스크로 분할
    태스크가 다시 서브태스크로 분할되도록 메시지를 재귀적으로 호출
    모든 서브태스크의 연산이 왑료될때까지 대기
    각 서브태스크의 결과를 합침
}

ForkJoinSumCalculator 실행 

ForkJoinSumCalculator를 ForkJoinPool로 전달하면 풀의 스레드가 ForkJoinSumCalculator의 compute 메서드를 실행하면서 작업을 수행합니다. compute 메서드는 병렬로 실행할 수 있을만큼 태스크의 크기가 충분히 작아졌는지 확인하고, 아직 태스크의 크기가 크다고 판단되면 숫자 배열을 반으로 분할해 두 개의 새로운 ForkJoinSumCalculator로 할당합니다.

그러면 다시 ForkJoinPool이 새로 생성된 ForkJoinSumCalculator를 실행합니다.

결국 이 과정이 재귀적으로 반복하며 주어진 조건을 만족할 때까지 태스크 분할을 반복합니다.

7.2.2 포크/조인 프레임워크를 제대로 사용하는 방법

• join 메서드를 태스크에 호출하면 태스크가 생상하는 결과가 준비될 때까지 호출자를 블록시킵니다. 따라서, 두 서브 태스크가 모두 시작된 다음에 join을 호출하지 않으면, 각각의 서브 태스크가 다른 태스크가 끝나는걸 기다리는 일이 발생하며 원래 순차 알고리즘보다 느리고 복잡한 프로그램이 될 수 있습니다.
• RecursiveTask 내에서는 ForkJoinPool의 invoke 메서드를 사용하지 않고, compute나 fork 메서드를 사용하며, 순차 코드에서 병렬 계산을 시작할 때만 invoke를 사용합니다.
• 서브 태스크에 fork 메서드를 호출해서 ForkJoinPool의 일정을 조절할 수 있습니다. 한쪽 작업에만 fork를 호출하는 것보다는 compute를 호출하는 것이 효율적입니다. 그러면 두 서브 태스크의 한 태스크에는 같은 스레드를 재사용할 수 있으므로 풀에서 불필요한 태스크를 할당하는 오버헤드를 피할 수 있습니다.
• 포크/조인 프레임워크의 병렬 계산은 디버깅하기 어렵습니다. fork라 불리는 스레드에서 compute를 호출하므로 스택 트레이스가 도움이 되지 않습니다.
• 병렬 처리로 성능을 개선하려면 태스크를 여러 독립적인 서브 태스크로 분할할 수 있어야 하며, 각 서브 태스크의 실행 시간은 새로운 태스크를 포킹하는 데 드는 시간보다 길어야 합니다.

7.2.3 작업 훔치기

이론적으로 CPU의 코어 개수만큼 병렬화된 태스크로 작업부하를 분할하면 모든 코어에서 태스크를 실행할 것이고, 같은 시간에 종료될 것이라고 생각할 수 있습니다. 하지만 다양한 이유로 각각의 서브 태스크의 작업 완료 시간이 크게 달라질 수 있습니다. 포크/조인 프레임워크에서는 작업 홈치기라는 기법으로 이 문제를 해결할 수 있습니다.

작업 홈치기 기법에서는 ForkJoinPool의 모든 스레드를 공정하게 분할합니다. 각각의 스레드는 자신에게 할당된 태스크를 포함하는 이중 연결 리스트를 참조하면서 작업이 끝날 때마다 큐의 헤드에서 다른 태스크를 가져와서 작업을 처리합니다. 이때 한 스레드는 다른 스레드보다 자신에게 할당된 태스크를 더 빨리 처리할 수 있습니다.

즉, 다른 스레드는 바쁘게 일하지만, 할 일이 없어진 한 스레드는 유휴 상태로 바뀌는 것이 아니라 다른 스레드 큐의 꼬리에서 작업을 홈쳐옵니다. 모든 태스크가 작업을 끝낼 때까지, 즉 모든 큐가 빌 때가지 이 과정을 반복합니다. 따라서 태스크의 크기를 작게 나누어야 작업자 스레드 간의 작업 부하를 비슷한 수준으로 유지할 수 있습니다.

7.3 Spliterator 인터페이스

Spliterator는 '분할할 수 있는 반복자'라는 의미입니다. Iterator처럼 Spliterator는 소스의 요소 탐색 기능을 제공하며, 병렬 작업에 특화되어 있습니다.

커스텀 Spliterator를 꼭 직접 구현할 필요는 없지만 어떻게 동작하는지 이해하면 병렬 스트림의 동작에 대해 이해하기 도움이 됩니다. 자바 8은 컬렉션 프레임워크에 포함된 모든 자료구조에 사용할 수 있는 디폴트 Spliterator 구현을 제공합니다. 컬렉션 spliterator라는 메서드를 제공하는 Spliterator 인터페이스를 구현합니다.

public interface Spliterator<T> {
    boolean tryAdvance(Consumer<? super T> action);
    Spliterator<T> trySplit();
    long estimateSize();
    int characteristics();
}

여기서 T는 Spliterator에서 탐색하는 요소의 형식입니다.

tryAdvance 메서드는 Spliterator의 요소를 하나씩 순차적으로 소비하며 탐색해야 할 요소가 있으면 참을 반환합니다.

trySplit 메서드는 Spliterator의 일부 요소를 분할해서 두 번째 Spliterator를 생성하는 메서드입니다.

estimateSize 메서드는 탐색해야 할 요소 수 정보를 제공할 수 있고, 특히 탐색해야 할 요소 수가 정확하지 않더라고 제공된 값을 이용해 더 공평하게 Spliterator를 분할할 수 있습니다.

characteristics는 Spliterator의 특성을 정의할 수 있습니다.

7.3.1 분할 과정

다음 그림처럼 스트림을 여러 스트림으로 분할하는 과정은 재귀적으로 일어납니다.

1단계에서 첫 번째 Spliterator에 trySplit을 호출하면 두 번째 Spliterator가 생성됩니다.

2단계에서 두 개의 Spliterator에 trySplit을 다시 호출하면 네 개의 Spliterator가 생성되고 이처럼 trySplit의 결과가 null이 될 때까지 이 과정을 반복합니다.

trySplit이 null을 반환하는 것은 더 이상 자료구조를 분할할 수 없음을 의미하고, 4단계에서 Spliterator에 호출한 모든 trySplit의 결과가 null이면 재귀 분할 과정이 종료됩니다. 

이 분할 과정은 characteristics 메서드로 정의하는 Spliterator의 특성에 영향을 받습니다.

Spliterator 특성

• ORDERED :  
• DISTINCT
• SORTED
• SIZED
• NON-NULL
• IMMUTABLE
• CONCURRENT
• SUBSIZED

7.3.2 커스텀 Spliterator 구현하기