[Java] parallelStream() 메서드 사용시 예상되는 문제점 및 해결방안 [2]

앞선 포스팅에서 자바 parallelStream()사용시 예상되는 문제점과 그 이유에 대해 알아보았는데요. 이번 포스팅에서는 언급된 문제점들에 대한 해결방안을 간단한 예시코드를 통해서 설명드리겠습니다.

 

제약사항 및 예상 문제점 관련하여 정리한 포스팅은 아래 링크를 참고해주세요.

 

[Java] parallelStream() 메서드 사용시 예상되는 문제점 및 해결방안 [1]

 

자바 parallelStream() 메서드 예상문제점 해결방안

 

 1. 동기화 기법

1-1. synchronized 키워드 사용하여 

numbers.parallelStream().forEach(num -> {
            synchronized (numbers) {
                System.out.println(Thread.currentThread().getName() + " : " + num);
                numbers.remove(num);
            }
        });

 요소를 순회할 때, synchronized 키워드를 블록으로 사용해서 동기화 시켜주면 됩니다. 하지만 동기화를 하게되면 parallelStream() 메서드 병렬 처리시 성능이 떨어지게 됩니다.

 

 위 예시에서는 설명을 위해 간단한 예시를 사용했지만, 실무에서는 처리량이 많은 상황에서 동시에 많은 작업을 처리하는 경우가 위해 parallelStream을 사용합니다. 위와 같은 문제가 예상되는 경우에는 synchronized 키워드를 사용하여 다소 성능이 느려지더라도 안정성을 위해 사용하는 것이 좋습니다. 여러가지 상황에 따라 알맞게 사용하는게 중요합니다.

 

 

1-2. Lock 사용하기

import java.util.concurrent.locks.ReentrantLock;

...

ReentrantLock lock = new ReentrantLock();
        numbers.parallelStream().forEach(num -> {
            lock.lock();
            try {
                System.out.println(Thread.currentThread().getName() + " : " + num);
                numbers.remove(num);
            } finally {
                lock.unlock();
            }
        });

ReentranLock 클래스는 대기 시간을 설정할 수 있기 때문에 synchronized 키워드 보다 디테일한 설정을 할 수 있다는 장점이 있습니다. parallelStream은 내부적으로 데이터를 처리하기에 알맞은 스레드 풀을 생성하여 병렬처리를 수행하는데, ReentranLock은 스레드 풀 생성을 방해하여 병렬처리하는 의미를 떨어트릴 수 있습니다.

 

 2. Thread-Safe한 자료구조 사용하기

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

public class Main {
    public static void main(String[] args) {
        List<Integer> numbers = Collections.synchronizedList(new ArrayList<>());
        for (int i = 0; i < 10; i++) {
            numbers.add(i);
        }

        numbers.parallelStream().forEach(num -> {
            System.out.println(Thread.currentThread().getName() + " : " + num);
            numbers.remove(num);
        });
    }
}

ArrayList를 동기화된 리스트로 만들어 사용하여 해결하는 방법입니다.

 

 3. Stream.collect() 메서드 이용하여 결과를 새로운 리스트에 저장하기

import java.util.ArrayList;
import java.util.List;

public class Main {
    public static void main(String[] args) {
        List<Integer> numbers = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            numbers.add(i);
        }

        List<Integer> result = numbers.parallelStream()
                .collect(ArrayList::new, (list, num) -> {
                    System.out.println(Thread.currentThread().getName() + " : " + num);
                    list.add(num);
                }, ArrayList::addAll);
        System.out.println(result);
    }
}

각 스레드에서 처리한 결과를 새로운 리스트에 추가해서 스레드 간의 경쟁 문제를 해결 할 수 있습니다. Thread-safe한 자료구조를 사용하는 방법 보다는 성능이 떨어질 수 있습니다.