• 이 장의 내용

    • 병렬 스트림으로 데이터를 병렬 처리하기
    • 병렬 스트림의 성능 분석
    • 포크/조인 프레임워크
    • Spliterator로 스트림 데이터 쪼개기

  • 병럴 스트림

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

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

    • 컬렉션에서 parallelStream()을 호출하면 병렬 스트림이 생성됨

    • 예시 코드

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

  • 순차 스트림을 병렬 스트림으로

    병렬 스트림의 간단한 처리과정

    병렬 스트림의 간단한 처리과정

    • 순차 스트림에서 parallel()을 호출하면 기존의 함수형 리듀싱 연산이 병렬로 처리

    • 예시 코드

      // 앞선 예제와 코드에서 달라지는 점은 parallel()부분만 다름
// 내부적으로는 여러 청크로 분할되어 처리됨(정확하게는 병렬처리 플래그가 true로 변경됨)
// 반대로 sequential()을 호출하면 병렬 스트림을 순차 스트림으로 바꿀 수 있음
public long parallelSum(long n) {
	return Stream.iterate(1L, i -> i + 1)
		.limit(n)
		.parallel()
		.reduce(0L, Long::sum);
}

// 아래처럼 parallel, sequential이 여러번 사용된 경우 마지막으로 호출한 부분이 적용
// 예제는 병렬 스트림으로 진행
stream.parallel()
	.filter(...)
	.sequential()
	.map(...)
	.parallel()
	.reduce();

  • 병렬 스트림에서 사용하는 스레드 풀 설정

    • 병럴 스트림은 내부적으로 ForkJoinPool을 사용함
    • 기본적으로 ForkJoinPool은 프로세서 수, 즉 Runtime.getRuntim().availableProcessors()가 반환하는 값에 상응하는 스레드를 가짐
    • System.setProperty(”java.util.concurrent.ForkJoinPool.common.parallelism”, “12”)를 통해 전역으로 설정 가능(하나의 병렬 스트림에서만 설정하는 것은 아직 불가능)
    • 기본적으로 프로세서 수와 같게 설정되므로 특별한 경우가 아니라면 커스텀하지 않는 것이 권장됨

  • 스트림 성능 측정

    • **자바 마이크로벤치마크 하니스(JMH)**라는 라이브러리를 이용해 벤치마크 측정 가능

    • JMH를 이용하면 어노테이션을 통해 처리 가능하며, JVM을 대상으로 하는 다른 언어용 벤치마크 구현가능

    • dependency

      implementation "org.openjdk.jmh:jmh-core:1.17.4"

    • 예시 코드

      @BenchmarkMode(Mode.averageTime)
@OutputTimeUnit(TimeUnit.MILLISECONDS)
@Fork(2, jvmArgs={"-Xms46", "-Xmx4G"})
public class ParallelStreamBenchmark {
	
	private static final long N = 10_000_000L;
	
	@Benchmark
	public long sequentialSum() {
		return Stream.iterate(1L, i -> i + 1).limit(N).reduce(0L, Long::sum);
	}

