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 = 1L; i <= n; i++) {
        result += i;
    }
    return result;
}

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

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

@State(Scope.Thread)
@BenchmarkMode(Mode.AverageTime) //벤치마크 대상 메서드를 실행하는데 걸린 평균 시간 측정
@OutputTimeUnit(TimeUnit.MILLISECONDS) //벤치마크 결과를 밀리초 단위로 출력
@Fork(value = 2, jvmArgs = {"-Xms4G", "-Xms4G"}) //4Gb의 힙 공간을 제공한 환경에서 두 번 벤치마크를 수행해 결과의 신뢰성 확보
@Measurement(iterations = 20)
@Warmup(iterations = 3)
public class ParallelStreamBenchmark {
    private static final long N = 10_000_000L;

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

    @TearDown(Level.Invocation) //매 번 벤치마크를 실행한 다음에는 가비지 컬렉터 동작 시도
    public void tearDown() {
        System.gc();
    }
}

@State(Scope.Thread)
@BenchmarkMode(Mode.AverageTime)
@OutputTimeUnit(TimeUnit.MILLISECONDS)
@Fork(value = 2, jvmArgs = { "-Xms4G", "-Xmx4G" })
@Measurement(iterations = 20)
@Warmup(iterations = 3)
public class ParallelStreamBenchmark {

  private static final long N = 10_000_000L;

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

  @TearDown(Level.Invocation) //매 번 벤치마크를 실행한 다음에는 가비지 컬렉터 동작 시도
  public void tearDown() {
    System.gc();
  }
}

@State(Scope.Thread)
@BenchmarkMode(Mode.AverageTime)
@OutputTimeUnit(TimeUnit.MILLISECONDS)
@Fork(value = 2, jvmArgs = { "-Xms4G", "-Xmx4G" })
@Measurement(iterations = 20)
@Warmup(iterations = 3)
public class ParallelStreamBenchmark {

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

  @TearDown(Level.Invocation) //매 번 벤치마크를 실행한 다음에는 가비지 컬렉터 동작 시도
  public void tearDown() {
    System.gc();
  }
}

@State(Scope.Thread)
@BenchmarkMode(Mode.AverageTime)
@OutputTimeUnit(TimeUnit.MILLISECONDS)
@Fork(value = 2, jvmArgs = { "-Xms4G", "-Xmx4G" })
@Measurement(iterations = 20)
@Warmup(iterations = 3)
public class ParallelStreamBenchmark {

  private static final long N = 10_000_000L;
  
  @Benchmark
  public long rangedSum() {
    return LongStream.rangeClosed(1, N).reduce(0L, Long::sum);
  }
  
  @TearDown(Level.Invocation)
  public void tearDown() {
    System.gc();
  }

}

@State(Scope.Thread)
@BenchmarkMode(Mode.AverageTime)
@OutputTimeUnit(TimeUnit.MILLISECONDS)
@Fork(value = 2, jvmArgs = { "-Xms4G", "-Xmx4G" })
@Measurement(iterations = 20)
@Warmup(iterations = 3)
public class ParallelStreamBenchmark {

  private static final long N = 10_000_000L;
  
  @Benchmark
  public long parallelRangedSum() {
    return LongStream.rangeClosed(1, N).parallel().reduce(0L, Long::sum);
  }
  
  @TearDown(Level.Invocation)
  public void tearDown() {
    System.gc();
  }

}

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

  public static class Accumulator {

    private long total = 0;

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

  }
}

public class ParallelStreams {

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

public class ParallelStreamsHarness {
  public static void main(String[] args) {
    System.out.println("SideEffect parallel sum done in: " + measurePerf(ParallelStreams::sideEffectParallelSum, 10_000_000L) + " msecs" );
  }
}

protected abstract R compute();

public class ParallelStreamsHarness {
  public static final ForkJoinPool FORK_JOIN_POOL = new ForkJoinPool();
}

import static modernjavainaction.chap07.ParallelStreamsHarness.FORK_JOIN_POOL;

import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;
import java.util.stream.LongStream;

//RecursiveTask를 상속받아 포크/조인 프레임워크에서 사용할 태스크를 생성
public class ForkJoinSumCalculator extends RecursiveTask<Long> {

  //이 값 이하의 서브태스크는 더 이상 분할할 수 없음
  public static final long THRESHOLD = 10_000;

  private final long[] numbers; //더할 숫자 배열
  private final int start; //이 서브태스크에서 처리할 배열의 초기 위치
  private final int end; //최종 위치

  //메인 태스크를 생성할 때 사용할 공개 생성자
  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;
  }

