Java8에서의 병렬처리 방법

Donghyun Kim·2023년 2월 16일

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()을 이용하면 다음과 같이 작성할 수 있습니다.

Parallel Stream 병렬처리 예제

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

실행 결과

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

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을 통해 합치게 됩니다.

기본적으로는 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)하게 되어 있습니다. 이러한 경우에는 순차적으로 실행하는 경우가 더 효과적일 수 있습니다.