7장: 병렬 데이터 처리와 성능

스트림의 꽃, 병렬 처리 과정을 알아보자.

자바 7 이전에는 데이터를 서브 파트로 분할해 각각의 스레드로 할당하고 적절한 동기화 처리를 하고, 마지막으로 결과를 모두 합쳐야 하는 것이 개발자의 몫이었다.
자바 7은 포크/조인 프레임워크 기능을 통해 더 쉽게 병렬화를 수행하며 에러를 최소화할 수 있도록 한다.

병렬 스트림

각각의 스레드에서 처리할 수 있도록 스트림 요소를 여러 청크로 분할한 스트림이다.
컬렉션에 parallelStream을 호출하면 병렬 스트림이 생성된다.

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

순차 스트림에 parallel 메서드를 호출하면 기존의 함수형 리듀싱 연산이 병렬로 처리된다. 즉, 리듀싱 연산을 여러 청크로 나누어 병렬 수행이 가능하다.
```
public long sequentialSum(long n) {
  return Stream.iterate(1L, i -> i + 1)
    .limit(n)
    .parallel() //병렬 스트림으로 변환
    .reduce(0L, Long::sum);
  }
```
parallel() 메서드와 sequential() 메서드를 사용해 병렬로 실행할 연산과 순차로 실행할 연산을 제어할 수 있다.
두 메서드 중 최종적으로 호출된 메서드가 전체 파이프라인에 영향을 미친다.
아래의 경우 parallel() 메서드가 마지막에 호출되었으므로 파이프라인은 전체적으로 병렬로 실행된다.

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

스레드 풀 설정
스트림의 parallel 메서드에서 병렬로 작업을 수행하는 스레드는 ForkJoinPool에 의해 관리된다.
보통 프로세서 수에 따라 스레드 개수를 자동으로 설정하기 때문에 특별한 이유가 없으면 기본 값 사용을 권장한다.

스트림 성능 측정

병렬화를 한다고 반드시 성능이 좋아지는 것이 아니므로 직접 측정해보아야 한다.
JMH를 사용해 작은 벤치마크를 구현할 수 있다.
- JVM으로 실행되는 프로그램의 벤치마크 과정은 핫스팟이 바이트 코드 최적화를 위해 필요한 웜업 시간, GC로 인한 오버헤드 등으로 영향을 받기 때문에 어려운 편이다.

병렬 처리 시 조심해야 할 점

iterate()와 같이 순차적인 연산은 이전 연산의 결과에 따라 다음 함수의 입력이 달라지기 때문에 청크로 분할할 수 없다.
따라서 병렬 처리 시에는 IntStream.rangeClosed() 와 같이 청크로 분할이 용이한 데이터 소스를 사용해야 한다.
병렬화 이용 시 스트림을 재귀적으로 분할, 각 서브스트림을 서로 다른 스레드에 할당, 결과를 하나의 값으로 합치는 오버헤드가 발생한다.
따라서 코어 간 데이터 전송 시간보다 훨씬 오래걸리는 작업만 병렬화하는것이 바람직하다.
공유된 상태(누적자)를 바꾸는 알고리즘에서 제대로 동기화를 해주지 않는다면 문제가 발생하게 된다.

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

장비의 사양과 연산의 특성 등 상황에 따라 천차만별이기 때문에 어떤 상황에서 병렬 스트림을 사용할 지에 대한 정확한 가이드는 없다.
확신이 서지 않을때는 순차 스트림과 병렬 스트림 구현 시의 성능을 직접 측정한다.
오토박싱과 언박싱은 성능을 크게 저하시킬 수 있는 요소이므로 주의해야 하며, 기본형 특화 스트림(IntStream, LongStream, DoubleStream)을 사용하는 것이 좋다.
limit이나 findFirst처럼 요소의 순서에 의존하는 연산은 병렬 스트림에서 성능이 더 떨어진다. 스트림 요소의 순서가 상관없다면 unordered를 호출해서 비정렬된 스트림을 얻은 후 limit을 호출하는 것이 더 효율적이다.
스트림에서 수행하는 전체 파이프라인 연산 비용을 고려해 처리해야 할 요소 수가 많고 한 요소 당 처리 비용이 높다면 병렬 스트림으로 성능을 개선할 여지가 있다.
소량의 데이터에서는 병렬화 과정의 부가 비용을 상쇄하지 못하므로 병렬스트림을 쓰지 말자.
스트림을 구성하는 자료구조가 적절한지 확인한다.
- ArrayList는 요소를 탐색하지 않고도 분할할 수 있지만 LinkedList는 모든 요소를 탐색해야 분할할 수 있다.
- range 팩토리 메서드로 만든 기본형 스트림이나 커스텀 Spliterator를 구현하면 쉽게 분해할 수 있다.
스트림의 특성과 파이프라인 중간 연산이 스트림의 특성을 어떻게 바꾸는지에 따라 분해 과정의 성능이 달라질 수 있다.
- SIZED 스트림은 정확히 같은 크기의 두 스트림으로 분할할 수 있으므로 효과적으로 병렬 처리가 가능하다.
- 필터 연산이 있으면 스트림의 길이를 예측할 수 없어 분할이 어려우므로 병렬 처리가 어려워진다.
최종 연산의 병합 과정 비용이 비싸다면 병렬 스트림으로 얻은 이익이 상쇄될 수 있다.

