병렬 데이터 처리와 성능

병렬 스트림

컬렉션에 parallelStream을 호출하면 병렬 스트림이생성된다. 병렬 스트림이란 각각의 스레드에서 처리할수 있도록 스트림 요소를 여러 청크로 분할한 스트림이다.

• 순차 스트림을 병려 스트림으로 변환하기
  • parallel()를 통해 리듀싱 연산을 여러 청크에 병렬로 수행 후 리듀싱 연산으로 생성된 결과를 합쳐서 결과를 도출한다.

     Stream.iterate(1L, i-> i+1).limit(n).parallel().reduce(0L, Long::sum);
    

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

• 스트림 성능 측정
  • 병렬화를 했다고 해서 성능이 더 좋아졌다고 생각하면 안되고 실제 측저을 통해 확인을 해야 한다.
  • JHM 라이브러리를 이용해 작은 벤치마크를 구현할수있으며 어노테이션 기반방식을 지우하며, 안정적으로 자바 프로그램이나 JVM을 대상하는 다른 언어용 벤치마크를 구현할 수있다.
  • LongSteram.rageClosed
    • 기본형 long을 직접 사용하므로 박싱과 언박싱 오버헤드가 사라진다.
    • 쉽게 청크로 분할할 수 있는 숫자 범위를 생산한다.
  • 병렬 스트림의 올바른 사용법
    • 병렬 스트림을 잘못 사용하면서 발생하는 많은 문제는 공유된 상태를 바꾸는 알고리즘을 사용하기 때문에 일어난다.
    • 여러 스레드에서 동시에 누적자를 실행하면 문제가 발생한다. 보통 객체 상태를 바꾸는 forEach블록내부에서 add같은 메서드를 호출하면 문제가 발생한다.
• 병렬 스트림 효과적으로 사용하기
  • 항상 병렬 스트림이 순차 스트림보다 빠른 것은 아니기 때문에 적절한 벤치마크로 직접성능을 측정하는 것이 바람직하다.
  • 자동 박싱과 언박싱은 성능을 크게 저하시킬 수 있는 요소다. 자바 8은 기본형 특화 스트림(IntStream, LongStream, DoubleStream)을 제공한다. 되도록이면 기본형 특화 스트림을 사용하는 것이 좋다.
  • limit나 findFirst처럼 요소의 순서에 의존하는 연산은 병렬스트림에서 수행하려면 성능이 떨어지기도 한다.
  • 처리해야 할 요소 수가 N이고 처리 비용을 Q라고 할때 전체 스트림 파이프라인 처리비용은 N*Q로 예상 할수 있으며 Q가 높아진다는 것은 병렬 스트림으로 성능을 개선할 수 있는 가능성을 의미한다.
  • 소량의 데이터에서는 병렬 스트림이 도움 되지 않는다.
  • 스트림을 구성하는 자료구조가 적절한지 확인하라. 예를 들면 ArrayList를 LinkedList보다 효율적으로 분할할 수 있다. ArrayList는 요소를 탐색하지 않고도 리스트를 분할할 수 있다.
  • 파이프 라인 중간 연산이 특성을 어떻게 바꾸는지에 따라 성능이 달라질수 있다. SIZED 스트림은 정확히 같은 크기의 두 스트림으로 분할할 수 있고 필터 연산은 정확한 길이를 예측할 수 없어 효과적으로 처리 된다는 보장이 없다.
  • 최종 연산의 병합과정의 비용이 비싸다면 병렬 스트림으로 얻은 성능의 이익이 서브스트림의 부분결과를 합치는 과정에서 상쇄될 수 있다.
  • 병렬화 친밀도
    • ArayList : 훌륭함
    • LinkedList : 나쁨
    • IntStream.range : 훌륭함
    • Stream.iterate : 나쁨
    • HashSet : 좋음
    • TreeSet : 좋음

포크/조인 프레임워크

