[Java] parallelStream 완전분석 (feat. fork/join framework)

 

단순하게 stream은 순차처리, parallelStream은 병렬 처리가 된다.

그럼 무조건 parallelStream을 쓰면 빠를 텐데?

하지만 그러면 안 되는 이유가 있을 것만 같은 느낌적인 느낌

parallelStream도 분명 뭔가가 불편하거나 힘들었기 때문에 나왔을 것이다.

파헤쳐보자.

 

자바 7 이전의 컬렉션 데이터 병렬 처리 방식

데이터를 서브 파트로 분할 후, 분할된 서브 파트에 따라서 각각의 스레드로 할당한다. 각각의 할당된 스레드에서 경쟁상태가 발생하지 않도록 적절한 동기화를 해줘야 하며, 마지막으로 부분 결과를 다시 합쳐야 한다.

 

자바 7 이후의 컬렉션 데이터 병렬 처리 방식

자바 7부터 포크/조인 프레임워크 기능을 제공한다. 또한 자바 8에서는 스트림을 사용함으로써 병렬 스트림 처리를 할 수 있다.

포크/조인 프레임워크라는 게 있네?

뭔지는 잘 모르겠지만, 프레임워크이니 복잡한 과정을 줄여줄것만 같다.

 

1. 포크/조인 프레임워크

병렬화할 수 있는 작업을 재귀적으로 작은 작업으로 분할 후, 서브 태스크의 각각의 결과를 합쳐서 최종 결과를 만든다. 내부적으로 ForkJoinPool이라는 스레드 풀을 사용하며, ExecutorService 인터페이스를 구현한다.

스레드 풀을 이용하기 위해 RecursiveTask의 서브 클래스를 만들고 추상 메소드 compute를 구현해줘야 한다.

protected abstract R compute();

이 메소드는 태스크를 서브 태스크로 분할하는 로직과 더 이상 분할할 수 없을 때 개별 서브 태스크의 결과를 합치는 형식이다. 의사 코드로 나타내면 아래와 같다.

if(태스크가 충분히 작거나 더 이상 분할할 수 없으면){
        순차계산
    }else{
        태스크를 두 서브태스크로 분할
        태스크가 다시 서브태스크로 분할되도록 이 메소드를 재귀적으로 호출
        모든 서브태스크의 연산이 완료될 때까지 기다림
        각 서브태스크의 결과를 합침
    }

 

분할 정복 알고리즘을 기반으로 한다.

음 그럼 포크/조인 프레임워크를 사용해서 병렬로 처리하면 다 되는 건가??

 

주의사항

• task.join을 호출하면 task의 결과가 준비될 때까지 호출자를 블록 시키기 때문에, 서브 태스크가 모두 시작된 다음 join을 호출해야 한다.
• 그렇지 않으면 각각의 서브 태스크가 다른 태스크를 기다리느라 오히려 더 느린 프로그램이 될 수 있다.
• 포크/조인 프레임워크를 사용했다고 해서 무조건 빠른 것은 아니다. 여러 독립적인 서브 태스크로 분할할 수 있어야 하고, 각 서브 태스크의 실행시간은 새로운 태스크를 forking 하는 데 드는 시간보다 길어야 한다.

분할하는 건 알겠는데 그럼 어디까지 fork 해야 하지??

 

작업 훔치기

포크/조인 프레임워크에서는 작업 훔치기라는 기법을 사용한다. 이 기법을 통해 ForkJoinPool의 모든 스레드를 거의 공정하게 분할한다.

각각의 스레드는 자신에게 할당된 태스크를 포함하는 이중 연결 리스트를 참조하면서 작업이 끝날 때마다 큐의 헤드에서 다른 태스크를 가져와서 작업을 처리한다.

이때 한 스레드가 다른 스레드보다 할당된 태스크를 빨리 처리했다면, 유휴 상태로 바뀌는 것이 아닌 다른 스레드의 큐의 꼬리에서 작업을 훔쳐온다. 모든 큐가 빌 때까지 이 과정을 반복하기 때문에, 태스크의 크기를 작게 나누어야 작업 스레드 간 작업 부하를 비슷하게 유지할 수 있다.

태스크를 작게 나누는 게 중요!

task가 없으면 다른 작업 중인 task를 가져와 처리함으로써 CPU 자원이 놀지 않고 최적의 성능을 낼 수 있다.

 

스트림에서의 작업 분할

자바 8은 Spliterator 인터페이스를 제공한다. 병렬 작업에 특화되어있는 분할 반복자이다.

Spliterator 인터페이스

public interface Spliterator<T> {
    boolean tryAdvance(Consumer<? super T> action); //iterator 처럼 순차 탐색하면서 탐색해야 하는 요소가 있다면 true 리턴
    Spliterator<T> trySplit(); //T는 spliterator에서 탐색하는 요소의 형식
    long estimateSize();    // 탐색해야하는 요소의 수
    int characteristics();
}

 