포크/조인 프레임워크

병렬화할 수 있는 작업을 재귀적으로 작은 작업으로 분할하여 서브태스크로 각 스레드에 분산 할당하여 처리한 뒤, 각각의 결과를 합쳐서 전체 결과로 만든다.

RecursiveTask 활용

스레드풀을 이용하려면 RecursiveTask<R>의 서브클래스를 만들어 추상메서드 compute()를 구현해야 한다.

protected abstract R compute();

compute()

태스크를 서브 태스크로 분할하는 로직과, 더이상 분할 불가일 때 서브 태스크의 결과를 생산할 알고리즘을 정의한다.

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

병렬로 합계 구하기

RecursiveTask를 구현하여 포크/조인 프레임워크를 통해 범위의 숫자를 더할 수 있다.
numbers라는 배열을 입력받아 해당 배열의 합계를 구해야 한다.
한 태스크의 요소가 10000개 이하가 될 때까지 태스크를 분할하고, 이후 각 태스크의 작업 결과를 합친다.

public class ForkJoinSumCalculator extends RecursiveTask<Long> {
    private final long[] numbers;
    private final int start; // 배열 초기 위치
    private final int end; // 배열 최종 위치
    private static final long THRESHOLD = 10_000; // 서브태스크 분할 최소 기준값

    // 해당 클래스를 생성할때 사용, 숫자 배열을 입력받는다.
    public ForkJoinSumCalculator(long[] numbers) {
        this(numbers, 0, numbers.length);
    }

    // 재귀 호출을 위한 비공개 생성자
    private ForkJoinSumCalculator(long[] numbers, int start, int end) {
        this.numbers = numbers;
        this.start = start;
        this.end = end;
    }

    @Override
    protected Long compute() {
        // 이 태스크에서 처리할 길이
        int length = end - start;

        // 최소 기준값 이하이면 순차적으로 결과 계산
        if (length <= THRESHOLD) return computeSequentially();

        // 배열 길이의 반
        int halfLength = start + length / 2;
        
        // 서브 태스크 생성
        ForkJoinSumCalculator leftTask = 
            new ForkJoinSumCalculator(numbers, start, halfLength);
        leftTask.fork(); // 새로 생성된 태스크는 다른 스레드에서 비동기로 실행

        // 두번째 서브 태스크 생성
        ForkJoinSumCalculator rightTask = 
            new ForkJoinSumCalculator(numbers, halfLength, end);
        Long rightResult = rightTask.compute(); // 동기로 실행
        Long leftResult = leftTask.join(); // leftTask의 결과를 읽거나, 아직 결과가 없으면 기다림

        return leftResult + rightResult; // 태스트 결과 합치기
    }

    // 더 분할할 수 없을 때 서브태스크의 결과 계산
    private long computeSequentially() {
        long sum = 0;
        for (int i = start; i < end; i++) {
            sum += numbers[i];
        }
        return sum;
    }
}

아래와 같이 ForkJoinTask를 ForkJoinPool의 invoke 메서드로 전달하면, compute 메서드를 실행하여 태스크 결과를 얻을 수 있다.

long[] numbers = LongStream.rangeClosed(1, 1000000L).toArray();
ForkJoinSumCalculator task = new ForkJoinSumCalculator(numbers);
Long result = new ForkJoinPool().invoke(task);

일반적으로 애플리케이션에서는 둘 이상의 ForkJoinPool을 사용하지 않는다.
즉, 소프트웨어의 필요한 곳에서 언제든 가져다 쓸 수 있도록 ForkJoinPool을 한 번만 인스턴스화해서 정적 필드에 싱글턴으로 저장한다.
ForkJoinPool의 인스턴스 생성 시 인자를 주지 않으면 JVM에서 이용할 수 있는 모든 프로세서가 자유롭게 풀에 접근할 수 있다는 의미이다.
ForkJoinPool은 기본적으로 Runtime.availableProcessors의 반환값으로 풀에 사용할 스레드 수를 결정하는데, 이 반환값에는 하이퍼스레딩과 관련된 가상 프로세서도 개수에 포함이 된다.

