🐹 chap7. 병렬 데이터 처리와 성능

7.1 병렬 스트림

: 각각의 스레드에서 처리할 수 있도록 스트림 요소를 여러 청크로 분할한 스트림이다. 모든 멀티코어 프로세서가 각각의 청크를 처리하도록 할당할 수 있다.

컬렉션에 parallelStream을 호출하면 병렬 스트림이 생성된다.

ex) 숫자 n을 인수로 받아 1부터 n까지의 모든 숫자의 합계를 반환하는 메서드 구현

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

전통적인 자바에서는 다음과 같이 반복문으로 이를 구현할 수 있다.

public long iterativeSum(long n) {
    long result = 0;
    for (long i = 1L; i <= n; i++) {
        result += 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 두 메서드 중 최종적으로 호출된 메서드가 전체 파이프라인에 영향을 미친다.

위 코드에서 파이프라인의 마지막 호출은 parallel이므로 파이프라인은 전체적으로 병렬로 실행된다.

7.1.2 스트림 성능 측정

병렬화를 이용하면 순차나 반복 형식에 비해 성능이 좋아질 것이라 추측하지만, 실제 측정을 해봐야한다.

따라서 자바 마이크로벤치마크 하니스(JMH)라는 라이브러리를 이용해 작은 벤치마크를 구현한다. JMH를 이용하면 간단하고, 어노테이션 기반 방식을 지원하며, 안정적으로 자바 프로그램이나 자바 가상 머신(JVM)을 대상으로 하는 다른 언어용 벤치마크를 구현할 수 있다.

메이븐(Maven) 빌드 도구를 사용한다면 빌드 과정을 정의하는 pom.xml 파일에 몇 각지 의존성을 추가해 프로젝트에서 JMH를 사용할 수 있다.

<dependency>
  <groupId>org.openjdk.jmh</groupId>
  <artifactId>jml-core</artifactId>
  <version>1.17.4</version>
</dependency>
<dependency>
  <groupId>org.openjdk.jmh</groupId>
  <artifactId>jmh-generator-annprocess</artifactId>
  <version>1.17.4</version>
</dependency>

첫 번째 라이브러리는 핵심 JMH 구현을 포함하고, 두 번째 라이브러리는 자바 아카이브(Java Archive, JAR) 파일을 만드는 데 도움을 주는 어노테이션 프로세서를 포함한다.

메이븐 설정에 다음 플러그인도 추가한 다음 자바 아카이브 파일을 이용해서 벤치마크를 편리하게 실행할 수 있다.

<build>
  <plugins>
	<plugin>
	  <groupId>org.apache.maven.plugins</groupId>
	  <artifactId>maven-shade-plugin</artifactId>
	  <executions>
		<execution>
		  <phase>package</phase>
		  <goals><goal>shade</goal></goals>
		  <configuration>
			<finalName>benchmarks</finalName>
			<transformers>
			  <transformer implementation=
				"org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
				<mainClass>org.openjdk.jmh.Main</mainClass>
			  </transformer>
			</transformers>
		  </configuration>
		</execution>
	  </executions>
	</plugin>
  </plugins>
</build>

아래 코드처럼 이전의 sequentialSum 메서드를 간단하게 벤치마크할 수 있다.

• n개의 숫자를 더하는 함수의 성능 측정

@BenchmarkMode(Mode.AverageTime) // 벤치마크 대상 메서드를 실행하는 데 걸린 평균 시간 측정
@OutputTimeUnit(TimeUnit.MILLISECONS) // 벤치마크 결과를 밀리초 단위로 출력
@Fork(2, jvmArgs={"-Xms4G", "-Xmx4G"}) // 4Gb의 힙 공간을 제공한 환경에서 두 번 벤치마크를 수행해 결과의 신뢰성 확보
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();
    }
}

클래스를 컴파일하면 이전에 설정한 메이븐 플러그인이 benchmarks.jar라는 두 번째 파일을 만든다. 다음처럼 실행할 수 있다.

java -jar ./target/benchmarks.jar ParallelStreamBenchmark 

→ 벤치마크가 가능한 가비지 컬렉터의 영향을 받지 않도록 힙의 크기를 충분하게 설정

→ 벤치마크가 끝날 때마다 가비지 컬렉터가 실행되도록 강제

