java8 병렬 스트림 효율적으로 사용하는 방법

1. 병렬 스트림이란 ?

병렬 스트림이란 스트림에 각 요소를 청크(Chunk)로 분할한 스트림이다.

따라서 병렬 스트림을 이용해 멀티코어 프로세스가 각각의 청크를 처리하도록 할당할 수 있다.

1.1 병렬 스트림 속도차이 확인하기

1.1.1 첫번째 시도

무한스트림 생성이 가능하다는 성질을 이용하여

숫자 n 개를 받아 1부터 n 까지의 모든 숫자를 합계로 반환하는 메서드를 구현해보자

일반 스트림 연산

public static long sequentialSum(long n) { return Stream.iterate(1L, i -> i + 1).limit(n).reduce(0L, Long::sum); }

전통적인 자바 반복문을 통한 연산

public static long iterativeSum(long n) { long result = 0; for (long i = 1L; i <= n; i++) { result += i; } return result; }

병렬 스트림을 이용한 연산

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

자 이제 이 3가지 방법에 관해 성능 측정을 시도해보자

성능측정함수

public static long measureSumPerf(Function<Long, Long> adder, long n) { long fastest = Long.MAX_VALUE; for (int i = 0; i < 10; i++) { long start = System.nanoTime(); long sum = adder.apply(n); long duration = (System.nanoTime() - start) / 1_000_000; System.out.println("Result : " + sum); if (duration < fastest) { fastest = duration; } } return fastest; }

실제 성능 측정을 해보았다

실험 환경은

Intel(R) Core(TM) i5-7200U CPU @ 2.50GHz

로 구성되어 졌다.

System.out.println("Sequential sum done in : " + measureSumPerf(Main::sequentialSum, 10_000_000)); System.out.println("ParallelSum sum done in : " + measureSumPerf(Main::parallelSum, 10_000_000)); System.out.println("IterativeSum sum done in : " + measureSumPerf(Main::iterativeSum, 10_000_000));

결과

Result : 50000005000000

               .

               .

               .

Result : 50000005000000

Sequential sum done in : 99

Result : 50000005000000

               .

               .

               .

Result : 50000005000000

ParallelSum sum done in : 153

Result : 50000005000000

               .

               .

               .

Result : 50000005000000

IterativeSum sum done in : 3

실패한 이유는??

병렬 버전이 순차 버전보다 훨씬 느리게 동작한다.

그 이유는 무엇일까?

두가지에 이유가 존재한다.

첫번째는

• iterate 가 박싱된 객체를 생성함으로 이를 다시 언박싱 하는 과정이 필요하다.

(위 내용은 앞서 포스팅한 기본형 특화 스트림 부분을 참고하기 바란다.)

두번째는

• iterate 는 병렬로 실행될 수 있도록 독립적인 청크로 분할하기가 어렵다

두번째 문제는 조금은 크리티컬할 수 있는 문제이다.

병렬로 실행되기 위해서는 그에 합당한 모델이 필요하다.

하지만 iterate 는 다음 결과가 입력이 되고 다시 연산 후 출력이 결정되기 때문에 이를 청크로 분할 하기가 어렵다

이와 같은 상황에서는 각각에 청크 및 스트림 요소들이 리듀싱연산에서 분리되어 일어나기가 어렵다.

병렬 스트림을 사용하여 각각에 다른 스레드에서 수행되었지만 결국 순차처리 방법과 크게 다르지 않았다.

오히려 스레드를 할당하고 변경하는 오버헤드만 증가하는 바람에 더 느린 속도가 나와버렸다.

1.1.2 두번째 시도

그렇다면 위의 두가지 이유를 한번 해결해 보자

첫번째 문제인 언박싱 문제는 LongStream 이라는 기본형 특화 스트림을 사용하면 쉽게 해결 할 수 있을 것이다.

그렇다면 두번째 문제는 청크분할 문제는 어떤 방식이 있을까?

바로 rnageClose 라는 LongStrea, IntStream 에서 제공하는 정적 메소드 rangeClosed 를 활용하면 쉽게 해결 될 수 있다.

이 메서드는 숫자의 범위를 생산해 쉽게 청크로 분할 할 수 있도록 한다.

이를 통해 만들어진 새로운 병렬처리 방법은 아래와 같다.

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

이제 시간을 다시한번 측정해보자

System.out.println("ParallelSum sum done in : " + measureSumPerf(Main::parallelRangeSum, 10_000_000));

결과 는 

Result : 50000005000000

               .

               .

               .

Result : 50000005000000

ParallelSum sum done in : 2

결과는 순차 실행보다 빠른 성능을 보여주었다.

1.2 병렬 스트림의 올바른 사용법

기본적으로 병렬스트림을 사용하면서 발생하는 대부분의 문제는 바로 공유된 상태를 바꾸는 알고리즘을 사용하는 데 있다