포크/조인 프레임워크 제대로 사용하기

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

작업 훔치기

이론적으로는 CPU의 코어 개수만큼 병렬화된 태스크로 작업부하를 분할하면 모든 코어에서 태스크를 실행할 것이고, 같은 시간에 종료될 것이라고 생각할 수 있다.
하지만 디스크 접근 속도 저하, 외부 서비스 호출 지연 등의 이유로 각각의 서브태스크의 작업완료 시간이 크게 달라질 수 있다.
작업 훔치기
- ForkJoinPool의 모든 스레드를 공정하게 분할하기 위해, 어떤 스레드가 자신에게 할당된 작업이 모두 끝났다면 다른 스레드 작업 큐의 tail에서 작업을 가져와 처리한다.
실제 코어 수 보다 더 잘게 나누면, 스레드 간의 작업부하를 비슷한 수준으로 유지할 수 있다.
결론은, 코어 개수와 관계없이 적절한 크기로 분활된 많은 태스크를 포킹하는 것이 바람직하다.

Spliterator 인터페이스

자동으로 스트림을 분할하는 기법
Iterator처럼 소스의 요소 탐색 기능을 제공하며, 병렬 작업에 특화되어 있다.
Spliterator 동작 방법을 잘 이해한다면 병렬 스트림 동작에 대한 이해를 높일 수 있을 것!

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

tryAdvance: Spliterator의 요소를 순차적으로 소비하면서 탐색해야할 요소가 있으면 참을 반환
trySplit: Spliterator의 일부 요소(자신이 반환한 요소)를 분할해서 두 번째 Spliterator를 생성
estimateSize: 탐색해야할 요소 수
characteristics: Spliterator의 특성을 정의

분할 과정

초기 Spliterator에 trySplit을 호출하여 하위 Spliterator를 생성한다. 이 과정을 아래 그림과 같이 반복하고, trySplit 메서드에서 null이 나오면 재귀 분할 과정이 종료된다.

Spliterator 특성

characteristics 메서드는 Spliterator 자체의 특성 집합을 포함하는 int를 반환한다.

특성	의미
ORDERED	리스트처럼 정해진 순서가 있으므로 요소를 탐색하고 분할할 때 순서에 유의해야 함
DISTINCT	x, y 두 요소를 방문했을 때 x.equals(y)는 항상 false를 반환
SORTED	탐색된 요소는 미리 정의된 정렬 순서를 따른다.
SIZED	크기가 알려진 소스로 생성했으므로 estimatedSize()는 정확한 값을 반환한다.
NON-NULL	탐색하는 모든 요소는 null이 아니다.
IMMUTABLE	이 Spliterator의 소스는 불변이므로 요소를 탐색하는 동안 추가/삭제/수정할 수 없다.
CONCURRENT	동기화 없이 Spliterator의 소스를 여러 스레드에서 동시에 고칠 수 있다.
SUBSIZED	이 Spliterator와 분할되어 생성되는 모든 spliterator는 SIZED 특성을 갖는다.

특성

의미

ORDERED

리스트처럼 정해진 순서가 있으므로 요소를 탐색하고 분할할 때 순서에 유의해야 함

DISTINCT

x, y 두 요소를 방문했을 때 x.equals(y)는 항상 false를 반환

SORTED

탐색된 요소는 미리 정의된 정렬 순서를 따른다.

SIZED

크기가 알려진 소스로 생성했으므로 estimatedSize()는 정확한 값을 반환한다.

NON-NULL

탐색하는 모든 요소는 null이 아니다.

IMMUTABLE

이 Spliterator의 소스는 불변이므로 요소를 탐색하는 동안 추가/삭제/수정할 수 없다.

CONCURRENT

동기화 없이 Spliterator의 소스를 여러 스레드에서 동시에 고칠 수 있다.

SUBSIZED

이 Spliterator와 분할되어 생성되는 모든 spliterator는 SIZED 특성을 갖는다.

커스텀 Spliterator 구현

문자열의 단어 수 계산

문자열에 해당하는 단어 개수를 반환하는 예제를 통해 커스텀 Spliterator의 구현을 알아보자.
커스텀 Spliterator의 필요성
- 병렬로 문자열의 단어 수를 계산하려 할 때 원본 문자열을 임의의 위치에서 둘로 나누면 하나의 단어를 둘로 계산하는 상황이 발생할 수 있다.
- 따라서 문자열을 임의의 위치에서 분할하지 않고 단어가 끝나는 위치에서만 분할하도록 커스텀 Spliterator 클래스를 만들어 trySplit() 메서드를 구현해주어야 한다.
먼저 문자열의 단어 수를 계산하기 위해 변수 상태를 캡슐화하는 WordCounter 클래스를 만들자.
- 스트림을 탐색하며 새로운 문자를 찾을때마다 accumulate 메서드를 호출한다.
- accumulate 메서드는 새로운 WordCounter 클래스를 어떤 상태로 생성할 것인지 정의한다.
  - 공백이 아닌 문자이고 직전 글자는 공백이었을 경우에만 counter(단어의 개수)를 추가한다.