동작 방식

특성상 작업을 균등하게 처리하기 위해 Spliterator의 trySplit()을 사용하는데, trySplit의 결과가 null이 나올 때까지 재귀의 형태로 계속 반복된다.

나누어지는 작업에 대한 비용이 높지 않아야 순차적 방식보다 효율적으로 이루어질 수 있다.

 

2. parallelStream

컬렉션에 parallelStream을 호출하기만 하면 병렬 스트림이 생성된다.

각각의 스레드에서 처리할 수 있도록 스트림 요소를 여러 chunk 단위로 분할한 스트림이다. 기본적으로 위에서 언급한 ForkJoinPool을 사용한다.

또한, Runtime.getRuntime(). availableProcessors()가 반환하는 만큼의 스레드를 갖는대 보통 기기의 프로세서 수와 같다.

 

예제) 1부터 n까지 모든 숫자의 합계 구하기

public long sequentialSum(long n){
    return Stream.iterate(1L, i -> i+1) //무한 자연수 스트림
                             .limit(n)                            //개수 제한
                             .reduce(0L, Long::sum); //모든 숫자 더하기
}

만약 n이 엄청 크다면, 병렬로 처리하는 게 당연히 좋을 것이다.

 

예제) 1부터 n까지 모든 숫자의 합계 구하기 (병렬)

public long parallelSum(long n){
    return Stream.iterate(1L, i -> i+1) 
                             .limit(n)                            
                             .parallel() //스트림을 병렬 스트림으로 변환
                             .reduce(0L, Long::sum);
}

단순히 parallel 만 추가해주면 병렬 스트림 처리가 끝난다.

단순해 보이지만, 병렬화를 이용하기 위해서는 스트림을 재귀적으로 분할해야 하고, 각 서브 스트림을 서로 다른 스레드의 리듀싱 연산으로 할당하고, 결과를 하나의 값으로 합쳐야 한다.

또한 내부적으로 공유된 가변 상태를 가지지 않아야 한다. -> 람다 스트림을 사용하는 이유 (=불변 보장)

• 변경 가능한 여부를 가진다는 것이 상당이 위험하다.

 

ParallelStream 사용 시 주의사항

• 성능에 대한 확신이 없는 경우 자바 마이크로 벤치마크 하니스(JMH)를 통해 성능을 직접 측정할 수 있다.
• 박싱을 주의하자.
• 박싱은 성능을 크게 저하시킬 수 있는 요소이기 때문에 이를 방지하기 위해 기본형 특화 스트림(IntStream, LongStream, DoubleStream)을 제공한다.
• 순서 연산에 유의하자. 순서가 상관없는 findAny 같은 경우 병렬 처리가 빠르다.
• limit이나 findFirtst처럼 요소의 순서에 의존하는 연산의 경우 비싼 비용을 요구한다.
• 스트림에서 수행하는 전체 파이프라인 연산 비용을 고려하자.
• 처리할 요소 수를 N, 처리 비용을 Q라고하면 전체 비용은 N * Q 라고 할 수 있는데, Q가 높아진다면 병렬 스트림으로 성능 개선 가능성이 있음을 의미한다.
• 소량의 데이터에서는 병렬 처리가 도움이 되지 않는다.
• 자료구조가 적절한지 확인하자.
• LinkedList는 분할 하기 위해서 모든 요소를 탐색해야 하지만, ArrayList는 탐색하지 않아도 분해할 수 있다. 커스텀 Spliterator를 구현해 분해를 제어할 수 있다.
• 스트림의 특성과 파이프라인의 중간 연산에 따라 성능이 달라진다.
• SIZED 스트림의 경우 정확히 같은 크기로 분할이 가능하지만, filter 연산은 스트림의 길이를 예측할 수 없어 효과적이지 않다.
• 최종 연산의 병합 과정 비용을 살펴보자.
• 병합 과정의 비용이 비싸다면, 병렬 스트림으로 얻은 이익이 상쇄되고 만다.

 

자료구조에 따른 분해 성능

자료구조	분해 성능
ArrayList	매우 좋음
LinkedList	나쁨
IntStream.range	매우 좋음
Stream.iterate	나쁨
HashSet	좋음
TreeSet	좋음

 

결론

병렬 처리라는 게 결국 여러 스레드가 작업을 분할받아, 처리한 결과를 다시 합치는 일이기 때문에 분할 과정이 매우 중요하고, 분할이 중요하기에 분할하기 적합한지에 대한 여부 또한 중요하다.

항상 병렬 처리가 빠르다고 보장할 수 없기 때문에 충분한 성능 테스트 후 도입하는 게 좋아 보인다.

 

참고

사진출처 : https://livebook.manning.com/concept/java

모던 자바 인 액션