다음은 n까지의 자연수를 더하면서 공유된 누적자를 바꾸는 프로그램을 구현한 코드다.

public static long sideEffectSum(long n) { Accumulator accumulator = new Accumulator(); LongStream.rangeClosed(1, n).forEach(accumulator::add); return accumulator.getTotal(); } public class Accumulator { private long total = 0; public void add(long value) { this.total += value; } getter/setter... }

이상해 보일것이 없는 코드이다.

total  이라는 누적자를 초기화하고, 누적자에 하나씩 순차적으로 더하는 코드이다.

그럼 이 코드에 어떤 문제가 있을까?

위 코드는 본질적으로 순차적으로 실행할 수 있도록 구현되어 있다.

특히 total 이라는 누적자에 병렬 처리시 다수의 스레드가 동시 접근하면 엄청난 문제가 발생해버린다.

여기에 동기화를 사용하여 문제를 해결하다보면 결국 스레드 오버해드만 남은 플로우가 되어버린다.

그럼 직접 어떤 문제가 발생하는지 알아보자

public static long sideEffectParallelSum(long n) { Accumulator accumulator = new Accumulator(); LongStream.rangeClosed(1, n).parallel().forEach(accumulator::add); return accumulator.getTotal(); } System.out.println(measureSumPerf(Main::sideEffectParallelSum, 10_000_000));

결과는 아래와 같다

Result : 25698090864402

Result : 5536190918784

Result : 10869687842769

Result : 12840926759819

Result : 13986224333863

Result : 16117123421142

Result : 12020482325134

Result : 9343904078545

Result : 13280567842732

Result : 6441209927034

메서드의 성능은 볼 필요도 없이 올바른 결과값(50000005000000)조차 나오지 않았다.

여러 스레드가 동시에 누적자(total)에 접근하는 바람에 이런 문제가 발생하였다.

여러 스레드가 접근하는 forEach 안에서 add메서드를 호출하면 당연히 이런 문제가 발생할 수밖에 없다.

결국 상태공에 따른 부작용을 피해야 하는 것이다.

1.3 병렬 스트림을 효과적으로 사용하는 방법

몇개 이상의 요소일때 와 같은 조건은 병렬스트림 사용 기준이 될 수 없다.

가장 적합한 조건일때 병렬 스트림을 사용해야 한다.

• 측정하라! 순차 스트림을 병렬 스트림으로 바꾸는 것은 쉽다. 하지만 병렬 스트림으로 바꾼다고하여 이것이 짜잔 하면서 환골탈태를 할 수는 없다. 따라서 항상 어떤 방식이 좋을지 성능 측정을 통하여 결정하도록 하자.

• 박싱과 언박싱 문제는 성능을 크게 좌우한다. 기본형 스트림을 사용하도록하자.

• 순차 스트림보다 병렬 스트림에서 성능이 떨어지는 연산들이 있다. 특히 limit나 findFirst처럼 요소의 순서에 의존하는 연산을 병력 스트림에서 수행하려면 오히려 비싼 비용을 치루게 된다.

• 스트림에서 수행하는 전체 파이프라인 연산 비용을 고려해야한다. 처리해야할 요소 수가 N 이고 비용이 Q라면 N*Q의 비용이 들 것이다. Q가 높아진다면 당연히 병렬스트림을 통해서 성능 개선할 수 있는 가능성이 열려있다는 소리다.

• 소량의 데이터는 오히려 병렬스트림이 도움이 되지 않는다. 그 이유는 병렬 스트림에서 발생하는 스레드 오버헤드가 비용이 더 크기 때문이다.

• 스트림을 구성하는 자료구조가 적절해야한다. 예를 들어 LinkedList와 ArrayList는 아주 큰 차이가 있다. LinkedList의 요소를 분할하기 위해서는 당연히 모든 요소를 탐색해야하는 과정이 필요하다. 하지만 ArrayList같은 경우는 인덱싱을 통한 요소 분할이 가능함으로 훨씬 효율적인 결과를 도출하게 된다.

또한 병렬화에 잘 어울리는 스트림 요소들은 아래와 같다.

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

글을 정리하며...

병렬 스트림을 사용할 때는 고려해야 할께 많다. 

특히 복잡한 연산에서 잘못된 병렬화는 찾기가 힘들고 오히려 독이 될 수도 있다는 것을 알게 되었다.

자바 8에 훌륭하고 좋은 기능들이 추가된건 확실하지만 오히려 이런 기능들을 사용할 때는 개발자의 더욱 전문화된 사고와 개념을 필요로 한다는 것을 또한번 생각할 수 있는 부분이다.

이 포스팅은 한빛미디어에 'JAVA 8 in Action' 내용을 발췌 또는 인용을 통해 작성되었습니다.