스트림 병렬 처리 이해

• 스트림 API에서는 parallelStream메서드를 이용해서 스트림객체를 생성하는것만으로도 병렬처리가 된다.

import java.util.Arrays;
import java.util.List;

public class ParallelReduceMinMax {

	public static void main(final String[] args) {
		final List<Integer> intList = Arrays.asList(4, 2, 8, 1, 9, 6, 7, 3, 5);

		// 최대값 구하기
		final Integer max = intList.parallelStream().reduce(1, Integer::max);
		System.out.println("최대값 = " + max);

		// 최소값 구하기
		final Integer min = intList.parallelStream().reduce(1, Integer::min);
		System.out.println("최소값 = " + min);
	}
}

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 병렬 스트림 API 사용 예
 */
public class InsideParallelStream {

	public static void main(final String[] args) {
		final List<Integer> intList = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);

		final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

		// 스트림 내부의 스레드 값을 구함
		intList.parallelStream().forEach(value -> {
			// 현재 스레드 이름을 구한다.
			final String threadName = Thread.currentThread().getName();

			// 스레드 이름과 데이터 값을 출력한다.
			final LocalDateTime currentTime = LocalDateTime.now();
			System.out.printf(currentTime.format(formatter) + " -> Thread Name : %s, Stream Value : %s\\n", threadName, value);

			// 시간 확인을 위해 2초간 sleep한다.
			try {
				TimeUnit.SECONDS.sleep(2);
			} catch (final InterruptedException ignore) {
			}
		});
	}
}

2022-07-02 22:48:30 -> Thread Name : ForkJoinPool.commonPool-worker-27, Stream Value : 5
2022-07-02 22:48:30 -> Thread Name : ForkJoinPool.commonPool-worker-5, Stream Value : 9
2022-07-02 22:48:30 -> Thread Name : ForkJoinPool.commonPool-worker-3, Stream Value : 4
2022-07-02 22:48:30 -> Thread Name : ForkJoinPool.commonPool-worker-31, Stream Value : 8
2022-07-02 22:48:30 -> Thread Name : ForkJoinPool.commonPool-worker-9, Stream Value : 1
2022-07-02 22:48:30 -> Thread Name : ForkJoinPool.commonPool-worker-19, Stream Value : 3
2022-07-02 22:48:30 -> Thread Name : ForkJoinPool.commonPool-worker-17, Stream Value : 6
2022-07-02 22:48:30 -> Thread Name : ForkJoinPool.commonPool-worker-13, Stream Value : 10
2022-07-02 22:48:30 -> Thread Name : main, Stream Value : 7
2022-07-02 22:48:30 -> Thread Name : ForkJoinPool.commonPool-worker-23, Stream Value : 2

병렬처리의 특징

• 병렬 처리이기 때문에 데이터의 순서가 1부터 10까지 차례대로 나오지 않고 섞여서 출력된다. 이러한 규칙때문에 실행할 때마다 그 결과가 다르게 나온다.
  • 어떤 데이터가 먼저 처리될 지 알 수 없다.
  • 따라서 스트림 데이처 처리를 할 때 순서가 중요하다면 병렬 처리를 해서는 안된다.
• PC나 서버의 코어 수에 따라 스레드가 생성된다.
  • paralleStream 메서드를 사용하면 내부적으로 PC 혹은 서버의 코어 수를 계산한 다음 자동으로 스레드를 생성한다.
  • main 스레드는 스트림을 처리하기 위한 기본 스레드를 의미하고, 이 스레드가 n개의 ForkJoinPool의 스레드를 생성한다.
  • 이렇게 코어 수 기반의 스레드 생성은 스트림에서 제어한 것이 아니라, 컨커런트 API의 ForkJoinPool의 기본값이며, 여기에 영향을 받은것이다.

스레드 개수 제어

ForkJoinPool의 기본 스레드 값을 변경한다.

• JVM 옵션 설정

  • 자바 가상머신 전체에 영향을 미침.

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

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;

/**
 * 병렬 스트림 API 사용 예
 */
public class InsideParallelStream2 {

	public static void main(final String[] args) {
		final List<Integer> intList = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);

		final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

		// 스레드 개수 2개로 설정
		System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "2");
		System.out.printf("## Thread Pool Size: %s\\n", ForkJoinPool.getCommonPoolParallelism());

		// 스트림 내부의 스레드 값을 구함
		intList.parallelStream().forEach(value -> {
			// 현재 스레드 이름을 구한다.
			final String threadName = Thread.currentThread().getName();

			// 스레드 이름과 데이터 값을 출력한다.
			final LocalDateTime currentTime = LocalDateTime.now();
			System.out.printf(currentTime.format(formatter) + " -> Thread Name : %s, Stream Value : %s\\n", threadName, value);

			// 시간 확인을 위해 2초간 sleep한다.
			try {
				TimeUnit.SECONDS.sleep(2);
			} catch (final InterruptedException ignore) {
			}
		});
	}
}

## Thread Pool Size: 2
2022-07-02 22:56:59 -> Thread Name : main, Stream Value : 7
2022-07-02 22:56:59 -> Thread Name : ForkJoinPool.commonPool-worker-3, Stream Value : 3
2022-07-02 22:56:59 -> Thread Name : ForkJoinPool.commonPool-worker-1, Stream Value : 9
2022-07-02 22:57:01 -> Thread Name : main, Stream Value : 6
2022-07-02 22:57:01 -> Thread Name : ForkJoinPool.commonPool-worker-1, Stream Value : 10
2022-07-02 22:57:01 -> Thread Name : ForkJoinPool.commonPool-worker-3, Stream Value : 5
2022-07-02 22:57:03 -> Thread Name : ForkJoinPool.commonPool-worker-3, Stream Value : 4
2022-07-02 22:57:03 -> Thread Name : ForkJoinPool.commonPool-worker-1, Stream Value : 8
2022-07-02 22:57:05 -> Thread Name : ForkJoinPool.commonPool-worker-3, Stream Value : 2
2022-07-02 22:57:05 -> Thread Name : ForkJoinPool.commonPool-worker-1, Stream Value : 1

• 주의사항
  • 설정한 스레드 값(ForkJoinPool)에 하나를 더한 스레드(main)가 총 스레드 개수다.
    • 총 2개의 스레드로 실행하려면 값을 1로 설정해야 한다.
  • ForkJoinPool의 기본값을 변경하는 것이기 때문에, 자바 가상 머신 전체에 영향을 미친다.
    • 특정한 프로세스 혹은 스레드에 대해서만 변경할 수 없으며, 가상 머신이 종료될 때까지 설정이 유효하다.

ForkJoinPool이 아닌 다른 스레드 풀을 사용한다.

• ForkJoinPool의 commonPool이 아닌, 사용자 정의 풀을 사용한다.