<div id="readability-page-1" class="page"><div><img src="https://velog.velcdn.com/images/joosing/post/f547dbda-95bb-4fc8-bcf2-e9d24f569ed1/image.png">
Sprint icons created by <a href="https://www.freepik.com/free-icon/race_14823113.htm#query=%EB%B3%91%EB%A0%AC%20%EC%8B%A4%ED%96%89&amp;position=1&amp;from_view=search&amp;track=robertav1_2_sidr">Freepik - Flaticon</a>
Stream의 parallel() API를 사용하면 쉽게 병렬 작업을 구현할 수 있습니다. 병렬 스트림은 내부적으로 common ForkJoinPool을 사용하고 디폴트로 CPU 코어 개수만큼의 쓰레드를 생성합니다. CompletableFuture을 사용해서 병렬 작업을 직접 실행시키는 것과 비교해 보면 Join에 신경쓰지 않아도 되기 때문에 코드가 더 단순해 집니다. 
<h2 id="병렬-스트림-예제">병렬 스트림 예제</h2>
예를 들어 다음과 같이 약 1초의 실행시간을 가지는 CPU 코어 개수 만큼의 태스크를 병렬 스트림으로 실행하면 중간연산(mapToObj)이 병렬로 약 1초 안에 동시에 실행되고 Join된 이후 최종 연상(reduce)이 완료되는 것을 확인할 수 있습니다.
<pre><code>@Test
@DisplayName("병렬스트림은 병렬로 중간연산을 수행하고 Join 이후 최종연산을 수행합니다.")
void runParallelAndJoinIntermediateOperation() {
 // Given: 시스템의 CPU 코어 개수만큼의 작업 준비
 int nTasks = Runtime.getRuntime().availableProcessors();
 log.info("The number of CPU Core : {}", nTasks);

 StopWatch stopWatch = new StopWatch();
 stopWatch.start();

 // When: CPU 코어 개수 만큼의 병렬 작업(1초 실행 시간 모의)을 수행
 Integer result = IntStream.rangeClosed(1, nTasks)
 .parallel()
 .mapToObj(i -&gt; {
 log.info("index : {}", i);
 SleepUtils.sleep(1000);
 return i;
 })
 .reduce(0, Integer::sum);
 stopWatch.stop();
 log.info("Total time : {} ms", stopWatch.getTotalTimeMillis());

 // Then: 1초 안에 병렬로 모든 작업이 완료됩니다.
 assertTrue(stopWatch.getTotalTimeMillis() &gt; 1000);
 assertTrue(stopWatch.getTotalTimeMillis() &lt; 1100);
 assertEquals(sumOfSequence(nTasks), result);
}</code></pre>
위 테스트 코드의 실행 결과는 다음과 같습니다. ForJoinPool commonPool에서 쓰레드 7개가 할당되어 각 작업이 병렬로 실행되고 1개의 작업은 main 쓰레드에 의해 처리되는 것을 볼 수 있습니다. 전체 실행 시간은 1.065초 입니다. 
<pre><code>11:52:33.596 [main] INFO tech.java.stream.ParallelTest - The number of CPU Core : 8
11:52:33.618 [main] INFO tech.java.stream.ParallelTest - index : 6
11:52:33.619 [ForkJoinPool.commonPool-worker-1] INFO tech.java.stream.ParallelTest - index : 3
11:52:33.633 [ForkJoinPool.commonPool-worker-2] INFO tech.java.stream.ParallelTest - index : 8
11:52:33.634 [ForkJoinPool.commonPool-worker-5] INFO tech.java.stream.ParallelTest - index : 5
11:52:33.634 [ForkJoinPool.commonPool-worker-4] INFO tech.java.stream.ParallelTest - index : 2
11:52:33.635 [ForkJoinPool.commonPool-worker-3] INFO tech.java.stream.ParallelTest - index : 4
11:52:33.640 [ForkJoinPool.commonPool-worker-6] INFO tech.java.stream.ParallelTest - index : 1
11:52:33.640 [ForkJoinPool.commonPool-worker-7] INFO tech.java.stream.ParallelTest - index : 7
11:52:34.675 [main] INFO tech.java.stream.ParallelTest - Total time : 1065 ms</code></pre>
<h2 id="일반-스트림-예제">일반 스트림 예제</h2>
반면에 일반적인 스트림으로 위 코드를 바꾸어 실행해 보면 중간연산(mapToObj)이 순차적으로 실행되어 전체 작업에 각 작업에 소요되는 시간의 합 만큼(약 8초)의 시간이 소요되는 것을 확인할 수 있습니다.
<pre><code>@Test
@DisplayName("일반스트림은 차례대로 중간연산을 수행한 후 최종연산을 수행합니다.")
void runSequentialIntermediateOperation() {
 // Given: 시스템의 CPU 코어 개수만큼의 작업 준비
 int nTasks = Runtime.getRuntime().availableProcessors();
 log.info("The number of CPU Core : {}", nTasks);

 StopWatch stopWatch = new StopWatch();
 stopWatch.start();

 // When: CPU 코어 개수 만큼의 작업(1초 실행 시간 모의)을 순차적으로 수행
 Integer result = IntStream.rangeClosed(1, nTasks)
 .mapToObj(i -&gt; {
 log.info("index : {}", i);
 SleepUtils.sleep(1000);
 return i;
 })
 .reduce(0, Integer::sum);
 stopWatch.stop();
 log.info("Total time : {} ms", stopWatch.getTotalTimeMillis());

 // Then: 각 작업 실행 시간의 합 만큼의 시간이 소요됨
 assertTrue(stopWatch.getTotalTimeMillis() &gt; nTasks * 1000);
 assertTrue(stopWatch.getTotalTimeMillis() &lt; nTasks * 1000 + 100);
 assertEquals(sumOfSequence(nTasks), result);
}</code></pre>
출력 결과는 다음과 같습니다.
<pre><code>11:51:13.081 [main] INFO tech.java.stream.ParallelTest - The number of CPU Core : 8
11:51:13.098 [main] INFO tech.java.stream.ParallelTest - index : 1
11:51:14.115 [main] INFO tech.java.stream.ParallelTest - index : 2
11:51:15.120 [main] INFO tech.java.stream.ParallelTest - index : 3
11:51:16.134 [main] INFO tech.java.stream.ParallelTest - index : 4
11:51:17.135 [main] INFO tech.java.stream.ParallelTest - index : 5
11:51:18.145 [main] INFO tech.java.stream.ParallelTest - index : 6
11:51:19.146 [main] INFO tech.java.stream.ParallelTest - index : 7
11:51:20.149 [main] INFO tech.java.stream.ParallelTest - index : 8
11:51:21.156 [main] INFO tech.java.stream.ParallelTest - Total time : 8060 ms</code></pre></div></div>