병렬화할 수 잇는 작업을 재귀적으로 작은 작업으로 분할한 다음에 서브태스크 각각의 결과를 합쳐서 전체 결과를 마들도록 설계도엇다. 서브태스크를 스레드 풀의 작업자 스레드에 분산 할당하는 ExcutorService 인터페이스를구현한다.

1. RecursiveTask
   • 스레드 풀을 이용하려면 RecursiveTask의 서브클래스를 만들어야 한다. 여기서 R은 병렬화된 태스크가 생성하는 결과 형식 또는 겨과가 없을때는 RecursiveAction 형식이다.
   • RecusiveTask를 정의하려면 추상 메서드 compute를 구현해야 한다.

       protected abstract R compute();
     

2. ForkJoinSumCalculator
   • ForkJoinSumCalculator를 ForkJoinPool로 전달하면 풀의 스레다가 ForkJoinSumCalculator의 compute 메서드를 실행하면서 작업을 수행한다.
   • compute 메서드는 병렬수 실행할 수 있는 만큼 태스크의크기가 충분히 작어졌는지 확인하며 재귀적으로 반복되면서 주어진 조건을 만족할 때까지 태스크 분할을 반복한다.
3. 포크/조인 프레임워크를 제대로 사용하는 방법
   • join 메서드를태스크에 호출하면 태스크가 생산하는 결과가 준비될 때까지 호출자를 블록시킨다. 따라서 서브태스크가 모두 시작된 다음에 join을 호출해야 한다.
   • RecursiveTask 내에서는 ForkJoinPool의 invoke 메서드를사용하지 않아야 하고 compute나 fork 메서드를 직접 호출해서 사용한다.
   • 서브태스크에 fork 메서드를 호출해서 ForkJoinPool의 일정을조절할수있다. 왼쪽, 오른쪽 작업 모두 fork 메서드를 호출하는게 아닌 한쪽엔 compute를 호출하여 불필요한 태스크를 할당하는 오버헤드를 피할 수 있다.
   • 포크/조인 프레임워크는 fork라 불리는 다른 스레드에서 compute를호출하므로 스택 트레이스가 도움되지 않는다.
   • 작업 훔치기

     작업 훔치기 기법은 ForkJoinPool의 모든 스레드를 거의 공정하게 분할한다. 각각의 스레드는 자신에게 할당된 태스크를 포함하는 이중연결 리스트를 참조하면서 작업이 끝날 때마다 큐의 헤드에서 다른 태스크를 가져와서 작업을 처리한다.

Spliterator 인터페이스

• 분할할 수 있는 반복자라는 의미로 Iterater 처럼 소스의 요소 탐색 기능을 제공한다는점은 같지만 병렬작업에 특화되어있다.
• T는 탐색하는 요소의 형식을 가리키며 tryAdvance 메서드는 요소를 하나씩 순차적으로 소비하면서 탐색해야 할 요소가 남았으면 참을 반환한다.
• trySplit 메서드는 일부 요소를 분할해서 두번째 Spliterator를 생성하는 메서드다.
• estimateSize 메서드는 탐색해야 할 요소 수 정보를 제공한다.

  public interface Splitertor<T>{
    boolean tryAdvance(Consumer<? super T> action);
    Spliterator<T> trySplit();
    long estmateSize(); 	
    int characteristics();
  }
  

• 분활과정
  1. 첫번째 Spliterator에서 trySplit을 호출하면 두번째 Spliterator가 생성된다.
  2. 두개의 Spliterator에 rySplit를 다시 호출하면 네개가 생성된다.
  3. 이처럼 trySplit의 결과가 null이 될때까지 이 과정을 반복한다.
  4. 호출한 모든 trySplit의 결과가 null 이면 재귀 분할 과정이 종료 된다.
• 특성
  • characteristics라는 추상메서드도 정의한다. characteristics 메서드는 Spliterator 자체의 특성 집합을 포함하는 int를 반환한다.
  • Spliterator를이용하는 프로그램은 이들 특성을 참고해서 Spliterator를 더 잘 제어하고 최적화 할 수 있다.