[Effective Java] 아이템48 - 스트림 병렬화는 주의해서 적용하라

자바 8 버전 이후로 parallel 메서드만 호출하면 파이프라인을 병렬 실행할 수 있는 스트림을 지원하고 있다.

이처럼 동시성 프로그램을 작성하기 쉬워졌지만, 프로그램 작성자는 그에대한 안전성과 응답가능 상태를 유지하기 위해 노력해야 한다.

병렬 스트림 파이프라인 프로그래밍도 예외는 아니다.

public static void main(String[] args) {
 primes().map(p -> TWO.pow(p.intValueExact()).subtract(ONE))
   .filter(mersenne -> mersenne.isProbablePrime(50))
   .limit(20)
   .forEach(System.out::println);
 
}  
static Stream<BigInteger> primes(){
 return Stream.iterate(TWO, BigInteger::nextProbablePrime);
}

위 메르센 소수를 생성하는 프로그램의 성능을 높이기 위해 스트림 파이프라인의  parrellel()을 호출하게 되면 CPU 사용율이 높아진 채 프로그램은 종료되지 않을 가능성이 높다.

스트림 라이브러리가 이 파이프라인을 병렬화하는 방법을 찾아내지 못했기 때문이다. 문제가 무엇일까?

데이터 소스가 Stream.interate거나 중간 연산으로 limit을 쓰면 파이프라인 병렬화로 성능개선을 기대할 수 없다. 

파이프라인 병렬화는 limit을 다룰때 CPU 코어가 남으면 원소 몇개를 더 처리한 후 제한된 갯수 이후의 것을 버리는데 문제는 새로운 메르센 소수를 찾는데 그전에 소요된 시간에 두배가 걸린다는 것이다(2^n) . 때문에 자동 병렬화 알고리즘이 제대로 작동할 수 없게 되는 것이다.

위의 예시처럼 Stream을 사용한다고 마구잡이로 병렬화하게 되면 성능이 나빠질 뿐만 아니라 오작동할 수 있다.

결과가 잘못되거나 오작동하는 것을 안전 실패(safety failure)라 하는데 이는 병렬화한 파이프라인의 mapper, fiilers 뿐만 아니라 직접 제공한 객체의 온전한 동작을 보장하지 못한다.

이에 Stream 명세에는 사용되는 함수 객체에 대한 규약이 정의되어있다. 

예를 들어 reduce연산에 건내지는 accumulator와 combiner 함수는 반드시 다음을 따라야한다.

• 결합법칙 만족(associative)
• 간섭받지 않음(non-interfering)
• 상태를 같지 않음(stateless)

이중 하나라도 어기면 순차적일때는 몰라도 병렬로 수행하게되면 결과가 엉망이 될 것이다.

따라서 위의 프로그램이 설령 완료되더라도 출력된 소수의 순서가 올바르지 않을 수 있다. 이를 해결하려면 forEach 대신 forEachOrdered로 바꿔주면 된다. 이는 병렬 스트림들을 순회하며 소수를 발견한 순서대로 출력되도록 보장해 줄 것이다.

 

병렬화가 잘 되어도 파이프라인이 수행하는 작업이 병렬화에 드는 추가비용을 상쇄하지 못하면 성능 향상은 미미하다.

조건에 잘 맞으면 parallel 메서드 호출로 코어수에 비례하는 성능 향상을 기대할 수 있지만 그 반대라면 잘못된 파이프 하나가 시스템 전체에 악영향을 끼칠 수 있다.

스트림 병렬화는 오직 성능 최적화 수단이므로 변경 전후로 성능을 테스트하여 병렬화할 가치가 있는지 확인해야 한다. 보통은 스트림 안에 원소 수와 원소당 수행되는 코드 줄 수를 곱했을 때 수십만 이상이어야 성능 향상을 기대할 수 있다.

 

병렬화를 통해 성능향상을 기대해도 좋을 때

스트림의 소스가 ArrayList, HashMap, HashSet, ConcurrentHashMap의 인스턴스이거나 배열, int 범위, long 범위

이 자료 구조들은 모두 데이터를 원하는 크기로 정확하고 쉽게 나눌 수 있고, 원소들을 순차적으로 실행할 때 참조 지역성이 뛰어나다는 것이다.

데이터를 쉽고 정확하게 나눌 수 있으면 일을 다수의 스레드에 분배하기 쉬워지기 때문에 병렬화에 적합하다. 이는 Spliterator가 담당한다.이웃한 원소들의 참조들이 메모리에 연속해서 저장되어 있으면 참조 지역성이 뛰어나다고 하는데 이는 캐시 히트율을 높이기 때문에 다량의 데이터를 처리하는 벌크연산을 병렬화할 때 중요하게 작용한다.

 

종단연산의 연산방식이 축소(reduction)일 때

종단 연산에서 수행하는 작업량이 파이프라인 전체 작업에서 상당비중 차지하면서 순차적이면 병렬 수행 효과는 제한적이다.

축소(reduction)은 파이프라인에서 만들어진 모든 원소를 하나로 합치는 작업으로 Stream의 reduce(중 하나) min, max, count, sum 같은 완성된 메서드 중 하나를 선택해 수행한다. anyMatch, allMatch, nonMatch 같이 조건에 맞으면 바로 반환되는 메서드도 병렬화에 적합하다. 반면 가변축소(mutable)를 수행하는 Stream의 collect는 병렬화에 적합하지 않다. 컬렉션들을 합치는 부담이 크기 때문이다.

 

위를 적용해 n 이하의  소수 개수를 계산하는 함수를 작성하면 다음과 같다

	static long pi(long n) {
		return LongStream.rangeClosed(2, n)
				.parallel()
				.mapToObj(BigInteger::valueOf)
				.filter(i -> i.isProbablePrime(50))
				.count();
	}

parallel이 있을 때와 없을때 각각 4, 18초가 걸렸다. 

 

추가적으로 무작위 수들로 이뤄진 스트림을 병렬화 하고싶으면 ThreadLocalRandom보다는 SplitableRandom 을 이용하자 . 

결론
계산도 올바르고 성능도 빨라질 것이라는 확신 없이는 스트림 파이프라인 병렬화는 하지말자.
병렬화 하는 편이 낫다고 판단하더라도 꼼꼼히 테스트하여 변경후에도 잘 돌아가는지, 성능이 좋아졌는지 테스트하자.