Java8 Parallel Stream, 효율적으로 사용해야합니다. 람다

참고

 

https://www.popit.kr/java8-stream%EC%9D%98-parallel-%EC%B2%98%EB%A6%AC/

 

https://m.blog.naver.com/tmondev/220945933678

 

---

 

Java8에서 최대 변경사항은 람다라고 할 수 있습니다. 람다식을 효과적으로 사용할 수 있도록 기존 API에 람다를 대폭 적용하였으며, 그 대표적인 인터페이스가 Stream입니다. 스트림 인터페이스는 컬렉션을 파이프 식으로 처리하도록 하면서 고차함수로 그 구조를 추상화합니다.
 
스트림을 사용하면서, 여러 줄의 코드로 작업했던 로직을 간편하게 처리 할 수 있게 되고, 가독성 또한 높아졌습니다. 특히 Parallel Stream은 병렬연산을 쉽고 간단하게 처리해주니 정말 매력적으로 보입니다.

하지만, 세상에 공짜는 없는 법!!
Parallel()은 공유된 thread pool을 사용하기 때문에 심각한 성능장애를 일으킬 수 있습니다. 본 글에서는 Parallel Stream의 동작 방식과 잠재적 위험성에 대해 논하고, 사용할 때에 고려해야 할 사항들을 이야기 해보려 합니다.

 

Parallel Stream으로 더 쉬워진,
병렬연산 처리

Java8에서의 병렬처리가 parallelStream()을 이용하면 얼마나 간단해지는지 부터 살펴보겠습니다.
Java8 이전의 병렬처리 방식은 일반적인 thread를 사용하기도 하지만, 주로 ExecutorService를 사용하였습니다.

일반 병렬처리 예제

ExecutorService executor = Executors.newFixedThreadPool(5); for (int i = 0; i < dealmaxList.size(); i++) { final int index = i; executor.submit(() -> { Thread.sleep(5000); System.out.println(Thread.currentThread().getName() + ", index=" + index + ", ended at " + new Date()); }); } executor.shutdown();

위의 코드를 parallelStream()을 이용하면 다음과 같이 작성할 수 있습니다. 

Java Parallel Stream 병렬처리 예제

dealmaxList.parallelStream().forEach(index -> { System.out.println("Starting " + Thread.currentThread().getName() + ", index=" + index + ", " + new Date()); try { Thread.sleep(5000); } catch (InterruptedException e) { } });

예제의 실행 결과는 메인 쓰레드를 포함해서 모두 4개의 thread가 실행되었음을 알 수 있습니다. 

실행결과

이러한 결과가 나타나는 이유는 내부적으로 Parallel Stream이 common fork join pool을 사용하게 되는데, 1 프로세서 당 1 thread를 사용하도록 되어있기 때문입니다. 예를들어 16 core 장치가 있다고 하면, 16개의 thread를 생성할 수 있습니다. 위 예제의 실행 결과, 4 threads는 실행환경의 맥북이 4 core이기 때문입니다. 

Parallel Stream의 매력적인 특징,
Thread의 크기 제어

Java8 이전 ExecutorService를 사용하는 경우, 다음과 같이 쓰레드의 개수를 지정해줄 수 있었습니다.

ExecutorService executor = Executors.newFixedThreadPool(5);

그렇다면, Parallel Stream에선 어떨까요? 
Parallel Stream에서 개발자가 임의로 Pool의 크기를 조절하는 방법은 두 가지가 있습니다.

1) Property값을 설정하는 방법

“java.util.concurrent.ForkJoinPool.common.parallelism” Property 값을 설정하는 방법입니다. 

예제1

System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism","6");

위 1번 예제에 코드를 추가로 설정하고, 실행해 보면 결과는 다음과 같이 메인을 포함하여 6 thread로 늘어난 것을 알 수 있습니다.

실행결과

2) ForkJoinPool을 사용하는 방법

예제2