	@TearDown(Level.Invocation)
	public void tearDown() {
		System.gc();
	}
}

  • 더 특화된 메서드 사용

    • rangeClosed 메서드는 사용하면 iterate에 비해 가지는 장점
      • rangeClosed는 기본형 long을 직접 사용하므로 박싱과 언박싱 오버헤드가 사라짐
      • rangeClosed는 쉽게 청크로 분할할 수 있는 숫자 범위를 생성
      • 예를 들어 1~20범위의 숫자를 각각 1~5, 6~10, 11~15, 16~20 범위로 구분 가능
      • 이러한 방식이 특화된 스트림
    • 위 예제를 통해 상황에 따라 어떤 알고리즘을 병렬화하는 것보다 적절한 자료구조를 선택하는 것이 중요할 수 있음
    • 하지만, 병렬 스트림은 항상 좋은 것은 아님 → 분할하고 합치는 과정등의 비용이 저렴하진 않기 때문

  • 병렬 스트림의 올바른 사용법

    • 많은 문제는 공유된 상태를 바꾸는 알고리즘을 사용하기 때문에 발생

    • 예시 코드

      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; }
}

    • 예시 코드의 문제

      • 위 코드는 본질적으로 순차 실행할 수 있도록 구현되어 있어 병렬로 실행하면 참사 발생
      • 특히 total을 접근할 때 데이터 레이스 문제 발생 → 다수의 스레드에서 동시에 데이터에 접근하는 문제
      • 하지만, 동기화로 문제를 해결한다고 하면 결국 병렬화 특성이 사라짐
      • 이러한 이유로 병렬 스트림을 올바르게 사용하기 위해서는 공유된 가변 상태를 피해야 하는 사실을 기억해야함

  • 병렬 스트림 효과적으로 사용하기

    • 확신이 서지 않으면 직접 측정

      • 순차 스트림을 병렬 스트림으로 바꾸는 것은 어렵지 않지만 무조건 병렬로 처리하는 것이 옳지는 않음(언제나 병렬 스트림이 빠르진 않음)
      • 또한 병렬 스트림의 수행 과정은 투명하지 않음
      • 따라서 판단이 잘 서지 않는다면 벤치마크로 직접 측정

    • 박싱 주의

      • 자동 박싱과 언박싱은 성능을 크게 저하시킬 수 있는 요소
      • Java 8에서는 박싱 동작을 방지하기 위해 IntStream, LongStream 등을 제공

    • 순차 스트림보다 병렬 스트림 성능이 떨어지는 연산 존재

      • 특히 limit, findFirst 처럼 요소의 순서에 의존하는 연산을 병렬 스트림에서 수행하려면 비용이 높음
      • 예를 들어 findAny의 경우 요소의 순서와 관계 없으므로 findFirst보다 성능이 좋음
      • 정렬된 스트림에서 unordered를 호출하면 비정렬 스트림을 얻을 수 있음
      • 스트림에 N개 요소가 존재할 때 요소의 순서가 상관없다면, 비정렬 스트림을 통해 limit을 호출하는 것이 효과적

    • 스트림에서 수행하는 전체 파이프라인 연산 비용을 고려

      • 처리해야 할 요소 수가 N이고 하나의 요소를 처리하는데 드는 비용을 Q라고 할 때 전체 스트림 파이프라인 처리 비용을 N * Q로 예상 가능
      • 즉, Q가 높아진다는 것은 병렬 스트림으로 성능을 개선할 수 있는 가능성이 있음을 의미

    • 소량의 데이터는 병렬 스트림이 도움 되지 않음

      • 소량의 데이터를 처리하는 상황에서는 병렬화 과정에서 생기는 부가 비용을 상쇄할 수 있을 만큼의 이득을 얻지 못하기 때문

    • 스트림을 구성하는 자료구조의 적절성 확인

      • 예를 들어 ArrayList를 LinkedList보다 효율적으로 분할할 수 있음
      • LinkedList를 분할하려면 모든 요소를 탐색해야하지만 ArrayList는 요소를 탐색하지 않고도 분할이 가능
      • 또한 range도 쉽게 분리 가능
      • Spliterator를 구현하면 분리 과정의 제어 가능

    • 스트림의 특성과 파이프라인의 중간 연산이 스트림의 특성을 어떻게 바꾸는지에 따라 분해 과정의 성능이 달라질 수 있음

      • 예를 들어 SIZED 스트림은 정확히 같은 크기의 두 스트림으로 분할할 수 있으므로 효과적으로 스트림을 병렬 처리 가능
      • 반면 필터 연산이 있으면 스트림의 길이를 예측할 수 없으므로 효과적으로 병렬 처리할 수 있는지 알 수 없음

    • 최종 연산의 병합 과정 비용을 살펴보자

      • 예를 들면 Collector의 combiner 메서드가 있음
      • 병합 과정의 비용이 비싸다면 병렬 스트림으로 얻은 성능의 이익이 서브스트림의 부분 결과를 합치는 과정에서 상쇄될 수 있음

    • 분해와 관련된 다양한 스트림 소스의 친밀도

      Untitled

  • 포크/조인 프레임워크

    • 포크/조인 프레임워크는 병렬화할 수 있는 작업을 재귀적으로 작은 작업으로 분할한 다음에 서브테스크 각각의 결과를 합쳐서 전체 결과를 만들도록 설계됨
    • 서브테스크를 스레드 풀의 작업자 스레드에 분살 할당하는 ExecutorService 인터페이스 구현
    • Recursive Task 활용

      • 스레드 풀을 이용하려면 RecursiveTask<R>의 서브클래스를 만들어야 함

      • 여기서 R은 병렬화된 테스크가 생성하는 결과 형식, 결과가 없을 때는 RecursiveAction 형식

      • RecursiveTask에는 compute()를 구현해서 활용

      • compute() 메서드는 테스크를 서브테스크로 분할하는 로직과 더 이상 분할할 수 없을 때 개별 서브테스크의 결과를 생산할 알고리즘 정의

      • 다음과 같은 형식을 유지함

        Untitled

      • 예시 코드

        • 일반적으로 애플리케이션에서는 둘 이상의 ForkJoinPool을 사용하지 않음
        • ForkJoinPool을 한 번만 인스턴스화해서 정적 필드에 싱글턴으로 저장해서 사용
        // RecursiveTask 상속받아 포크/조인 프레임워크에서 사용할 테스크 생성