이렇게 해도 결과는 정확하지 않을 수 있다. (기계가 지원하는 코어의 개수 등이 실행 시간에 영향을 미칠 수 있기 때문이다.

• 위 코드 프로그램 실행 횟수
  • 벤치마크 준비: JMH 명령은 핫스팟이 코드를 최적화 할 수 있도록 20번 실행
  • 최종 결과 계산: 20번 더 실행

  JMH는 기본적으로 20+20회 프로그램을 반복 실행

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

벤치마크 결과

다음은 인텔 i7-4600U 2.1GHz 쿼드 코어에서 실행한 결과다.

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

@Benchmark
public long iterativeSum() {
    long result = 0;
    for (long i = 1L; i <= N; i++) {
        result += i;
    }
    return result;
}

같은 컴퓨터로 두 번째 벤치마크 코드를 돌린 결과이다.

예상대로 순차적으로 스트림을 사용하는 버전에 비해 거의 4배가 빠르다는 것을 확인할 수 있다.

이번에는 벤치마크 클래스에 병렬 스트림 메서드를 사용하는 버전의 결과다.

병렬 버전이 쿼드 코어 CPU를 사용하지 못하고 순차 버전에 비해 다섯 배나 느린 결과가 나왔다. 이 결과에 대해 두 가지 문제를 발견할 수 있다.

• 반복 결과로 박싱된 객체가 만들어지므로 숫자를 더하려면 언박싱을 해야 한다.
• 반복 작업은 병렬로 수행할 수 있는 독립 단위로 나누기가 어렵다.

아래 그림처럼 이전 연산의 결과에 따라 다음 함수의 입력이 달라지기 때문에 iterate 연산을 청크로 분할하기가 어렵다.

리듀싱 과정을 시작하는 시점에 전체 숫자 리스트가 준비되지 않았으므로 병렬로 처리할 수 있도록 청크로 분할할 수 없다. 스트림이 병렬로 처리되도록 지시했고 각각의 합계가 다른 스레드에서 수행되었지만 결국 순차처리 방식과 크게 다른 점이 없으므로 스레드를 할당하는 오버헤드만 증가하게 된다.

병렬 프로그래밍을 오용하면 오히려 전체 프로그램의 성능이 더 나빠질 수도 있다. 따라서 parallel 메서드를 호출했을 때 내부적으로 어떤 일이 일어나는지 꼭 이해해야 한다.

더 특화된 메서드 사용

LongStream.rangeClosed라는 메서드는 iterate에 비해 다음과 같은 장점을 제공한다.

• LongStream.rangeClosed는 기본형 long을 직접 사용하므로 박싱과 언박싱 오버헤드가 사라진다.
• LongStream.rangeClosed는 쉽게 청크로 분할할 수 있는 숫자 범위를 생산한다. 예를 들어 1-20 범위의 숫자를 각각 1-5, 6-10, 11-15, 16-20 범위의 숫자로 분할할 수 있다.

언박싱과 관련한 오버헤드는 얼마나 될까? 우선 순차 스트림을 처리하는 시간을 측정한다.

@Benchmark
public long rangedSum() {
    return LongStream.rangeClosed(1, N)
                     .reduce(0L, Long::sum);
}

기존의 iterate 팩토리 메서드로 생성한 순차 버전에 비해 이 예제의 숫자 스트림 처리 속도가 더 빠르다. 특화되지 않은 스트림을 처리할 때는 오토박싱, 언박싱 등의 오버헤드를 수반하기 때문이다.

상황에 따라서는 어떤 알고리즘을 병렬화하는 것보다 적절한 자료구조를 선택하는 것이 더 중요하다는 사실을 단적으로 보여준다. 이번엔 병렬 스트림은 어떨까?

@Benchmark
public long parallelRangedSum() {
    return LongStream.rangedClosed(1, N)
                     .parallel()
                     .reduce(0L, Long::sum);
}

이번엔 실질적으로 리듀싱 연산이 병렬로 수행된다. 올바른 자료구조를 선택해야 병렬 실행도 최적의 기능을 발휘할 수 있다는 것을 확인했다.

함수형 프로그래밍을 올바로 사용하면 반복적으로 코드를 실행하는 방법에 비해 최신 멀티코어 CPU가 제공하는 병렬 실행의 힘을 단순하게 직접적으로 얻을 수 있다.

But 병렬화가 완전 공짜는 아니다. 병렬화를 이용하려면

1. 스트림을 재귀적으로 분할해야 하고
2. 각 서브스트림을 서로 다른 스레드의 리듀싱 연산으로 할당하고
3. 이들 결과를 하나의 값으로 합쳐야 한다.

멀티코어 간의 데이터 이동은 생각보다 비싸다. 따라서 코어 간에 데이터 전송 시간보다 훨씬 오래 걸리는 작업만 병렬로 다른 코어에서 수행하는 것이 바람직하다.

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

System.out.println("SideEffect parallel sum done in: " +
    measurePerf(ParallelStreams::sideEffectParallelSum, 10_000_000L) + " msecs" );

결과는 다음과 같다.

메서드의 성능은 둘째 치고, 올바른 결과값(50000005000000)이 나오지 않는다. 여러 스레드에서 동시에 누적자, 즉 total += value를 실행하면서 이런 문제가 발생한다.

여러 스레드에서 공유하는 객체의 상태를 바꾸는 forEach 블록 내부에서 add 메서드를 호출하면서 이 같은 문제가 발생한다. 상태 공유에 따른 부작용을 피해야 한다.

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

• 확신이 서지 않으면 직접 측정하라. 순차 스트림을 병렬 스트림으로 쉽게 바꿀 수 있다. 적절한 벤치마크로 직접 성능을 측정하는 것이 바람직하다.
• 박싱을 주의하라. 자동 박싱과 언박싱은 성능을 크게 저하시킬 수 있는 요소다. 되도록이면 기본형 특화 스트림(IntStream, LongStream, DoubleStream)을 사용하는 것이 좋다.

• 순차 스트림보다 병렬 스트림에서 성능이 떨어지는 연산이 있다. 특히 limit나 findFirst처럼 요소의 순서에 의존하는 연산은 병렬 스트림 수행 시 비싼 비용을 치러야 한다.

  예를 들어 firstAny는 요소의 순서와 상관없이 연산하므로 findFirst보다 성능이 좋다. 정렬된 스트림에 unordered를 호출하면 비 정렬된 스트림을 얻을 수 있다. 스트림에 N개의 요소가 있을 때 요소의 순서가 상관없다면 비정렬된 스트림에 limit을 호출하는 것이 더 효율적이다.

• 스트림에서 수행하는 전체 파이프라인 연산 비용을 고려하라.

  처리해야 할 요소 수가 N이고 하나의 요소를 처리하는 데 드는 비용이 Q라하면 N*Q로 예상할 수 있다. Q가 높다는 것은 병렬 스트림으로 성능을 개선할 수 있는 가능성이 있음을 의미한다.

• 소량의 데이터에서는 병렬 스트림이 도움 되지 않는다.

  병렬화 과정에서 생기는 부가 비용을 상쇄할 만큼의 이득을 얻지 못하기 때문이다.

• 스트림을 구성하는 자료구조가 적절한지 확인해라

  예를 들어 ArrayList를 LinkedList보다 효율적으로 분할할 수 있다. LinkedList를 분할하려면 모든 요소를 탐색해야 하지만 ArrayList는 요소를 탐색하지 않고도 리스트를 분할할 수 있기 때문이다.

  또한 range 팩토리 메서드로 만든 기본형 스트림도 쉽게 분해할 수 있다.

• 스트림의 특성과 파이프라인의 중간 연산이 스트림의 특성을 어떻게 바꾸는지에 따라 분해 과정의 성능이 달라질 수 있다.

  예를 들어 SIZED 스트림은 정확히 같은 크기의 두 스트림으로 분할할 수 있으므로 효과적으로 스트림을 병렬처리 할 수 있다. 하지만 필터 연산이 있으면 스트림의 길이를 예측할 수 없으므로 효과적인 스트림을 병렬 처리할 수 있을지 알 수 없게 된다.

• 최종 연산의 병합 과정 비용을 살펴보라.

  병합 과정의 비용이 비싸다면 병렬 스트림으로 얻은 성능의 이익이 서브 스트림의 부분 결과를 합치는 과정에서 상쇄될 수 있다.

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

7.2 포크/조인 프레임워크

→ 병렬화할 수 있는 작업을 재귀적으로 작은 작업으로 분할한 다음에 서브태스크 각각의 결과를 합쳐서 전체 결과를 만들도록 설계되었다.

포크/조인 프레임워크에서는 서브태스크를 스레드 풀(ForkJoinPool)의 작업자 스레드에 분산 할당하는 ExecutorService 인터페이스를 구현한다.

7.2.1 Recursive Task 활용

스레드 풀을 이용하려면 **RecursiveTask**의 서브클래스를 만들어야 한다.

R : 은 병렬화된 태스크가 생성하는 결과 형식 또는 결과가 없을 때 RecursiveAction 형식

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

protected abstract R compute();

• compute 메서드
  • 태스크를 서브태스크로 분할하는 로직 정의
  • 더 이상 분할할 수 없을 때 개별 서브태스크의 결과를 생산할 알고리즘 정의

따라서 대부분의 compute 메서드 구현은 다음과 같은 의사코드 형식을 유지한다.

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

이 알고리즘은 분할 후 정복(divide-and-conquer) 알고리즘의 병렬화 과정이다. 다음 예제의 ForkJoinSumCalculator 코드에서 보여주는 것처럼 먼저 RecursiveTask를 구현해야 한다.

// RecursiveTask를 상속받아 포크/조인 프레임워크에서 사용할 태스크 생성
public class ForkJoinSumCalculator 
extends java.util.concurrent.RecursiveTast<Long> { 
    private final long[] numbers; // 더할 숫자 배열
    private final int start; // 이 서브태스크에서 처리할 배열의 초기 위치
    private final int end;   // 이 서브태스크에서 처리할 배열의 최종 위치
    public 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() { //RecursiveTask의 추상메서드 오버라이드
        int length = end - start; // 이 태스크에서 더할 배열의 길이
        if (length <= THRESHOLD) {
            // 기준값과 같거나 작으면 순차적으로 결과를 계산한다.
            return computeSequentially();
        }
        
        // 배열의 첫 번째 절반을 더하도록 서브 태스크 생성
        ForkJoinSumCalculator leftTask =
            new ForkJoinSumCalculator(numbers, start, start + length/2);
        // ForkJoinPool의 다른 스레드로 새로 생성한 태스크를 비동기로 실행
        leftTask.fork();
        // 배열의 나머지 절반을 더하도록 서브태스크 생성
        ForkJoinSumCalculator rightTask = 
            new ForkJoinSumCalculator(numbers, start + length/2, end);
        // 두 번째 서브태스크를 동기 실행(이때 추가로 분할이 일어날 수 있다.)
        Long rightResult = rightTask.compute();
        // 첫 번째 서브태스크의 결과를 읽거나 아직 결과가 없으면 기다린다. 
        Long leftResult = leftTask.join();
        // 두 서브태스크의 결과를 조합한 값이 이 태스크의 결과다.
        return leftResult + rightResult;
    }

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

위 메서드는 n까지의 자연수 덧셈 작업을 병렬로 수행하는 방법을 더 직관적으로 보여준다. ForkJoinSumCalculator의 생성자로 원하는 수의 배열을 넘겨주면 된다.

public static long forkJoinSum(long n) {
    long[] numbers = LongStream.rangeClosed(1, n).toArray();
    ForkJoinTask<Long> task = new ForkJoinSumCalculator(numbers);
    return new ForkJoinPool().invoke(task);
}

ForkJoinPool의 invoke 메서드로 생성한 태스크를 전달했다. ForkJoinPool에서 싫행되는 마지막 invoke 메서드의 반환값은 ForkJoinSumCalculator에서 정의한 태스크의 결과가 된다.

일반적으로 애플리케이션에서는 ForkJoinPool를 필요한 곳에서 언제든 가져다 쓸 수 있도록 한 번만 인스턴스화 해서 정적 필드에 싱글턴으로 저장한다. (ForkJoinPool을 만들면서 인수가 없는 디폴트 생성자 이용)

ForkJoinSumCalculator 실행

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

위 과정을 재귀적으로 반복하면서 태스크 분할을 반복한다.

이제 각 서브태스크는 순차적으로 처리되며 포킹 프로세스로 만들어진 이진트리의 태스크를 루트에서 역순으로 방문한다.

즉, 각 서브태스크의 부분 결과를 합쳐 태스크의 최종 결과를 계산한다.

하니스로 포크/조인 프레임워크의 합계 메서드 성능 확인

System.out.println("ForkJoin sum done in: " + measureSumPerf(
    ForkJoinSumCalcouator::forkJoinSum, 10_000_000) + " msecs");

다음은 출력 결과다

ForkJoin sum done in: 41 msecs

병렬 스트림을 이용할 때보다 성능이 나빠졌다. 하지만 이는 ForkJoinSumCalculator 태스크에서 사용할 수 있도록 전체 스트림을 long[]으로 변환했기 때문이다.

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

• 두 서브태스크가 모두 시작된 다음에 join을 호출한다

  → join 매서드를 태스크에 호출하면 태스크가 생산하는 결과가 준비될 때까지 호출자를 블록시키기 때문

  그렇지 않으면 각각의 서브태스크가 다른 태스크가 끝나길 기다리는 일이 발생하며 원래 순차 알고리즘보다 느리고 복잡한 프로그램이 되어버릴 수 있다.

• RecursiveTask 내에서는 ForkJoinPool의 invoke 메서드를 사용하지 말아야 한다.

  → 순차 코드에서 병렬 계산을 시작할 때만 invoke를 사용한다.

• 서브태스크에 fork 메서드를 호출해서 ForkJoinPool의 일정을 조절할 수 있다.

  이때 왼쪽 작업 오른쪽 작업 모두에 fork 메서드를 호출하는 것보다 한쪽 작업에는 compute를 호출하는 것이 효율적이다.

  그러면 두 서브 태스크의 한 태스크에서 같은 스레드를 재사용할 수 있으므로 불필요한 태스크를 할당하는 오버헤드를 피할 수 있다.

• 포크/조인 프레임워크를 이용하는 병렬 계산은 디버깅하기 어렵다.

  보통 IDE로 디버깅할 때 스택 트레이스로 문제가 일어난 과정을 쉽게 확인할 수 있는데, 포크/조인 프레임워크에서 fork라 불리는 다른 스레드에서 compute를 호출하므로 스택 트레이스가 도움이 되지 않는다.

7.2.3 작업 훔치기

대부분의 기기에 코어가 4개 뿐이므로 천 개 이상의 서브태스크는 자원만 낭비하는 것 처럼 보인다.

하지만 실제로는 코어 개수와 관계없이 적절한 크기로 분할된 많은 태스크를 포킹하는 것이 바람직하다. 이론적으로는 코어 개수만큼 병렬화된 태스크로 작업부하를 분할하면 모든 CPU 코어에서 태스크를 실행할 것이고 크기가 같은 각각의 태스크는 같은 시간에 종료될 것이라 생각할 수 있다.

하지만 현실은 각각의 서브태스크의 작업완료 시간이 크게 달라질 수 있다. 분할 기법이 효율적이지 않았기 때문일 수도 있고 예기치 않게 디스크 접근 속도가 저하되었거나 외부 서비스와 협력하는 과정에서 지연이 생길 수 있기 때문이다.

작업 훔치기

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

한 스레드의 할일이 다 떠러진 상태에서 해당 스레드의 작업이 멈추는 것이 아니라 다른 스레드 큐의 꼬리에서 작업을 훔쳐온다. 모든 태스크가 작업을 끝날 때까지, 즉 모든 큐가 빌 때까지 이 과정을 반복한다.

따라서 태스크의 크기를 작게 나누어야 작업자 스레드 간의 작업부하를 비슷한 수준으로 유지할 수 있다.

7.3 Spliterator 인터페이스

자바 8은 Spliterator라는 새로운 인터페이스를 제공한다. 컬렉션 프레임워크에 포함된 모든 자료구조에 사용할 수 있는 디폴트 Spliterator 구현을 제공한다.

Spliterator: 분할할 수 있는 반복자

Iterator처럼 Spliterator는 소스의 요소 탐색 기능을 제공한다는 점은 같지만 Spliterator는 병렬 작업에 특화되어 있다.

• Spliterator 인터페이스

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

여기서 T는 Spliterator에서 탐색하는 요소의 형식을 가리킨다.

• tryAdvance

  Spliterator의 요소를 하나씩 순차적으로 소비하면서 탐색해야 할 요소가 남아있으면 참을 반환한다. (즉, 일반적인 Iterator 동작과 같다)

• trySplit

  Spliterator의 일부 요소(자신이 반환한 요소)를 분할해서 두 번째 Spliterator를 생성하는 메서드다.

• estimateSize

  Spliterator에서 탐색해야 할 요소 수 정보를 제공한다.

탐색해야할 요소 수가 정확하진 않더라도 제공된 값을 이용해서 더 쉽고 공평하게 Spliterator를 분할할 수 있다.

7.3.1 분할 과정

위 그림처럼 스트림 분할 과정은 재귀적으로 일어난다.

• 1단계 - 첫 번째 Spliterator에서 trySplit을 호출하면 두 번째 Spliterator가 생성된다.
• 2단계 - 두 개의 Spliterator에 trySplit을 다시 호출하면 네 개의 Spliterator가 생성된다.

이처럼 trySplit의 결과가 null(더 이상 자료구조 분할 못함)이 될 때까지 이 과정을 반복한다.

• 4단계 - 모든 trySplit의 결과가 null이면 재귀 분할 과정 종료

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

Spliterator 특성

• characteristics
  • Spliterator 자체의 특성 집합을 포함하는 int 반환
  • 이 특성을 참고해서 Spliterator를 더 잘 제어하고 최적화 가능

7.3.2 커스텀 Spliterator 구현하기

• 반복형으로 단어 수를 세는 메서드

public int countWordsIteratively(String s) {
    int counter = 0;
    boolean lastSpace = true;
    for (char c : s.toCharArray()) { // 문자열의 모든 문자를 하나씩 탐색한다.
        if (Character.isWhitespace(c)) {
            lastSpace = true;
        } else {
            if (lastSpace) counter++;
            lastSpace = false;
        }
    }
    return counter;
}

함수형으로 단어 수를 세는 메서드 재구현하기

우선 String을 스트림으로 변환해야 한다. 스트림은 int, long, double 기본형만 제공하므로 Stream를 사용해야 한다.

Stream<Character> stream = IntStream.range(0, SENTENCE.length())
                                    .mapToObj(SENTENCE::charAt);

스트림에 리듀싱 연산을 실행하면서 단어 수를 계산할 수 있다.

• 문자열 스트림을 탐색하면서 단어 수를 세는 클래스

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.counter, wordCounter.lastSpace);
    }

    public int getCounter() {
        return counter;
    }
}