ForkJoinPool forkjoinPool = new ForkJoinPool(5); forkjoinPool.submit(() -> { dealmaxList.parallelStream().forEach(index -> { System.out.printIn("Thread : " + Thread.currentThread().getName() + ", index + ", " + new Date()); try{ Thread.sleep(5000); } catch (InterruptedException e){ } }); }).get();

실행결과

ForkJoinPool 생성자에 Thread 개수를 지정하여 사용할 수 있으며, 지정한 수만큼 새로운 Thread가 생성되지 않고 처리되는 것을 확인할 수 있습니다.

Parallel Stream이 내부적으로 common ForkJoinPool을 사용하기 때문에 ForkJoinPool을 사용하는 다른 thread에 영향을 줄 수 있으며, 반대로 영향을 받을 수도 있게 됩니다. 위 예제2처럼 사용하는 경우에는 예외가 되겠지만, 적어도 실행환경의 성능은 별도로 고려할 필요가 생기게 됩니다.
 

ForkJoinPool의 동작방식

앞서 언급된 ForkJoinPool에 대해서 간략하게 알아보겠습니다.
Java7에서 처음 소개된 fork-join pool은 기본적으로 쓰레드풀 서비스의 일종으로 분할정복 알고리즘과 비슷하다고 보면 되는데, 다음 그림처럼 fork를 통해 task를 분담하고 join을 통해 합치게 됩니다.

https://javatechnocampus.wordpress.com/2015/10/03/544/

기본적으로는 ExecutorService의 구현체이지만, 다른 점은 각 thread들이 개별 큐를 가지게 되며, 다음 그림의 B처럼 자신에게 아무런 task가 없으면 A의 task를 가져와 처리하게 됨으로써 CPU자원이 놀지 않고 최적의 성능을 낼 수 있게 됩니다.

Parallel Stream을 이용한다면,
고려해야 할 사항들

Parallel Stream을 이용하면 임의로 스레드 개수를 조정할 수 있어 잠재적으로 작업 처리가 가속화되지 않을까 기대하게 됩니다. 하지만 사용할 때에는 고려해야 할 사항들이 많습니다. 
 
ForkJoinPool의 특성상 나누어지는 job은 균등하게 처리가 되어야 합니다.
Parallel Stream은 작업을 분할하기 위해 Spliterator의 trySplit()을 사용하게 되는데, 이 분할되는 작업의 단위가 균등하게 나누어져야 하며, 나누어지는 작업에 대한 비용이 높지 않아야 순차적 방식보다 효율적으로 이루어질 수 있습니다. array, arrayList와 같이 정확한 전체 사이즈를 알 수 있는 경우에는 분할 처리가 빠르고 비용이 적게 들지만, linkedList의 경우라면 별다른 효과를 찾기가 어렵습니다.

public Spliterator<T> trySplit() { int lo = index, mid = (lo + fence) >>> 1; return (lo >= mid) ? null : new ArraySpliterator<>(array, lo, index = mid, characteristics); }

또한, 병렬로 처리되는 작업이 독립적이지 않다면, 수행 성능에 영향이 있을 수 있습니다.
예를 들어, stream의 중간 단계 연산 중 sorted(), distinct()와 같은 작업을 수행하는 경우에는 내부적으로 상태에 대한 변수를 각 작업들이 공유(synchronized)하게 되어 있습니다. 이러한 경우에는 순차적으로 실행하는 경우가 더 효과적일 수 있습니다.

그렇다면,
Parallel Stream은 언제 사용해야 할까?

Parallel Stream은 앞서 설명한 ForkJoinPool 방식을 이용하기 때문에 분할이 잘 이루어질 수 있는 데이터 구조이거나, 작업이 독립적이면서 CPU사용이 높은 작업에 적합하다고 볼 수 있습니다.

Parallel Stream을 어느 때 사용해야 하는지에 대한 판단 기준은 다음 링크를 통해 좀 더 자세히 확인하실 수 있습니다. (http://gee.cs.oswego.edu/dl/html/StreamParallelGuidance.html)
 

 

마치며

Parallel Stream을 사용해 보고자 내부 동작 원리들을 확인했습니다.
사용은 간단하지만,  독립적으로 실행되는 어플리케이션이 아니라면, 사용에 신중해야 한다고 느꼈습니다. 따라서 문제가 없을지 추측하는 것보다는 테스트를 통해 순차 연산과 비교해서 결과 값의 차이는 없는지, 처리시간의 단축이 병렬화 처리로 인해 사용되는 비용보다 효과가 높은지를 판단해야 할 것 입니다. 

참고 링크
http://coopsoft.com/ar/Calamity2Article.html  
http://www.ibm.com/developerworks/library/j-java-streams-5-brian-goetz/index.html

https://m.blog.naver.com/tmondev/220945933678