Portfolio

사전 지식 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 사용해보기


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

내부적으로 Parallel Stream이 common fork join pool을 사용하게 되는데, 1프로세스당 1 thread를 사용하도록 되어 있기 때문이다.

예를 들어 16core 장치가 있다면, 16개의 thread를 생성할 수 있다. 나의 컴퓨터에는 5개의 core 밖에 없어서 최대 5번의 thread 까지 생성된 것이다.

Parallel Stream 매력적인 특징

쓰래드의 개수를 지정해줄 수 있다 ❗

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


ExecutorService executor = Executors.newFixedThreadPool(5);

방법

Property 조정


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

실행 결과

100개의 쓰레드를 개수를 지정한 후 worker의 숫자가 99인 것을 확인할 수 있다.

ForkJoinPool 사용


ForkJoinPool forkjoinPool = new ForkJoinPool(${pool_size});

code


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

여도 마찬가지 99번째 worker가 출연한 것을 볼 수 있다.

하지만❗ thread 개수를 지정할 수 있지만, 지정한 수만큼 새로운 thread가 생성되지 않고 처리되는 것을 확인 할 수 있다.

병렬 스트림이 내부적으로 common ForkJoinPool을 사용하기 때문에 ForkJoinPool을 사용하는 다른 thred에 영향을 줄 수 있으며, 반대로 영향을 받기도 한다. 위 예처럼 사용하는 경우는 예외가 되겠지만, 적어도 실행환경의 성능을 별도로 고려할 필요가 생기게 된다.

❓Why [왜 장애가 낼 수]

⚠️

Paralle은 공유된 Thread pool을 사용하기 때문에 심각한 성능장애를 일으킬 수 있기 때문이다.

동작원리

fork-join pool은 기본적으로 스레드 풀 서비스의 일종으로 분할 정복 알고리즘과 비슷하다고 보면 된다.
다음 그림처럼 fork를 통해 쪼개고! join을 통해 합치게 된다.

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

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

💫

Parallel Stream을 이용하면 임의로 스레드 개수를 조정 할 수 있다는 큰 장점으로 작업 처리를 가속화 할 수 있는 기대를 할 수 있는데 하지만 고려할 사항이 꽤나 복잡하다는 점이다. 반드시 시간 측정을 꼭 해보고 넘어가야 하면서도 내부 동작을 정확하게 예측할 수 있어야 한다.

고려사항

⚠️

ForkJoinPool의 특성상 나누어지는 Job은 균등하게 처리가 되어야 한다.

작업을 분할 하기 위해 Spliator trySplit()을 사용하게 되는데, 이 분할되는 작업의 단위가 균등하게 나누어져야 하며, 나누어지는 작업에 대한 비용이 높지 않아야 순차적 방식보다 효율적으로 이루어 질 수 있다.
그래서 ArrayList, Array와 같이 정확한 전체 사이즈를 알 수 있는 경우에는 분할 처리가 빠르고 비용이 적게 들게 된다. 하지만 LinkedList의 경우라면 별다른 효과를 찾기가 어렵다 (LinkedList의 사이즈는 순회를 통해 알 수 있듯이..)
또한, 병렬로 처리되는 작업이 독립적이지 않다면, 수행 성능에 영향을 있을 수 있다.