  //RecursiveTask의 추상 메서드 오버라이드
  @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 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) {
    //LongStream으로 n까지의 자연수를 포함하는 배열을 생성
    long[] numbers = LongStream.rangeClosed(1, n).toArray();
    //생성된 배열을 ForkJoinSumCalculator의 생성자로 전달해서 ForkJoinTask를 생성
    ForkJoinTask<Long> task = new ForkJoinSumCalculator(numbers);
    //생성한 태스크를 새로운 ForkJoinPool의 invoke 메서드로 전달
    return FORK_JOIN_POOL.invoke(task);
    //ForkJoinPool에서 실행되는 마지막 invoke 메서드의 반환값은 ForkJoinSumCalculator에서 정의한 태스크의 결과가 됩니다.
  }

}

public interface Spliterator<T> { // T: 탐색하는 요소의 형식

    // tryAdvance: Spliterator의 요소를 하나씩 순차적으로 소비하면서 탐색해야 할 요소가 남아있으면 참을 반환 (Iterator 처럼)
    boolean tryAdvance(Consumer<? super T> action);
    
    // trySplit: Spliterator의 일부 요소(자신이 반환한 요소)를 분할해서 두 번째 Spliterator를 생성하는 메서드
    Spliterator<T> trySplit();
    
    // estimateSize: 탐색해야 할 요소 수 정보를 제공
    long estimateSize();
    
    int characteristics();
}

int characteristics();

import java.util.Spliterator;
import java.util.function.Consumer;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

public class WordCount {

  public static final String SENTENCE =
      " Nel   mezzo del cammin  di nostra  vita "
      + "mi  ritrovai in una  selva oscura"
      + " che la  dritta via era   smarrita ";

  public static void main(String[] args) {
    System.out.println("Found " + countWordsIteratively(SENTENCE) + " words");
  }

  public static 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 = Character.isWhitespace(c);
      }
    }
    return counter;
  }
}

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

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

  //두 WordCounter의 counter값을 더함
  public WordCounter combine(WordCounter wordCounter) {
    return new WordCounter(counter + wordCounter.counter, wordCounter.lastSpace); // 마지막 공백은 신경 안씀
  }

  public int getCounter() {
    return counter;
  }

}

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

public static void main(String[] args) {
    System.out.println("Found " + countWords(SENTENCE) + " words");
}

public static void main(String[] args) {
    System.out.println("Found " + countWords(SENTENCE) + " words");
}

public static int countWords(String s) {
    Stream<Character> stream = IntStream.range(0, s.length()).mapToObj(SENTENCE::charAt).parallel();
    Spliterator<Character> spliterator = new WordCounterSpliterator(s);
    return countWords(stream);
}

private static class WordCounterSpliterator implements Spliterator<Character> {

    private final String string;
    private int currentChar = 0;

    private WordCounterSpliterator(String string) {
        this.string = string;
    }

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

    @Override
    public Spliterator<Character> trySplit() {
        int currentSize = string.length() - currentChar;
        //파싱할 문자열을 순차 처리할 수 있을만큼 충분히 작아졌음을 알리는 null 반환
        if (currentSize < 10) {
            return null;
        }
        //파싱할 문자열의 중간을 분할 위치로 설정
        for (int splitPos = currentSize / 2 + currentChar; splitPos < string.length(); splitPos++) {
            //다음 공백이 나올 때까지 분할 위치를 뒤로 이동시킴
            if (Character.isWhitespace(string.charAt(splitPos))) {
                //처음부터 분할 위치까지 문자열을 파싱할 새로운 WordCounterSpliterator를 생성
                Spliterator<Character> spliterator = new WordCounterSpliterator(string.substring(currentChar, splitPos));
                //이 WordCounterSpliterator의 시작 위치를 분할 위치로 설정
                currentChar = splitPos;
                //공백을 찾았고 문자열을 분리했으므로 루프를 종료
                return spliterator;
            }
        }
        return null;
    }

    //탐색해야 할 요소의 개수
    @Override
    public long estimateSize() {
        //파싱할 문자열 전체 길이 - 현재 반복중인 위치
        return string.length() - currentChar;
    }

    @Override
    public int characteristics() {
        /*
          ORDERED: 문자열의 문자 등장 순서가 유의미함
          SIZED: estimatedSize 메서드의 반환값이 정확함
          SUBSIZED: trySplit으로 생성된 Spliterator도 정확한 크기를 가짐
          NONNULL: 문자열에는 null 문자가 존재하지 않음
          IMMUTABLE: 문자열 자체가 불변 클래스이므로 문자열을 파싱하면서 속성이 추가되지 않음
        */
        return ORDERED + SIZED + SUBSIZED + NONNULL + IMMUTABLE;
    }

}

public static void main(String[] args) {
    System.out.println("Found " + countWords(SENTENCE) + " words");
}

public static int countWords(String s) {
    Spliterator<Character> spliterator = new WordCounterSpliterator(s);
    //true는 병렬 스트림 생성 여부를 지시
    Stream<Character> stream = StreamSupport.stream(spliterator, true);

    return countWords(stream);
}