HomeAboutMeBlogGuest
© 2025 Sejin Cha. All rights reserved.
Built with Next.js, deployed on Vercel
🪂
Parallel Stream 성능장애를 조심하라❗
🪂

Parallel Stream 성능장애를 조심하라❗

🎫 개발에 도움이 되는 지식
progress
Done
SubTag
게시일자
Jul 16, 2022 01:44 PM
Tags
Java
사전 지식Build Up💨What [Parallel Stream이란]Parallel Stream 매력적인 특징❓Why [왜 장애가 낼 수]✅How [어떻게 잘 사용해야 할까?]📌 REFER

사전 지식

  • 스레드 풀 http://gee.cs.oswego.edu/dl/html/StreamParallelGuidance.html
  • 프로세스, 스레드의 차이
 

Build Up

java8에서 최대 변경사항은 람다라고 할 수 있다. 람다식을 효과적으로 사용할 수 있도록 기존 API에 람다를 대폭 적용하였으며, 대표적인 인터페이스가 Stream 이다.
  • 스트림 인터페이스는 컬렉션을 파이프 식으로 처리하도록 하면서 고차함수로 그 구조를 추상화 한다!
  • 스트림의 장점
    • 간편하게 로직을 처리할 수 있다.
    • 가독성 향상을 기대할 수 있다.
    • Parallel Stream을 통해 쉽게 병렬연산이 간단하게 가능하게 한다.
  • 스트림의 단점
    • 병렬 연산을 함부로 난무하면 심각한 성능장애를 일으킬 수 있다.

💨What [Parallel Stream이란]

💫
Stream을 사용함으로써 쉽게 병렬처리를 할 수 있도록 하여 성능 향상을 기대할 수 있는 메소드이다.
 
  • 스트림 parallel 사용해보기
  • 실행 결과
  • 내부적으로 Parallel Stream이 common fork join pool을 사용하게 되는데, 1프로세스당 1 thread를 사용하도록 되어 있기 때문이다.
  • 예를 들어 16core 장치가 있다면, 16개의 thread를 생성할 수 있다. 나의 컴퓨터에는 5개의 core 밖에 없어서 최대 5번의 thread 까지 생성된 것이다.

Parallel Stream 매력적인 특징

  1. 쓰래드의 개수를 지정해줄 수 있다 ❗
      • Java8 이전 ExecutorService를 사용하는 경우, 다음과 같이 쓰레드의 개수를 지정해줄 수 있었습니다.
       
      • 방법
          1. Property 조정
        • 코드
        • 실행 결과
          • 100개의 쓰레드를 개수를 지정한 후 worker의 숫자가 99인 것을 확인할 수 있다.
          notion image
          1. ForkJoinPool 사용
        • code
        • 여도 마찬가지 99번째 worker가 출연한 것을 볼 수 있다.
          • notion image
        • 하지만❗ thread 개수를 지정할 수 있지만, 지정한 수만큼 새로운 thread가 생성되지 않고 처리되는 것을 확인 할 수 있다.
          • notion image
        • 병렬 스트림이 내부적으로 common ForkJoinPool을 사용하기 때문에 ForkJoinPool을 사용하는 다른 thred에 영향을 줄 수 있으며, 반대로 영향을 받기도 한다. 위 예처럼 사용하는 경우는 예외가 되겠지만, 적어도 실행환경의 성능을 별도로 고려할 필요가 생기게 된다.

❓Why [왜 장애가 낼 수]

⚠️
Paralle은 공유된 Thread pool을 사용하기 때문에 심각한 성능장애를 일으킬 수 있기 때문이다.
  • 동작원리
      • notion image
    • fork-join pool은 기본적으로 스레드 풀 서비스의 일종으로 분할 정복 알고리즘과 비슷하다고 보면 된다.
    • 다음 그림처럼 fork를 통해 쪼개고! join을 통해 합치게 된다.
      •  
      기본적으로 ExecutorService의 구현체이지만, 다른 점은 thread들이 개별 큐를 가지게 되며, 다음 그림의 B처럼 자신에게 아무런 task가 없으면 A의 task를 가져와 처리하게 됨으로써 cpu 자원이 놀지 않고 최적의 성능을 낼 수 있게 된다.
      notion image
       
       

✅How [어떻게 잘 사용해야 할까?]

💫
Parallel Stream을 이용하면 임의로 스레드 개수를 조정 할 수 있다는 큰 장점으로 작업 처리를 가속화 할 수 있는 기대를 할 수 있는데 하지만 고려할 사항이 꽤나 복잡하다는 점이다. 반드시 시간 측정을 꼭 해보고 넘어가야 하면서도 내부 동작을 정확하게 예측할 수 있어야 한다.
  • 고려사항
      • ⚠️
      ForkJoinPool의 특성상 나누어지는 Job은 균등하게 처리가 되어야 한다.
    • 작업을 분할 하기 위해 Spliator trySplit()을 사용하게 되는데, 이 분할되는 작업의 단위가 균등하게 나누어져야 하며, 나누어지는 작업에 대한 비용이 높지 않아야 순차적 방식보다 효율적으로 이루어 질 수 있다.
    • 그래서 ArrayList, Array와 같이 정확한 전체 사이즈를 알 수 있는 경우에는 분할 처리가 빠르고 비용이 적게 들게 된다. 하지만 LinkedList의 경우라면 별다른 효과를 찾기가 어렵다 (LinkedList의 사이즈는 순회를 통해 알 수 있듯이..)
    • 또한, 병렬로 처리되는 작업이 독립적이지 않다면, 수행 성능에 영향을 있을 수 있다.
      • Stream의 중간 연산 단계 중 sorted(), distinct()와 같은 작업을 수행하는 경우에는 내부적으로 상태에 대한 변수를 각 작업들이 공유하게 되어 있다. 오히려 일헌 경우에는 순차적으로 실행하는 경우가 더 효과적일 수 있다.
       

      그렇다면 언제 사용해야 할까?

      ✨
      앞서 설명한 ForkJoinPool 방식을 이용하기 때문에 분할이 잘 이루어질 수 있는 데이터 구조이거나, 작업이 독립적으면서 cpu burst 사용이 높은 작업에 적합하다고 볼 수 있다.
      http://gee.cs.oswego.edu/dl/html/StreamParallelGuidance.html → guide 참조

