[Modern Java in Action] Chapter7. 병렬 데이터 처리와 성능

protected abstract R compute();

// RecursiveTask 를 상속받아 포크/조인 프레임워크에서 사용할 태스크 생성
public class ForkJoinSumCalculator extends java.util.concurrent.RecursiveTask<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;
     }
     
     // ForkJoinPool 에 전달되면 실행되는 메서드
     @Override
     protected Long compute() {
         int length = end - start;
         if (length <= THRESHOLD) {
             return computeSequentially(); // 기준값 이하라면 결과 계산
         }
         
         // 배열의 절반씩 더하도록 서브태스크들 생성
         ForkJoinSumCalculator leftTask = new ForkJoinSumCalculator(numbers, start, start + length/2); 
         leftTask.fork();  // ForkJoinPool 의 다른 스레드로 생성한 태스크를 비동기로 실행
         ForkJoinSumCalculator rightTask = new ForkJoinSumCalculator(numbers, start + length/2, end);
         Long rightResult = rightTask.compute(); // 두번째 서브태스크를 동기 실행, 분할이 일어날 수 있음
         Long rightResult = leftTask.join();  // 첫번째 서브태스크의 결과를 가져오거나 대기
         return leftResult + rightResult;
     }
     
     private long computeSequentially() {
         long sum = 0;
         for (int i = start; i < end; i++) {
             sum += numbers[i];
         }
         return sum;
     }
}

// 사용 코드
long[] numbers = LongStream.rangeClosed(1, n).toArrray();
ForkJoinTask<Long> task = new ForkJoinSumCalculator(numbers); 
new ForkJoinPool().invoke(task);

publid interface Spliterator<T> { 
    // 탐색해야 할 요소가 남아있으면 참
    boolean tryAdvance(Consumer<? super T> action);
    // 일부 요소를 분할해서 두번째 Spliterator 를 생성하는 메서드
    // null 이 반활 될 때까지 새로운 Spliterator 생성
    Spliterator<T> trySplit();
    // 메서드로 탐색해야 할 요소 수
    long estimateSize();
    // Spliterator 특성 집합을 포함하는 int 반환
    int characteristics();
}

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();    // 소비할 문자 남아있는지 확인
    }
    // 반복될 자료구조를 분할하는 메서드
    @Override
    public Spliterator<Character> trySplit() {
        int currentSize = string.length() - currentChar;
        if (currentSize < 10) {
            return null;  // 순차 처리 가능할 정도로 작아졌으므로 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;
    }
}

Spliterator<Character> spliterator = new WordCounterSpliterator(SENTENCE);
Stream<Character> stream = StreamSupport.stream(spliterator, true);  // 두 번째 불리언 인수는 병렬 스트림 생성 여부