public class ForkJoinSumCalculator extends RecursiveTask<Long> {
    private final long[] numbers; // 더할 숫자 배열
    private final int start; // 이 서브테스크에서 처리할 배열의 초기 위치
    private final int end; // 이 서브테스크에서 처리할 배열의 최종 위치
    public static final long THRESHOLD = 10_000; // 이 값 이하의 서브테스크는 더 이상 분할할 수 없음

    // 메인 테스크를 생성할 때 사용할 공개 연산자
    public ForkJoinSumCalculator(long[] numbers) {
        this(numbers, 0, numbers.length);
    }

    // 메인 테스크의 서브테크스를 재귀적으로 만들 때 사용할 비공개 생성자
    private ForkJoinSumCalculator(long[] numbers, int start, int end) {
        this.numbers = numbers;
        this.start = start;
        this.end = end;
    }

    // RecursiveTask 추상 메서드
    @Override
    protected Long compute() {
        // 이 테스크에서 더할 배열의 길이
        int length = end - start;
        
        // 기준값과 같거나 작으면 순차적으로 결과를 계산
        if(length <= THRESHOLD) {
            return computeSequentially();
        }
        
        // 배열의 첫 번째 절반을 더하도록 서브테스크를 생성
        ForkJoinSumCalculator leftTask =
                new ForkJoinSumCalculator(numbers, start, start + length/2);
        // ForkJoinPool 다른 스레드로 새로 생성한 테스크를 비동기로 실행        
        leftTask.fork();
        // 배열의 나머지 절반을 더하도록 서브테스크 생성
        ForkJoinSumCalculator rightTask =
                new ForkJoinSumCalculator(numbers, start + length/2, end);
        // 두 번째 서브테스크를 동시 실행, 이떄 추가로 분할이 일어날 수 있음
        Long rightResult = rightTask.compute();
        // 첫 번째 서브테스크의 결과를 읽거나 없다면 기다림
        Long leftResult = leftTask.join();
        // 두 테스크의 결과를 조합한 값이 이 테스크의 결과값이 되고 이를 반환
        return leftResult + rightResult;
    }

