[Java] 병렬 스트림 (parallel stream)

병렬 스트림이란?

병렬 스트림이란 각각의 스레드에서 처리할 수 있도록 스트림 요소를 여러 청크로 분할한 스트림이다.

따라서 병렬 스트림을 이용하면 모든 멀티코어 프로세서가 각각의 청크를 처리하도록 할당할 수 있다.

 

예를 들어, 숫자 n을 인수로 받아서 1부터 n까지의 모든 숫자의 합계를 반환하는 메서드를 구현한다했을 때,

public long sequentialSum(long n) {
    return Stream.iterate(1L, i -> i + 1) // 무한 자연수 스티림 생성
        .limit(n) // n개 이하로 제한
        .reduce(0L, Long::sum); // 모든 숫자를 더하는 스트림 리듀싱 연산
 }

위와 같은 코드에 n이 커진다면 이 연산을 병렬로 처리하는 것이 좋을 것이다.

병렬 스트림을 이용하면 병렬 처리와 관련된 문제들을 쉽게 해결할 수 있다.

---

순차 스트림을 병렬 스트림으로 변환

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

위 코드는 리듀싱 연산으로 스트림의 모든 숫자를 더한다.

이전 코드와 다른 점은 스트림이 여러 청크로 분할되어 있다는 것이다.

사실 순차 스트림에 parallel을 호출해도 스트림 자체에는 아무 변화도 일어나지 않는다.

내부적으로는 parallel을 호출하면 이후 연산이 병렬로 수행해야 함을 의미하는 불리언 플래그가 설정된다.

반대로 sequential로 병렬 스트림을 순차 스트림으로 바꿀 수 있다.

stream.parallel()
	  .filter(...)
      .sequential()
      .map(...)
      .parallel()
      .reduce();

parallel과 sequential 두 메서드 중 최종적으로 호출된 메서드가 전체 파이프라인에 영향을 미친다.

위와 같은 경우 마지막 호출은 parallel이므로 파이프라인은 전체적으로 병렬로 실행된다.

 

병렬 스트림에서 사용하는 스레드 풀 설정
병렬 스트림은 내부적으로 ForkJoinPool을 사용한다. 기본적으로 ForkJoinPool은 프로세서 수, 즉 Runtime.getRuntime(), availableProcessors()가 반환하는 값에 상응하는 스레드를 갖는다.

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

위 예제는 전역 설정 코드이므로 이후의 모든 병렬 스트림 연산에 영향을 준다.
현재는 하나의 병렬 스트림에 사용할 수 있는 특정한 값을 지정할 수 없다.
일반적으로 기기의 프로세서 수와 같으므로 특별한 이유가 없다면 ForkJoinPool의 기본값을 그대로 사용할 것을 권장한다.

---

성능 비교

for 루프를 사용하는 경우와, 순차적 스트림을 사용하는 경우, 병렬 스트림을 사용하는 경우를 비교해 보자.

public long iterativeSum(long n) {
      long result = 0;
      for (long i = 1L; i<=n; i++) {
        result += i;
      }
      return result;
}

for 루프를 사용해 반복하는 방법이 더 저수준으로 동작할 뿐 아니라 특히 기본값을 박싱 하거나 언박싱할 필요가 없으므로 더 빠를 것이라 예상할 수 있다. 

위의 결과가 for 루프를 사용한 결과고 아래가 순차적 스트림을 사용한 결과이다.

예상대로 순차적 스트림을 사용하는 버전에 비해 거의 4배가 빠르다는 것을 확인할 수 있다.

 

병렬 스트림을 사용하는 버전은 어떨까?

병렬 버전이 순차 버전에 비해 다섯 배나 느린 실망스러운 결과가 나온다.

왜 이러한 결과가 나올까? 다음의 두 가지 문제를 발견할 수 있다.

• 반복 결과로 박싱 된 객체가 만들어지므로 숫자를 더하려면 언박싱을 해야 한다.
• 반복 작업은 병렬로 수행할 수 있는 독립 단위로 나누기가 어렵다.

두 번째 문제는 iterate연산은 이전 연산 결과에 따라 다음 함수의 입력이 달라지기 때문에 청크로 분할하기 어렵다.

리듀싱 과정을 시작하는 지점에 전체 수자 리스트가 준비되지 않았으므로 스트림을 병렬로 처리할 수 있도록 청크로 분할할 수 없다.

스트림이 병렬로 처리되도록 지시했고 각각의 합계가 다른 스레드에서 수행되었지만 결국 순차처리 방식과 크게 다른 점이 없으므로 스레드를 할당하는 오버헤드만 증가하게 된다.

 

이처럼 병렬 프로그래밍은 까다롭고 때로는 이해하기 어려운 함정이 숨어 있다.

병렬 프로그래밍을 오용하면 오히려 전체 프로그램의 성능이 더 나빠질 수도 있다.

 

