[Java] 병렬 데이터 처리 "병렬 스트림"

"병렬 스트림" 은 데이터 처리를 병렬화하여 성능을 향상 시킬 수 있는 기능이다.

병렬 스트림은 스트림 API의 한 종류로,데이터 처리 작업을 병렬적으로 처리하는데 사용되는 기능이다. 

즉, 스트림의 요소들을 병렬적으로 처리함으로써 다중코어(CPU 코어) 시스템에서 병렬 처리 성능을 극대화할 수 있다. 대량의 데이터를 처리할 때 특히 유용하며, 병렬 처리를 활용하면 단일 스레드 처리보다 훨씬 빠른 속도로 데이터를 처리할 수 있다.

 

병렬 스트림은 내부적으로 Fork/Join 프레임워크를 사용하여 작동한다. 

Fork/Join 프레임워크는 작업을 작은 작업들로 분할하고, 이를 병렬로 처리한 후 결과를 합치는 방식으로 동작한다. 

이러한 방식으로 병렬 스트림은 작업을 더 작은 작업 단위로 나누어 멀티코어 프로세서에서 병렬로 처리할 수 있다.

병렬 스트림 사용하는 방법

기존의 스트림을 병렬 스트림으로 변환하기 위해 parallel() 메소드를 호출하여 사용하게 된다. 

이렇게 하면 내부적으로 스트림의 처리가 병렬로 이루어지게 된다. 

또한 반대로 순차적으로 처리하고 싶다면 병렬 처리가 끝난 후에 sequential() 메소드를 사용하면 된다.

List<String> myList = Arrays.asList("apple", "banana", "orange", "grape", "strawberry");

//병렬 스트림으로 변환하여 요소를 대문자로 출력한다.
myList.parallelStream().map(String::toUpperCase).forEach(System.out::println);

 

ParallelStream

Java 8에서 등장한 Stream 은 병렬 처리를 쉽게 할 수 있도록 메소드를 제공한다. 

이에 개발자가 직접 스레드를 생성하거나 관리할 필요가 없이 parallel(), parallelStream() 메소드를 호출함을 통해 알아서 ForkJoinFramework 관리 방식을 이용해 작업을 분할하고 병렬적으로 처리할 수 있도록 지원한다.

 

Fork / Join Framework 

그렇다면 Fork / Join Framework는 무엇일까?
이는 다양한 작업들을 분할 가능한 만큼 쪼개고, 이 쪼개진 작업들을 Work Thread를 통해 작업 후 결과를 합치는 과정으로 결과를 만든다. 

즉, 분할 정복 (Divide and Conquer) 알고리즘과 비슷한 작업이며, Fork 를 통해 Task를 분담하고 Join을 통해 결과를 합치는 과정이라고 할 수 있다.

 

Fork / Join Framework의 Work Thread의 수는 해당 애플리케이션이 구동되는 서버의 스펙에 따라 바뀐다.

Runtime.getRuntime().availableProcessors()으로 JVM에서 이용 가능한 CPU Core 개수를 확인할 수 있으며, 

스레드가 N개 생성되었을 때, 하나는 메인 스레드로 스트림을 처리하는 기본 스레드와 나머지 N-1개의 스레드를 ForkJoinPool 스레드라고 한다.

---

기본 스트림 vs 병렬 스트림 예제

스트림 예제)

public static void main(String[] args) {
	
    List<Integer>numbers = Arrays.asList(1,2,3,4); //Integer타입의 리스트 생성
    
    long start = System.currentTimeMillis(); //현재 시간 기록
    
    //numbers 리스트의 각 요소에 대해 작업을 수행한다.
    numbers.forEach(number -> {
    	try {
        	Thread.sleep(3000); //현재 스레드 3초 일시 중지
            System.out.println(number + ": " + Thread.currentThread().getName());
        } catch (InterruptedException e) {}
    });
    
    //작업이 완료된 후의 시간 계산
    long duration = (System.currentTimeMillis() - start);
    //밀리초는 초로 변환한다.
    double seconds = duration / 1000.0;
    
    System.out.printf("Done in %.2f sec\n", seconds);
}

 

//코드 실행 결과
1: main
2: main
3: main
4: main
Done in 12.05 sec

 

병렬 스트림 예제)

public static void main(String[] args) {   

		List<Integer> numbers = Arrays.asList(1, 2, 3, 4);     
		
        long start = System.currentTimeMillis();    
		numbers.parallelStream().forEach(number -> {        
			try {            
            	Thread.sleep(3000);            
				System.out.println(number + ": " + Thread.currentThread().getName());        
			} catch (InterruptedException e) {}    
       	});    
		long duration = (System.currentTimeMillis() - start);    
		double seconds = duration / 1000.0;     
		
        System.out.printf("Done in %.2f sec\n", seconds);
}

//코드 실행 결과
3: main
1: ForkJoinPool.commonPool-worker-2
2: ForkJoinPool.commonPool-worker-1
4: ForkJoinPool.commonPool-worker-3
Done in 3.04 sec

---

병렬 스트림 사용 시 주의할 점 

1. 병렬 처리는 오버헤드가 발생할 수 있기 때문에 작업이 충분히 크고, 연산량이 많을 때에만 사용하는것이 좋다. 

2. 스트림의 요소가 병렬 처리에 안전해야 한다. 

     ex. 스트림의 연산이 상태를 변경하는 경우(멤버 변수를 수정하는 경우) 등에 예기치 않은 결과가 발생할 수 있다. 

3. 병렬 스트림을 사용할 때는 성능 테스트와 함께 사용하는 것이 좋다. 

실제로 병렬 처리가 성능을 향상시키는지 여부를 확인하고 최적의 결과를 얻기 위해 스트림의 크기와 작업의 복잡성에 맞게 조정해야 한다.