Joo's Work

JooSing

Sprint icons created by Freepik - Flaticon
Stream의 parallel() API를 사용하면 쉽게 병렬 작업을 구현할 수 있습니다. 병렬 스트림은 내부적으로 common ForkJoinPool을 사용하고 디폴트로 CPU 코어 개수만큼의 쓰레드를 생성합니다. CompletableFuture을 사용해서 병렬 작업을 직접 실행시키는 것과 비교해 보면 Join에 신경쓰지 않아도 되기 때문에 코드가 더 단순해 집니다. 
병렬 스트림 예제
예를 들어 다음과 같이 약 1초의 실행시간을 가지는 CPU 코어 개수 만큼의 태스크를 병렬 스트림으로 실행하면 중간연산(mapToObj)이 병렬로 약 1초 안에 동시에 실행되고 Join된 이후 최종 연상(reduce)이 완료되는 것을 확인할 수 있습니다.
@Test
@DisplayName("병렬스트림은 병렬로 중간연산을 수행하고 Join 이후 최종연산을 수행합니다.")
void runParallelAndJoinIntermediateOperation() {
 // Given: 시스템의 CPU 코어 개수만큼의 작업 준비
 int nTasks = Runtime.getRuntime().availableProcessors();
 log.info("The number of CPU Core : {}", nTasks);
 StopWatch stopWatch = new StopWatch();
 stopWatch.start();
 // When: CPU 코어 개수 만큼의 병렬 작업(1초 실행 시간 모의)을 수행
 Integer result = IntStream.rangeClosed(1, nTasks)
 .parallel()
 .mapToObj(i -> {
 log.info("index : {}", i);
 SleepUtils.sleep(1000);
 return i;
 })
 .reduce(0, Integer::sum);
 stopWatch.stop();
 log.info("Total time : {} ms", stopWatch.getTotalTimeMillis());
 // Then: 1초 안에 병렬로 모든 작업이 완료됩니다.
 assertTrue(stopWatch.getTotalTimeMillis() > 1000);
 assertTrue(stopWatch.getTotalTimeMillis() < 1100);
 assertEquals(sumOfSequence(nTasks), result);
}
위 테스트 코드의 실행 결과는 다음과 같습니다. ForJoinPool commonPool에서 쓰레드 7개가 할당되어 각 작업이 병렬로 실행되고 1개의 작업은 main 쓰레드에 의해 처리되는 것을 볼 수 있습니다. 전체 실행 시간은 1.065초 입니다. 
11:52:33.596 [main] INFO tech.java.stream.ParallelTest - The number of CPU Core : 8
11:52:33.618 [main] INFO tech.java.stream.ParallelTest - index : 6
11:52:33.619 [ForkJoinPool.commonPool-worker-1] INFO tech.java.stream.ParallelTest - index : 3
11:52:33.633 [ForkJoinPool.commonPool-worker-2] INFO tech.java.stream.ParallelTest - index : 8
11:52:33.634 [ForkJoinPool.commonPool-worker-5] INFO tech.java.stream.ParallelTest - index : 5
11:52:33.634 [ForkJoinPool.commonPool-worker-4] INFO tech.java.stream.ParallelTest - index : 2
11:52:33.635 [ForkJoinPool.commonPool-worker-3] INFO tech.java.stream.ParallelTest - index : 4
11:52:33.640 [ForkJoinPool.commonPool-worker-6] INFO tech.java.stream.ParallelTest - index : 1
11:52:33.640 [ForkJoinPool.commonPool-worker-7] INFO tech.java.stream.ParallelTest - index : 7
11:52:34.675 [main] INFO tech.java.stream.ParallelTest - Total time : 1065 ms
일반 스트림 예제
반면에 일반적인 스트림으로 위 코드를 바꾸어 실행해 보면 중간연산(mapToObj)이 순차적으로 실행되어 전체 작업에 각 작업에 소요되는 시간의 합 만큼(약 8초)의 시간이 소요되는 것을 확인할 수 있습니다.
@Test
@DisplayName("일반스트림은 차례대로 중간연산을 수행한 후 최종연산을 수행합니다.")
void runSequentialIntermediateOperation() {
 // Given: 시스템의 CPU 코어 개수만큼의 작업 준비
 int nTasks = Runtime.getRuntime().availableProcessors();
 log.info("The number of CPU Core : {}", nTasks);
 StopWatch stopWatch = new StopWatch();
 stopWatch.start();
 // When: CPU 코어 개수 만큼의 작업(1초 실행 시간 모의)을 순차적으로 수행
 Integer result = IntStream.rangeClosed(1, nTasks)
 .mapToObj(i -> {
 log.info("index : {}", i);
 SleepUtils.sleep(1000);
 return i;
 })
 .reduce(0, Integer::sum);
 stopWatch.stop();
 log.info("Total time : {} ms", stopWatch.getTotalTimeMillis());
 // Then: 각 작업 실행 시간의 합 만큼의 시간이 소요됨
 assertTrue(stopWatch.getTotalTimeMillis() > nTasks * 1000);
 assertTrue(stopWatch.getTotalTimeMillis() < nTasks * 1000 + 100);
 assertEquals(sumOfSequence(nTasks), result);
}
출력 결과는 다음과 같습니다.
11:51:13.081 [main] INFO tech.java.stream.ParallelTest - The number of CPU Core : 8
11:51:13.098 [main] INFO tech.java.stream.ParallelTest - index : 1
11:51:14.115 [main] INFO tech.java.stream.ParallelTest - index : 2
11:51:15.120 [main] INFO tech.java.stream.ParallelTest - index : 3
11:51:16.134 [main] INFO tech.java.stream.ParallelTest - index : 4
11:51:17.135 [main] INFO tech.java.stream.ParallelTest - index : 5
11:51:18.145 [main] INFO tech.java.stream.ParallelTest - index : 6
11:51:19.146 [main] INFO tech.java.stream.ParallelTest - index : 7
11:51:20.149 [main] INFO tech.java.stream.ParallelTest - index : 8
11:51:21.156 [main] INFO tech.java.stream.ParallelTest - Total time : 8060 ms

병렬 스트림(Parallel Stream)을 활용한 성능 향상

댓글