- combine 메서드는 서브스트림으로 처리한 결과들을 어떻게 합칠 지 정의한다.

class WordCounter {
  private final int counter;
  private final boolean lastSpace;
  
  public WordCounter(int counter, boolean lastSpace) {
    this.counter = counter;
    this.lastSpace = lastSpace;
  }
  
  public WordCounter accumulate(Character c) {
    if (Character.isWhitespace(c)) {
      return lastSpace ? this : new WordCounter(counter, true);
    } else {
      return lastSpace ? new WordCounter(counter+1, false) : this;
    }
  }
  
  public WordCounter combine(WordCounter wordCounter) {
    return new WordCounter(counter + wordCounter.conter, wordCounter.lastSpace);
  }
  
  public int getConter() {
    return conter;
  }
}

병렬 스트림을 사용하기에 앞서 간단하게 리듀싱 연산을 통해 문자열의 단어 개수를 찾아보자.

public int countWords(Stream<Character> stream) {
    WordCounter wordCounter = stream
        .reduce(new WordCounter(0 ,true),
                WordCounter::accumulate,
                WordCounter::combine);
    return wordCounter.getCounter();
}

final String sentence = "words   in this  sentence should  be   7";
Stream<Character> stream = IntStream.range(0, sentence.length()).mapToObj(sentence::charAt);
System.out.println("Found " + countWords(stream) + " words"); // 7이 출력되어야 한다.

이제 앞서 말했던 병렬 처리의 문제를 해결하기 위해 Spliterator 구현 클래스를 만들자.
- tryAdvance: 탐색 할 요소가 남아있다면 true 반환
- trySplit: 요소를 분할해서 Spliterator 생성해 반환, 공백이 아닐 때에는 인덱스를 넘어가 해당 단어가 끝난 지점으로 이동한다.
- estimateSize: 탐색해야 할 요소의 수를 반환
- characteristics: Spliterator 객체에 포함된 모든 특성값의 합을 반환
  - ORDERED : 문자열의 순서가 유의미함
  - SIZED : estimatedSize 메서드의 반환값이 정확함
  - NONNULL : 문자열에는 null이 존재하지 않음
  - IMMUTABLE : 문자열 자체가 불변 클래스이므로 파싱하며 속성이 추가되지 않음

public class WordCounterSpliterator implements Spliterator<Character> {
    private final String string;
    private int currentChar = 0; // 분할 시작 위치를 문자열의 인덱스로 나타냄

    public WordCounter(String string) {
        this.string = string;
    }

    @Override
    public boolean tryAdvance(Consumer<? super Character> action) {
        action.accept(string.charAt(currentChar++)); // 현재 문자를 소비
        return currentChar < string.length(); // 소비할 문자가 남아있으면 true 반환
    }

    @Override
    public Spliterator<Character> trySplit() {
        int currentSize = string.length() - currentChar;
        if (currentSize < 10) return null; // 크기가 작으면 더이상 분할하지 않는다.
       
        for (int splitPos = currentSize / 2 + currentChar;  // 파싱할 문자열의 중간을 분할 위치로 설정
                splitPos < string.length(); splitPos++) { // 공백 문자가 아니라면 위치 이동
            if (Character.isWhitespace(string.charAt(splitPos))) {
                // 문자열을 분할해 Spliterator 생성
                Spliterator<Character> spliterator = 
                    new WordCounter(string.substring(currentChar, splitPos));

                // 분할 시작 위치 설정
                currentChar = splitPos;
                return spliterator;
            }
        }
        return null;
    }

    @Override
    public long estimateSize() {
        return string.length() - currentChar;
    }

    @Override
    public int characteristics() {
        return ORDERED + SIZED + SUBSIZED + NONNULL + IMMUTABLE;
    }
}

드디어 Spliterator를 사용해 병렬 스트림을 생성한 뒤 단어 수를 계산할 수 있다.

Spliterator<Character> spliterator = new WordCounterSpliterator(sentence);
Stream<Character> stream = StreamSupport.stream(spliterator, true); // StreamSupport로 병렬 스트림 생성
System.out.println("Found " + countWords(stream) + " words"); // 7이 출력되어야 한다.

Previous6장: 스트림으로 데이터 수집 Next8장: 컬렉션 API 개선

Last updated 8 months ago