Stream API 병렬 데이터 처리하기

/** * 순차 리듀싱 * 무한 스트림을 사용해 합계를 계산한다. */ 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); }

@State(Scope.Thread) //벤치 마크에 사용되는 Argument의 상태를 Thread별로 초기화한다. @BenchmarkMode(Mode.AverageTime) // 벤치마크 대상 메서드를 실행하는데 걸린 평균 시간 측정 @OutputTimeUnit(TimeUnit.MILLISECONDS)// 벤치마크 결과를 밀리초 단위로 출력 @Fork(value = 2, jvmArgs = {"-Xms4g", "-Xmx4g"})//4Gb의 힙 공간을 제공한 환경에서 두 번 벤치마크 수행 public class ParallelStreamBenchMark { private static final long N = 10_000_000L; @Benchmark //벤치마크 대상 메서드 public long parallelRangeSum() { return LongStream.rangeClosed(1, N) .parallel() .reduce(0L, Long::sum); } @TearDown(Level.Invocation) // 매 번 벤치마크를 실행한 다음 가비지 컬렉터 동작 시도 public void tearDown() { System.gc(); } public static void main(String[] args) throws IOException, RunnerException { Options opt = new OptionsBuilder() .include(ParallelStreamBenchMark.class.getSimpleName()) .warmupIterations(10) // 사전 테스트 횟수 .measurementIterations(10) // 실제 측정 횟수 .forks(1) // .build(); new Runner(opt).run(); // 벤치마킹 시작 } }

public long rangeSum(long n) { return LongStream.rangeClosed(1, n) .reduce(0L, Long::sum); }

public long parallelRangeSum(long n) { return LongStream.rangeClosed(1, n) .parallel() .reduce(0L, Long::sum); }

public class SideEffectAccumulator { private static class Accumulator { public long total = 0; public void add(long value) { total += value; } } public long sideEffectSum(long n) { Accumulator acc = new Accumulator(); LongStream.rangeClosed(1, n).forEach(acc::add); return acc.total; } public long sideEffectParallelSum(long n) { Accumulator acc = new Accumulator(); LongStream.rangeClosed(1, n).parallel().forEach(acc::add); return acc.total; } public static void main(String[] args) { final Long N = 10_000_000L; SideEffectAccumulator sea = new SideEffectAccumulator(); System.out.println("sideEffectSum: "+ sea.sideEffectSum(N)); System.out.println("sideEffectParallelSum: "+ sea.sideEffectParallelSum(N)); } }

protected abstract R compute();

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

public class ForkJoinSumCalculator extends 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; } @Override protected Long compute() { 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 rightResult + leftResult; } //더 분할할 수 없을 경우 서브태스크의 결과를 계산하는 알고리즘 private long computeSequentially() { long sum = 0; for (int i = start; i < end; i++) { sum += numbers[i]; } return sum; } public static long forkJoinSum(int n) { long[] numbers = LongStream.rangeClosed(1, n).toArray(); ForkJoinSumCalculator task = new ForkJoinSumCalculator(numbers); return new ForkJoinPool().invoke(task); } public static void main(String[] args) { long l = forkJoinSum(10_000_000); System.out.println("l = " + l); } }

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

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 = false; } } return counter; }

public static void main(String[] args) { final String SENTENCE = "Nel mezzo del cammin di nostra vita " + "mi ritrovai in una selva oscura " + "ch la dritta via era smarrita "; System.out.println("Found " + CountWord.countWordsIteratively(SENTENCE) + " words"); }

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

public static int countWord(Stream<Character> stream) { WordCounter counter = stream.reduce(new WordCounter(0, true), //초기 값 WordCounter::accumulate, //누적 로직 WordCounter::combine); //병합 로직 return counter.getCounter(); } public static void main(String[] args) { final String SENTENCE = "Nel mezzo del cammin di nostra vita " + "mi ritrovai in una selva oscura " + "ch la dritta via era smarrita "; Stream<Character> stream = IntStream.range(0, SENTENCE.length()) .mapToObj(SENTENCE::charAt); System.out.println("result = " + countWord(stream)); }

//before case String a = "mi ritrovai in una selva oscura "; // 이 단어를 stream.parallel()로 분할했을 경우 String splitted_before_a = "mi ritrova in un"; String splitted_after_a = "a selva oscura";

public 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(); // 파싱할 문자열을 순차 처리할 정도로 잘개 쪼갰을 경우 null을 반환한다. if (currentSize < 10) { return null; } //파싱할 문자열의 중간을 분할 위치로 설정한다. for (int splitPos = currentSize / 2 + currentChar; splitPos < string.length(); splitPos++) { //다음 공백이 나올 때까지 분할 위치를 뒤로 이동시킨다. if (Character.isWhitespace(string.charAt(splitPos))) { //처음부터 분할 위치까지 문자열을 파싱할 새로운 WordCounterSpliterator를 생성한다. WordCounterSpliterator spliterator = new WordCounterSpliterator(string.substring(currentChar, splitPos)); currentChar = splitPos; //이 WordCounterSpliterator의 시작 위치를 분할 위치로 설정한다. return spliterator;//공백을 찾고 문자열을 분리했기에 루프를 종료한다. } } return null; } @Override public long estimateSize() { return string.length() - currentChar; } @Override public int characteristics() { return ORDERED + SIZED + SUBSIZED + NONNULL + IMMUTABLE; } }

WordCounterSpliterator wordCounterSpliterator = new WordCounterSpliterator(SENTENCE); Stream<Character> stream = StreamSupport.stream(wordCounterSpliterator, true); System.out.println("countWord = " + countWord(stream));