    // 분할할 수 없을 때 서브테스크의 결과를 계산하는 단순한 알고리즘
    private long computeSequentially() {
        long sum = 0;
        for(int i = start; i < end; i++) {
            sum += numbers[i];
        }
        return sum;
    }
		// ForkJoinSumCalculator 생성자로 원하는 수의 배열을 넘겨줄 수 있음
    public static long forkJoinSum(long n) {
        long[] numbers = LongStream.range(1, n).toArray();
        ForkJoinTask<Long> task = new ForkJoinSumCalculator(numbers);
        return new ForkJoinPool().invoke(task);
				// invoke 메서드의 반환값은 ForkJoinSumCalculator에서 정의한 테스크의 결과가 됨
    }
}

      • ForkJoinSumCalculator의 실행

        • ForkJoinSumCalculator를 ForkJoinPool로 전달하면 풀의 스레드가 ForkJoinSumCalculator의 compute 메서드를 실행하며 작업 수행
        • compute메서드는 병렬로 실행할 만큼 테스크가 작아졌는지 확인하며, 테스크의 크기가 크다고 판단하면 숫자 배열을 반으로 분할해서 두 개의 새로운 ForkJoinSumCalculator로 할당
        • 그러면 다시 ForkJoinPool이 새로 생성된 ForkJoinSumCalculator를 실행하며 이 과정이 재귀적으로 이뤄지며 테스크를 분할
        • 이후 각 테스크는 순차적으로 실행되며 포킹 프로세스로 만들어진 이진트리의 역순으로 방문 → 서브테스크의 부분 결과를 합쳐서 테스크의 최종 결과 계산

  • 포크/조인 프레임워크 제대로 사용하는 방법

    • join 메서드를 테크스에 호출하면 테스크가 생산하는 결과가 준비될 때까지 호출자를 블록
      • 두 서브테스크가 모두 시작된 다음 join을 호출해야 함
      • 그렇지 않다면 각각의 서브테스크가 다른 테스크가 끝나길 기다리는 일이 발생하며 순차 알고리즘보다 느리고 복잡한 프로그램이 됨
    • RecursiveTask내에서는 ForkJoinPool의 invoke메서드를 사용하지 말아햐 함
      • 대신 compute, fork 메서드를 직접 호출할 수 있음
      • 순차 코드에서 병렬 계산을 시작할 때만 invoke 사용
    • 서브테스크에 fork 메서드를 호출해서 ForkJoinPool의 일정을 조절
      • 왼쪽/오른쪽 작업 모두 fork 메서드를 호출하는 것이 자연스러울 것 같지만 한쪽 작업에는 fork를 호출하는 것보다 compute를 호출하는 것이 효율적
      • 이를 통해 두 서브테스크의 한 테스크는 같은 스레드를 재사용할 수 있으므로 풀에서 불필요한 테스크를 할당하는 오버헤드를 줄일 수 있다
    • 포크/조인 프레임워크를 이용하는 병렬 계산은 디버깅이 어려움
      • 보통 스택 트레이스로 문제가 일어나 과정을 쉽게 확인 가능
      • 포크/조인 프레임워크에서는 fork라 불리는 다른 스레드에서 compute를 호출하므로 스택 트레이스가 도움이 되지 않음
    • 포크/조인 프레임워크를 사용하는 것이 순차 처리보다 무조건 좋지 않음
      • 병렬 처리로 성능을 개션하려면 테스크를 여러 독립적인 서브테스크로 불할할 수 있어야 함
      • 각 서브테크스의 실행시간은 새로운 테스크를 포킹하는 데 드는 시간보다 길어야 한다

  • 작업 훔치기

    • ForkJoinPool의 모든 스레드를 거의 공정하게 분할함
    • 각각 스레드는 자신에게 할당된 테스크를 포함하는 이중 연결 리스트를 참조하면서 작업이 끝날 때마다 큐의 헤드에서 다른 테스크를 가져와서 작업을 처리
    • 이때 한 스레드는 다른 스레드보다 자신에게 할당된 테스크를 빠르게 처리할 수 있음
    • 이때 할일이 없어진 스레드는 유휴 상태로 바뀌는 것이 아니라 다른 스레드 큐의 꼬리에서 작업을 훔쳐옴
    • 모든 테스크가 끝날 때까지 이 과정이 반복됨, 즉 테스크의 크기를 작게 나누어야 작업자 스레드 간의 작업부하를 비슷한 수준으로 유지 가능

  • Spliterator 인터페이스

    • Java 8에서는 Spliterator라는 인터페이스를 제공하고 분할할 수 있는 반복자라는 의미를 가짐
    • Iterator처럼 Spliterator는 소스의 요소 탐색 기능을 제공한다는 점은 같지만 Spliterator는 병렬 작업에 특화
    • 예시 코드
    • 분할 과정