• accumulate 메서드
  • WordCounter의 상태를 어떻게 바꿀 것인지, 또는 엄밀히 WordCounter는 불변 클래스이므로 새로운 WordCounter 클래스를 어떤 상태로 생성할 것인지 정의
  • 스트림을 탐색하면서 새로운 문자를 찾을 때마다 accumulate 메서드를 호출
  • 위의 반복형 메서드 countWordsIteratively에서처럼 새로운 비공백 문자를 탐색한 다음 마지막 문자가 공백이면 counter를 증가시킨다.

• combine 메서드
  • 문자열 서브스트림으로 처리한 WordCounter의 결과 합침
• 리듀싱 연산 구현

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

• 메서드 호출

Stream<Character> stream = IntStream.range(0, SENTENCE.length())
                                    .mapToObj(SENTENCE::charAt);
System.out.println("Found " + countWords(stream) + " words");

WordCounter 병렬로 수행하기

System.out.println("Found " + countWords(stream.parallel()) + " words");

병렬 스트림으로 처리하고자 했지만 결과가 이상하게 나온다.

원래 문자열을 임의의 위치에서 둘로 나누다보니 예상치 못하게 하나의 단어를 둘로 계산하는 상황이 발생할 수 있다. 순차 스트림을 병렬 스트림으로 바꿀 때 스트림 분할 위치에 따라 잘못된 결과가 나올 수 있다.

→ 문자열을 임의의 위치에서 분할하지 말고 단어가 끝나는 위치에서 분할하면 된다. 단어 끝에서 문자열을 분할하는 문자 Spliterator가 필요하다.

class WordCounterSpliterator implements Spliterator<Character> {
    private final String string;
    private int currentChar = 0;

    public WordCounterSpliterator(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<Character> spliterator = 
                    new WordCounterSpliterator(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 + NON-NULL + IMMUTABLE;
    }
}