위와 같은 문제는 LongStream.rangeClosed라는 메서드를 통해 해결할 수 있다.

이 메서드는 iterate에 비해 다음과 같은 장점을 제공한다.

• 기본형 long을 직접 사용하므로 박싱과 언박싱 오버헤드가 사라진다.
• 쉽게 청크로 분할할 수 있는 숫자 범위를 생산한다.

public long rangedSum(long n) {
    return LongStream.rangeClosed(1, n)
        .reduce(0L, Long::sum);
}

기존의 iterate를 사용한 버전보다 스트림 처리 속도가 더 빠르다.

특화되지 않은 스트림을 처리할 때는 오토박싱, 언박싱 등의 오버헤드를 수반하기 때문이다.

상황에 따라서는 어떤 알고리즘을 병렬화하는 것보다 적절한 자료구조를 선택하는 것이 더 중요하다는 사실을 단적으로 보여준다.

 

이제 병렬 스트림을 적용하면 다음과 같다.

public long parallelRangedSum(long n) {
    return LongStream.rangeClosed(1, n)
        .parallel()
        .reduce(0L, Long::sum);
}

드디어 순차 실행보다 빠른 성능을 갖는 병렬 리듀싱을 만들었다.

 

병렬화가 완전 공짜는 아니라는 사실을 기억하자.

병렬화를 이용하려면 스트림을 재귀적으로 분할해야 하고, 각 서브 스트림을 서로 다른 스레드의 리듀싱 연산으로 할당하고, 이들 결과를 하나의 값으로 합쳐야 한다. 멀티코어 간의 데이터 이동은 우리 생각보다 비싸다. 따라서 코어 간의 데이터 전송 시간보다 훨씬 오래 걸리는 작업만 병렬로 다른 코어에서 수행하는 것이 바람직하다.

---

병렬 스트림의 잘못된 사용

public long sideEffectSum(long n) {
    Accumulator accumulator = new Accumulator();
    LongStream.rangeClosed(1,n).forEach(accumulator::add);
    return accumulator.total;
}
  
public class Accumulator {
    public long total = 0;
    public void add (long value) {total += value;}
}

위 코드는 n까지의 자연수를 더하면서 공유된 누적자(total)를 바꾸는 프로그램이다.

위 코드는 본질적으로 순차 실행할 수 있도록 구현되어 있으므로 병렬로 실행하면 참사가 일어난다.

특히 total을 접근할 때마다 데이터 레이스 문제가 일어난다. 동기화로 문제를 해결하다 보면 결국 병렬화라는 특성이 없어져 버릴 것이다.

 

위 코드는 메서드의 성능은 둘째 치고, 올바른 결괏값이 나오지 않는다.

여러 스레드에서 동시에 누적자, 즉 total += value를 실행하면서 이런 문제가 발생한다.

결국 여러 스레드에서 공유하는 객체의 상태를 바꾸는 forEach 블록 내부에서 add 메서드를 호출하면서 이 같은 문제가 발생한다.

이 예제처럼 병렬 스트림을 사용했을 때 이상한 결과에 당황하지 않으려면 상태 공유에 따른 부작용을 피해야 한다.

---

병렬 스트림 사용 기준

양을 기준으로 병렬 스트림 사용을 결정하는 것은 적절하지 않다.

• 확신이 서지 않으면 직접 측정하라. 순차 스트림을 병렬 스트림으로 쉽게 바꿀 수 있다.
• 박싱을 주의하라. 오토박싱 와 언박싱은 성능을 크게 저하시킬 수 있는 요소다. 기본형 특화 스트림을 통해 박싱 동작을 피할 수 있다. (IntStream, LongStream, DoubleStream)
• 순차 스트림보다 병렬 스트림에서 성능이 떨어지는 연산이 있다. 특히 limit이나 findFirst처럼 요소의 순선에 의존하는 연산을 병렬 스트림에서 수행하려면 비싼 비용을 치러야 한다.
• 스트림에서 수행하는 전체 파이프라인 연산 비용을 고려하라.
• 소량의 데이터에서는 병렬 스트림이 도움 되지 않는다.
• 스트림을 구성하는 자료구조가 적절한지 확인하라. 예를 들어 ArrayList를 LinkedList보다 효율적으로 분할할 수 있다.
• 스트림의 특성과 파이프라인의 중간 연산이 스트림의 특성을 어떻게 바꾸는지에 따라 분해 과정의 성능이 달라질 수 있다.
• 최종 연산의 병합 과정(예를 들면 Collector의 combiner 메서드) 비용을 살펴보라. 별합 과정의 비용이 비싸다면 병렬 스트림으로 얻은 성능의 이익이 서브스트림의 부분 결과를 합치는 과정에서 상쇄될 수 있다.

---

참고 자료 : https://ebook-product.kyobobook.co.kr/dig/epd/ebook/E000002942391

모던 자바 인 액션 | 라울-게이브리얼 우르마 | 한빛미디어- 교보ebook