📌 REFER

Java8 Parallel Stream, 성능장애를 조심하세요!
Java8에서 최대 변경사항은 람다라고 할 수 있습니다. 람다식을 효과적으로 사용할 수 있도록 기존 API...
Java8 Parallel Stream, 성능장애를 조심하세요!
https://m.blog.naver.com/tmondev/220945933678
Java8 Parallel Stream, 성능장애를 조심하세요!
Java8 Stream의 parallel 처리 | Popit
이번 글에서는 Java8 Stream의 parallel에 대해 살펴보도록 하겠습니다. Stream API는 설계 시 실제 실행에 대한 부분을 추상화 시켜 각 Element에 대한 처리가 어떻게 실행되는지에 대해서는 프로그래머는 신경쓰지 않이도 된다. 즉, 병렬 처리 등도 쉽게 할 수 있게 설계되어 있다.
Java8 Stream의 parallel 처리 | Popit
https://www.popit.kr/java8-stream%EC%9D%98-parallel-%EC%B2%98%EB%A6%AC/
Java8 Stream의 parallel 처리 | Popit
OKKY | Java Stream 병렬(Parallel)은 성능을 위한것이 아닌가요? (사진 추가)
둘 다 로직은 100 * 100 * 100만큼 숫자를 합치는 로직입니다 각각 시간을 체크해봤는데 parallel를 사용한 것이 2배 넘게 소요되었습니다.. 전 병렬 처리 목적이 데이터가 많으면 효율적으로 처리하기 위함으로 생각했었고 그 의미가 성능이 빨라진다는 의미일줄 알았는데 시간 비교 해보니 충격적이네요.. 혹시 이것에 대해 아
OKKY | Java Stream 병렬(Parallel)은 성능을 위한것이 아닌가요? (사진 추가)
https://okky.kr/article/887334?note=2265642
OKKY | Java Stream 병렬(Parallel)은 성능을 위한것이 아닌가요? (사진 추가)
www.baeldung.com
https://www.baeldung.com/java-when-to-use-parallel-stream
package com.programmers.cicdtest;

import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ThreadTest {
	public static void main(String[] args) {
		ExecutorService executor = Executors.newFixedThreadPool(5);
		int taskSize = 10;
		
		for (int i = 0; i < taskSize; i++) {
			final int idx = i;
			executor.submit(() -> {
				try {
					Thread.sleep(5000);
					System.out.println(Thread.currentThread().getName() + ", index=" + idx + ", ended at" + new Date());
				} catch (InterruptedException e) {
					throw new RuntimeException(e);
				}
			});
		}

		executor.shutdown();
	}
}
pool-1-thread-1, index=0, ended atSun Jul 17 00:20:34 KST 2022
pool-1-thread-5, index=4, ended atSun Jul 17 00:20:34 KST 2022
pool-1-thread-4, index=3, ended atSun Jul 17 00:20:34 KST 2022
pool-1-thread-3, index=2, ended atSun Jul 17 00:20:34 KST 2022
pool-1-thread-2, index=1, ended atSun Jul 17 00:20:34 KST 2022
pool-1-thread-2, index=6, ended atSun Jul 17 00:20:39 KST 2022
pool-1-thread-4, index=5, ended atSun Jul 17 00:20:39 KST 2022
pool-1-thread-3, index=7, ended atSun Jul 17 00:20:39 KST 2022
pool-1-thread-5, index=8, ended atSun Jul 17 00:20:39 KST 2022
pool-1-thread-1, index=9, ended atSun Jul 17 00:20:39 KST 2022
ExecutorService executor = Executors.newFixedThreadPool(5);
System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism","100");
package com.programmers.cicdtest;

import java.util.Date;
import java.util.stream.IntStream;

public class ThreadTest {
	public static void main(String[] args) {
		// note: poll size 조정
		System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "100");

		IntStream.range(0, 100).parallel().forEach(index -> {
			System.out.println(
				"Starting " + Thread.currentThread().getName() + ",    index=" + index + ", " + new Date());
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
			}
		});
	}
}
ForkJoinPool forkjoinPool = new ForkJoinPool(${pool_size});
package com.programmers.cicdtest;

import java.util.Date;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.stream.LongStream;

public class ThreadTest {
	public static void main(String[] args) throws ExecutionException, InterruptedException {
		final int customPollSize = 100;
		ForkJoinPool forkJoinPool = new ForkJoinPool(customPollSize);

		forkJoinPool.submit(() -> {
			LongStream.rangeClosed(0, customPollSize).parallel()
				.forEach(val -> {
					System.out.println("Thread : " + Thread.currentThread().getName()
						+ ", index + " + new Date());
					try {
						Thread.sleep(5000);
					} catch (InterruptedException e) {
					}
				});
		}).get();

		forkJoinPool.shutdown();
	}
}