02.함수형 데이터 처리

PART II. 함수형 데이터 처리

Stream

Stream : 데이터 처리 연산을 지원하도록 소스에서 추출된 연속된 요소

  • 선언형으로 컬렉션 데이터를 처리

  • 멀티스레드 코드를 구현하지 않아도 데이터를 투명하게 병렬로 처리

  • 복잡한 루프, 조건문 등이 필요 없이 선언형 코드와 동작 파라미터화를 활용하면 변하는 요구사항에 쉽에 대응할 수 있다.

  • 스트림 API 를 통해 데이터 처리를 병렬화로 진행하면서 스레드와 락을 걱정할 필요가 없다.

특징

  • 선언형 : 간결성 + 가독성 향상

  • 조립 : 유연성 향상

  • 병렬화 : 성능 향상

상태 있는 연산과 상태 없는 연산

  • 상태 있는 연산(stateful operation) : 값을 계산하는 데 필요한 상태를 저장하는 연산 (reduce, sorted, distinct)

  • 상태 없는 연산(stateless operation) : 상태를 저장하지 않는 연산 (filter, map)

Java7

List<Dish> lowCaloricDishes = new ArrayList<>();
for (Dish d : dishes) {
    if (d.getCalories() < 400) {
        lowCaloricDishes.add(d);
    }
}

Collections.sort(lowCaloricDishes, new Comparator<Dish>() {
    @Override
    public int compare(Dish d1, Dish d2) {
        return Integer.compare(d1.getCalories(), d2.getCalories());
    }
});
List<String> lowCaloricDishesName = new ArrayList<>();
for (Dish d : lowCaloricDishes) {
    lowCaloricDishesName.add(d.getName());
}

Java8

//stream()
List<String> lowCaloricDishesName =
            menu.stream()
                .filter(d -> d.getCalories() < 400) //조건
                .sorted(comparing(Dish::getCalories)) //정렬
                .map(Dish::getName) //추출
                .collect(toList()); //리스트화

//parallelStream()
List<String> lowCaloricDishesName =
            menu.parallelStream()
                .filter(d -> d.getCalories() < 400)
                .sorted(comparing(Dish::getCalories))
                .map(Dish::getName)
                .collect(toList());

Example

//stream()
List<String> threeHighCaloricDishNames =
                menu.stream()
                    .filter(dist -> dist.getCalories() < 300) //필터링 
                    .map(Dish::getName) //추출
                    .limit(3) //축소
                    .collect(toList()); //리스트화

  • filter : 스트림에서 특정 요소 제외

  • map : 한 요소를 다른 요소로 변환하거나 정보 추출

  • limit : 정해진 요소 개수 제한

  • sorted : 요소 정렬

  • collect : 스트림을 다른 형식으로 변환

  • 중간 연산 : filter, map, limit, sorted, distinct ...

    • 중간 연산을 합친 다음 합쳐진 중간 연산을 최종 연산으로 한 번에 처리(LAZY)

  • 최종 연산 : collect, forEach, count ..

    • 스트림 파이프라인에서 결과를 도출

Stream VS Collection

데이터 계산 시점

  • Collection : 현재 자료구조가 포함하는 모든 값을 메모리에 저장하고 모든 요소는 컬렉션에 추가 전에 요소는 미리 계산되어야 함

    • 모든 정보가 로딩될 때까지 기다려야만 한다. ex) DVD

  • Stream : 요청할 때만 요소를 계산 (LAZY)

    • 로딩된 일부 데이터를 먼저 볼 수 있다. ex) 스트리밍

    • 단 한 번만 소비할 수 있다. -> 다시 탐색 시 새로운 스트림 생성이 필요

데이터 반복 처리 방법

  • Collection : 사용자가 직접 요소를 반복 (외부 반복)

  • Stream : 반복을 알아서 처리하고 결과 스트림값을 어딘가에 저장 (내부 반복)

    • 내부 반복의 경우 투명하게 병렬 처리, 더 최적화된 다양한 순서로 처리가 가능

Stream 활용

Filtering

  • filter, distinct : Predicate 를 인수로 받아 일치하는 모든 요소를 포함하는 스트림 반환

    List<Dish> vegetarianMenu = menu.stream()
                                .filter(Dish::isVegetarian)
                                .distinct() //중복 필터링(hashCode, equals 로 결정)
                                .collect(toList());

Slicing

  • takeWhile, dropWhile : Predicate 를 이용한 슬라이싱 (필터에 걸리면 반복 중단)

    List<Dish> sliceMenu1 = specialMenu.stream()
                                    .takeWhile(dish -> dish.getCalories() < 320)
                                    .collect(toList());

  • 슬라이싱 후 나머지 요소 선택

    List<Dish> sliceMenu2 = specialMenu.stream()
                                    .dropWhile(dish -> dish.getCalories() < 320)
                                    .collect(toList());

  • limit : 스트림 축소

    List<Dish> dishes = specialMenu.stream()
                                .filter(dish -> dish.getCalories() > 300)
                                .limit(3)
                                .collect(toList());

  • skip : 건너뛰기(처음 n개 요소 제외)

    List<Dish> dishes = menu.stream()
                        .filter(dish -> dish.getCalories() > 300)
                        .skip(2)
                        .collect(toList());

Mapping

  • map : 특정 함수 적용 결과 매핑

    List<String> dishNames = menu.stream()
                            .map(Dish::getName) //요리명 추출
                            .map(String::length) //요리명 글자 길이 추출
                            .collect(toList());

  • flatMap : 생성된 스트림을 하나의 스트림으로 평명화

    List<String> uniqueCharacters = 
    words.stream()
    	.map(word -> word.split("")) // 각 단어를 개별 문자를 포함하는 배열로 변환
        .flatMap(Arrays:stram)
        .distinct()
        .forEach(toList());

Serching & Maching

  • allMatch : 스트림에서 적어도 한 요소와 일치하는지 확인

    boolean isVegetarian = menu.stream()
    						.anyMatch(Dish::isVegetarian)

  • anyMatch : 스트림의 모든 요소가 일치하는지 검사

    boolean isHealthy = menu.stream()
    					.allMatch(dish -> dish.getCalories() < 1000);

  • nonMatch : 스트림의 요소 중 일치하는 요소가 없는지 확인

    boolean isHealthy = menu.stream()
    					.noneMatch(dish -> dish.getCalories() >= 1000);

  • findFirst : 스트림에서 첫 번째 요소 찾기

    Optional<Integer> firstSquaredDivisibleByThree = someNumbers.stream()
    													.map(n -> n * n)
    													.filter(n -> n % 3 == 0)
    													.findFirst();

  • findAny : 스트림에서 임의의 요소 반환

    Optional<Dish> dish = menu.stream()
    					.filter(Dish:isVegetarian)
    					.findAny();

Reducing

  • reduce : 스트림의 모든 요소를 처리해서 값으로 도출 (초기값, BinaryOperator)

  • 누적자를 초깃값으로 설정한 다음 BinaryOperator 로 스트림의 각 요소를 반복적으로 누적자와 합쳐 스트림을 하나의 값으로 리듀싱

    //덧셈
int sum = numvers.stream()
    			.reduce(0, Integer::sum) //.reduce(0, (a, b) -> a + b);
// 곱셈
int product = numvers.stream()
    			.reduce(1, (a, b) -> a * b);
// 최댓값
Optional<Integer> max = numbers.stream()
    					.reduce(Integer::max);

숫자형 스트림

  • mapToInt, mapToDouble, mapToLong 메서드는 map 과 같은 기능을 수행하지만 Stream 대신 특화된 스트림을 반환

    int calories = menu.stream() //Stream<Dish> 반환
    				.mapToInt(Dish:getCalories) //IntStream 반환
    				.sum();

  • boxed : 특화 스트림을 일반 스트림으로 변환하기

    IntStream intStream = menu.stream().mapToInt(Dish::getCalories);
Stream<Integer> stream = intStream.boxed();

  • OptionalInt, OptionalDouble, OptionalLong 으로 이전 값의 존재 여부를 확인할 수 있다.

    OptaionInt maxCalories = menu.stream()
    				.mapToInt(Dish:getCalories)
    				.max();

int max = maxCalories.orElse(1);

  • range(인수 미포함), rangeClosed(인수 포함) 로 숫자를 생성할 수 있다.

    int count = IntStream.rangeClosed(1, 100) //1~100 범위
    				.filter(n -> n % 2 == 0) //짝수 스트림
    				.count(); //50

스트림 생성

값으로 스트림 생성

Stream<String> stream = Stream.of("Modern", "Java", "in", "Action");
stream.map(String::toUpperCase).forEach(System.out::println);

Stream<String> emptyStream = Stream.empty();

Null이 될 수 있는 객체로 스트림 생성

//null이 될 수 있는 객체를 포함하는 Stream
Stream<String> values = Stream.of("config", "home", "user")
    						.flatMap(key -> Stream.ofNullable(System.getProperty(key)));

배열로 스트림 생성

int[] numbers = {2, 3, 5, 7, 11, 13, 15, 17};
int sum = Arrays.stream(numbers).sum();

파일로 스트림 생성

  • Files.lines : 주어진 파일의 행 스트림을 문자열로 반환

//고유 단어의 수 계산
long uniqueWords = 0;
try(Stream<String> lines = Files.lines(Paths.get("modernJavaInAction/data.txt"), Charset.defaultCharset())) { //AutoCloseable
    uniqueWords = lines.flatMap(line -> Arrays.stream(line.split(" "))) //고유 단어 수
                        .distinct()
                        .count();
} catch(IOException e) {
    
}

무한 스트림 생성

  • Stream.iterate : 생산된 각 값을 연속적으로 계산

    //(iterate) 짝수 스트림 생성 : 생산된 각 값을 연속적으로 계산
Stream.iterate(0, n -> n + 1)
    	.limit(10)
    	.forEach(System.out::println);

//무한 스트림 생성을 중단하는 방법
IntStream.iterate(0, n -> n<100, n -> n+4) 
    		.forEach(System.out::println);

IntStream.iterate(0, n -> n+4) 
    		.takeWhile(n -> n<100)
    		.forEach(System.out::println);

  • Stream.generate : 생산된 각 값을 연속적으로 계산하지 않음 (상태가 없는 메서드에 주로 사용)

    Stream.generate(Math::random)
    	.limit(5)
    	.forEach(System.out::pringln);

스트림으로 데이터 수집

  • Collectors Class 에서 제공하는 메서드

    • 리듀싱과 요약

    • 요소 그룹화

    • 요소 분할

Reducing

// Collectors.counting
long howManyDishes1 = menu.stream().collect(Collectors.counting());
long howManyDishes2 = menu.stream().count();

// Collectors.maxBy (minBy)
//:주어진 비교자를 이용해서 스트림의 최솟값 요소를 Optional로 감싼 값을 반환 (요소가 없을 경우 return Optional.empty())
Comparator<Dish> dishCaloriesComparator = Comparator.comparingInt(Dish::getCalories);
Optional<Dish> mostCaloriesDish = menu.stream().collect(maxBy(dishCaloriesComparator));

// Collectors.summingInt (summingLong, summingDouble)
//:스트림 항목에서 정수 프로퍼티값의 합
int totalCalories = menu.stream().collect(summingInt(Dish::getCalories));

// Collectors.averagingInt (averagingLong, averagingDouble)
//:스트림 항목에서 정수 프로퍼티값의 평균값
double avgCalories = menu.stream().collect(averagingInt(Dish::getCalories));

// Collectors.summarizingInt (summarizingLong, summarizingDouble)
//:스트림 내 항목의 최대, 최소, 합계, 평균 등의 정수 정보 통계 수집
IntSummaryStatistics menuStatistics = menu.stream().collect(summarizingInt(Dish::getCalories));
IntSummaryStatistics{count=10, sum=4500, min=80, average=512.1221, max=780}

// Collectors.joining
//:내부적으로 StringBuilder를 이용하여 문자열 생성
String shortMenu = menu.strea().map(Dish::getName).collect(joining(", "));

범용 리듀싱 요약 연산

  • 문제를 해결할 수 있는 다양한 해결 방법 중 문제에 특화된 해결책을 골라, 가독성과 성능을 모두 잡아보자.

// max (시작값이 없으므로 Optional 반환)
Optional<Dish> mostCalorieDish = menu.stream()
    				.collect(reducing((d1, d2) -> d1.getCalories() > d2.getCalories() ? d1 : d2));

// sum
int totalCalories = menu.stream().collect(reducing(0, Dish::getCalories, (i, j) -> i + j));

int totalCalories = menu.stream().collect(Dish::getCalories, Integer::sum);

int totalCalories = menu.stream().map(Dish::getCalories).reduce(Integer::sum).get
    
int totalCalories = menu.stream().mapToInt(Dish::getCalories).sum()

Grouping

  • 분류 함수

    • Collectors.groupingBy(f, toList())

    • 하나의 프로퍼티값(Key)을 기준으로 스트림의 항목을 그룹화

// Type Groupging
//{FISH=[salmon], OTHER=[fries, rice], MEAT=[pork, beef, chichen]}
Map<Dish.Type, List<Dish>> dishesByType = menu.stream().collect(groupingBy(Dish::getType));

// Calories Groupging
public enum CaloricLevel { DIET, NORMAL, FAT }
Map<CaloricLevel, List<Dish>> dishesByCaloricLevel = menu.stream().collect(
    groupingBy(dish -> {
        if (dish.getCalories() <= 400) return CaloricLevel.DIET;
        else if (dish.getCalories() <= 700) return CaloricLevel.NORMAL;
        else return CaloricLevel.FAT;
    }));

그룹화된 요소 조작

// Type & Calories Groupging (데이터가 없는 Type 도 포함)
//{FISH=[], OTHER=[fries, rice], MEAT=[pork, beef, chichen]}
Map<Dish.Type, List<Dish>> caloricDishedByType = menu.stream()
    .collect(groupingBy(Dish::getType, filtering(dish -> dish.getCalories() > 500, toList())));

// 그룹의 각 요리를 관리 이름 목록으로 변환
Map<Dish.Type, List<String>> dishNamesByType = menu.stream()
    .collect(groupingBy(Dish::getType, mapping(Dish.getName, toList())));

// 각 형식의 요리 태그 추출
//{FISH=[roasted, tasty], OTHER=[fried, fresh], MEAT=[salty, greasy]}
Map<Dish.Type, Set<String>> dishNamesByType = menu.stream()
    .collect(groupingBy(Dish::getType, flatMapping(dish -> dishTags.get(dish.getName()).stream(), toSet())));

다수준 그룹화

public enum CaloricLevel { DIET, NORMAL, FAT }
Map<Dish.Type, Map<CaloricLevel, List<Dish>>> dishesByTypeCaloricLevel = menu.stream()
    .collect(
    	groupingBy(Dish::getType, // 첫 번째 수준의 분류 함수    
            groupingBy(dish -> { // 두 번째 수준의 분류 함수
                if (dish.getCalories() <= 400) return CaloricLevel.DIET;
                else if (dish.getCalories() <= 700) return CaloricLevel.NORMAL;
                else return CaloricLevel.FAT;
            })
     )
);
//{FISH={DIET=[prawns], NORMAL=[salmon]}, MEAT={DIET=[chicken], NORMAL=[beef], FAT=[pork]}, ...}

서브그룹으로 데이터 수집

  • 처음부터 값이 존재하지 않는 Key 는 Map 에 추가되지 않으므로 Optional wrapper 를 사용할 필요가 없음

    • groupingBy Collector 는 Stream 첫 번째 요소를 찾은 이후에 그룹화 맵에 새로운 키를 추가(lazy)

      • 리듀싱 컬렉터는 절대 Optional.empty()를 반환하지 않음

    • collectingAndThen 는 적용할 컬렉터와 변환 함수를 인수로 받아 다른 컬렉터를 반환

//요리 수를 종류별로 연산
//{MEATH=3, FISH=5, OTHER=2}
Map<Dish.Type, Long> typesCount = menu.stream().collect(
											groupingBy(Dish::getType, counting()));

//요리 종류 중 가장 높은 칼로리를 갖는 요리
//{MEATH=Optional[pork], FISH=Optional[salmon], OTHER=Optional[buger]}
Map<Dish.Type, Optional<Dish>> mostCaloricByType = menu.stream().collect(
						groupingBy(Dish::getType, maxBy(comparingInt(Dish::getCalories))));

//{MEATH=pork, FISH=salmon, OTHER=buger}
Map<Dish.Type, Dish> mostCaloricByType = menu.stream().collect(
    groupingBy(Dish::getType, //분류 함수
               collectingAndThen(
                   maxBy(comparingInt(Dish::getCalories)), // 감싸인 컬렉터
                   Optional::get) // 변환 함수
              )
); 

//각 요리 형식에존재하는 모든 CaloricLevel 값
//{MEATH=[DIET, NORMAL], FISH=[NORMAL, FAT], OTHER=[DIET, NORMAL]}
Map<Dish.Type, Set<CaloricLevel>> caloricLevelsByType = menu.stream().collect(
    groupingBy(Dish::getType, mapping(dish -> {
        if (dish.getCalories() <= 400) return CaloricLevel.DIET;
        else if (dish.getCalories() <= 700) return CaloricLevel.NORMAL;
        else return CaloricLevel.FAT;
    },
	toCollection(HashSet::new)))
);

Partitioning

  • 분할 함수

    • return Boolean (true or false)

    • 분할 함수는 참, 거짓 두 가지 요소의 스트림 리스트를 모두 유지하는 장점이 있다.

    • 프레디케이트를 스트림의 각 항목에 적용한 결과로 항목 분할

//{false=[pork, beef, salmon], true=[pizza, rice]}
Map<Boolean, List<Dish>> partitionedMenu = menu.stream().collect(
    										partitioningBy(Dish::isVegetarian));

//{false={MEATH=[DIET, NORMAL], FISH=[NORMAL, FAT]}, true={OTHER=[DIET, NORMAL]}}
Map<Boolean, Map<Dish.Type, List<Dish>>> vegetarianDishesByType = menu.stream().collect(
    										partitioningBy(Dish::isVegetarian, //- 분할 함수
                                                          groupingBy(Dish::getType))); //- 두 번째 컬렉터

//채식 요리와 채식이 아닌 요리 각 그룹에서 가장 칼로리가 높은 요리
//{false=pork, true=pizza}
Map<Boolean, Dish> mostCaloricPartitionedByVegetarian = menu.stream().collect(
								partitioningBy(Dish::isVegetarian,
                                              collectingAndThen(maxBy(comparingInt(Dish::getCalories)), 
                                                               Optional::get)));

Example

  • 숫자를 소수와 비소수로 분할하기

public boolean isPrime(int candidate) {
    int candidateRoot = (int) Math.sqrt((double) candidate);
    return IntStream.range(2, candidateRoot)
        .noneMatch(i -> candidate % i == 0);
}

public Map<Boolean, List<Integer>> partitionPrimes(int n) {
    return IntStream.rangeClosed(2, n).boxed()
        .collect(partitioningBy(candidate -> isPrime(candidate)));
}

Collector

  • Collector Interface

/* T : 수집될 스트림 항목의 제네릭 형식
 * A : 수집 과정에서 중간 결과를 누적하는 객체의 형식(누적자)
 * R : 수집 연산 결과 객체의 형식
 * supplier() -> accumulator() -> combiner() -> finisher()
 */
public interface Collector<T, A, R> {
    //supplier() : 새로운 결과 컨테이너 만들기(수집 과정에서 빈 누적자 인스턴스를 만드는 파라미터가 없는 함수)
    Supplier<A> supplier();
    //accumulator() : 결과 컨테이너에 스트림 요소 추가하기(리듀싱 연산을 수행하는 함수 반환)
    BiConsumer<A, T> accumulator();
    //combiner() : 두 결과 컨테이너 병합(스트림의 서로 다른 서브파트를 병렬로 처리할 때 누적자가 결과를 어떻게 처리할지 정의)
    BinaryOperator<A> combiner();
    //finisher() : 최종 변환값을 결과 컨터네이너 적용하기
    Function<A, R> finisher();
    // characteristics() : 컬렉터의 연산을 정의하는 Characteristics 형식의 불변 집합 반환(어떤 최적화를 이용해 리듀싱 연산을 수행할 것인지 힌트 제공)
    Set<Characteristics> characteristics();
}

  • Example toList()

public class ToListCollector<T> implements Collector<T, List<T>, List<T>> {
    @Override
    public Supplier<List<T>> supplier() { //<- 수집 연산의 시작
        return () -> new ArrayList<T>(); //= return ArrayList::new; (생성자 참조)
    }
    @Override
    public BiConsumer<List<T>, T> accumulator() { //<- 탐색한 항목을 누적하고 바로 누적자를 수정
        return (list, item) -> list.add(item); //= return LisT::add;
    }
    @Override
    public Function<List<T>, List<T>> finisher() { 
        return Fcuntion.identity(); //<- 항등 함수
    }
    @Override
    public BinaryOperator<List<T>> combiner() {
        return (list1, list2) -> { //<- 두 번째 콘텐츠와 합쳐서 첫 번째 누적자 수정
            list1.addAll(list2); //<- 변경된 첫 번째 누적자 반환
            return list1;
        };
    }
    @Override
    public Set<Characteristics> characteristics() {
        /* UNORDERED: 리듀싱 결과는 스트림 요소의 방문 순서나 누적 순서에 영향을 받지 않음
      * CONCURRENT: 다중 스레드에서 accumulator 함수를 동시에 호출할 수 있으며, 스트림의 병렬 리듀싱 수행 가능
      * IDENTITY_FINISH: 리듀싱 과정의 최종 결과로 누적자 객체를 바로 사용
      */
        return Collections.unmodifiableSet(EnumSet.of(IDENTITY_FINISH, CONCURRENT));
    }
}

  • 사용하기

//Before
List<Dish> dishes = menu.stream().collect(toList());
//After
List<Dish> dishes = menu.stream().collect(new ToListCollector<Dish>());

병렬 데이터 처리와 성능

병렬 스트림

  • parallelStream() : 각각의 스레드에서 처리할 수 있도록 스트림 요소를 여러 청크로 분할한 스트림

  • 내부적으로 ForkJoinPool 을 사용

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

스트림 성능 측정

  • Java Microbenchmark Harness(JMH) 라이브러리를 통해 벤치마크 구현이 가능

  • Microbenchmarking with Java

    // https://mvnrepository.com/artifact/org.openjdk.jmh/jmh-core (핵심 JMH 구현 포함)
implementation group: 'org.openjdk.jmh', name: 'jmh-core', version: '1.34'
    
// https://mvnrepository.com/artifact/org.openjdk.jmh/jmh-generator-annprocess (JAR 파일 생성에 도움을 주는 어노테이션 프로세서 포함)
testImplementation group: 'org.openjdk.jmh', name: 'jmh-generator-annprocess', version: '1.34'

  • 함수 성능 측정

    • 올바른 자료구조를 선택해야 병렬 실행도 최적의 성능을 발휘할 수 있다.

    • 함수형 프로그래밍을 올바르게 사용하여 병렬 실행의 힘을 이용해보자.

    @BenchmarkMode(Mode.AverageTime) //<- 벤치마크 대상 메서드 실행에 걸린 평균 시간 측정
@OutputTimeUnit(TimeUnit.MILLISECONDS) //<- 벤치마크 결과를 밀리초 단위 출력
@Fork(value = 2, jvmArgs = { "-Xms4G", "-Xmx4G" }) //<- 4GB 힙 공간을 제공한 환경에서 두 번의 수행을 통해 결과 신뢰성 확보
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);
    }
    
	@Benchmark
	public long parallelRangedSum() {
		/* iterate() 대신 LongStream.rangeClosed()
		 * 기본형 long을 직접 사용하여 박싱, 언박싱 오버헤드 해결
          * 청크로 분할할 수 있는 숫자 범위 생산
          */
		return LongStream.rangeClosed(1, N).parallel().reduce(0L, Long::sum);
	}

    @TearDown(Level.Invocation) //<- 매 벤치마크 실행 후 GC 동작 시도
    public void tearDown() {
        System.gc();
    }
}

병렬 스트림 주의점

  • 병렬화를 이용하려면, 스트림을 재귀적으로 분할해야 하고,

  • 각 서브 스트림을 서로 다른 스레드의 리듀싱 연산으로 할당해야 하고,

  • 이들 결과를 하나의 값으로 합쳐야 한다.

  • 멀티 코어 간의 데이터 이동은 생각보다 비싸므로, 코어 간 데이터 전송 시간보다 훨씬 오래 걸리는 작업만 병렬로 처리하자.

  • 또한, 병렬 스트림과 병렬 계산에서는 공유된 가변 상태를 피하자. 상태 공유에 따른 부작용

    public static long sideEffectParallelSum(long n) {
    Accumulator accumulator = new Accumulator();
    LongStream.rangeClosed(1, n).parallel().forEach(accumulator::add);
    return accumulator.total;
}

public static class Accumulator {
    private long total = 0;
    public void add(long value) { total += value; }
}

// 여러 스레드에서 동시에 누적자를 수정하면서 올바른 결과값이 나오지 않게 된다.
System.out.println(ParallelStreams.sideEffectParallelSum(10_000_000L))

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

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

    • 순차 스트림과 병렬 스트림 중 어떤 것이 좋을지 모르겠다면 벤치마크로 성능을 측정해보자.

  • 박싱을 주의하자.

    • 자동 박식/언박싱은 성능을 크게 저하시킬 수 있는 요소다.

    • 박싱 동작을 피하도록 되도록 기본형 특화 스트림을 사용해보자. (IntStream, LongStream, DoubleStream)

  • 순차 스트림보다 병렬 스트림에서 성능이 떨어지는 연산이 있다.

    • 요소의 순서에 의존(limit, findFirst)하는 연산은 병렬 스트림에서 더 비싼 비용이 들어간다.

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

    • N(처리해야 할 요소 수) * Q(하나의 요소 처리 비용)

    • Q가 높아지는 것은 병렬 스트림으로 성능 개선 가능성이 있음을 의미한다.

  • 소량의 데이터에서는 병렬 스트림이 도움이 되지 않는다.

    • 소량의 데이터는 병렬화 과정에서 생기는 부가 비용을 상쇄할 만큼의 이득을 얻지 못한다.

  • 스트림을 구성하는 자료구조가 적절한지 확인하자.

    • ex. ArrayList를 LinkedList 보다 효율적으로 분할할 수 있다.

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

    • Sized 스트림은 정확히 같은 크기의 두 스트림으로 분할 가능하여 효과적이지만, 필터 연산은 스트림 길이 예측이 불가하여 효과적인지 알 수 없다.

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

    • 병합 과정 비용이 비싸다면 병렬 스트림으로 얻은 성능 이익이 상쇄

소스분해성

ArrayList

Excellence

LinkedList

Bad

IntStream.range

Excellence

Stream.iterate

Bad

HashSet

Good

TreeSet

Good

포크/조인 프레임워크

병렬 스트림을 제대로 사용하기 위해 병렬 스트림 내부 구조를 살펴보자.

병렬화할 수 있는 작업을 재귀적으로 작은 작업으로 분할한 후, 서브태스크 각각의 결과를 합쳐 전체 결과를 만들도록 설계

서브태스크를 스레드 풀(ForkJoinPool)의 작업자 스레드에 분산 할당하는 ExecutorService Intergace 구현

  • 스레드 풀을 이용하려면, RecursiveTask의 서브클래스를 만들어야 한다.

포크/조인 프레임워크의 알고리즘은 분할/정복 알고리즘의 병렬화 버전이다.

포크/조인 프레임워크 제대로 사용하기

  • join 메서드를 태스크에 호출하면 태스크가 생산하는 결과가 준비될 때까지 호출자를 블록시킨다.

  • RecursiveTask 내에서는 ForkJoinPool의 invoke 메서드를 사용하지 말자.

    • 대신 compute나 fork 메서드 직접 호출하고, 순차 코드에서 병렬 계산을 시작 시에만 invoke 사용

  • 서브태스크에서 fork 메서드를 호출해서 ForkJoinPool의 일정 조절

    • 한 쪽 작업에는 fork, 다른 한 쪽 작업에는 compute 를 호출하자.

    • 두 서브 태스크의 한 태스크에는 같은 스레드를 재사용할 수 있다.

  • 포크/조인 프레임워크를 이용하는 병렬 계산은 디버깅이 어렵다.

  • 멀티코어에 포크/조인 프레임워크를 사용하는 것이 순차 처리보다 무조건 빠른 것은 아니다.

포크/조인 프레임워크의 작업 훔치기

포크/조인 프레임워크의 병렬 처리 방법

  • 풀에 있는 작업자 스레드의 태스크를 재분배하고 균형을 맞출 때 작업 훔치기 알고리즘을 사용

    • 작업 훔치기 알고리즘 : 스레드는 자신에게 할당된 태스크를 포함하는 이중 연결 리스트를 참조하며 작업이 끝날 때마다 큐의 헤드에서 다른 태스크를 가져와 작업을 처리 -> 할 일이 없어진 스레드는 유휴 상태로 바뀌는 것이 아니라 (모든 큐가 빌 때까지) 다른 스레드 큐의 꼬리에서 작업을 훔쳐온다.

  • 따라서, 태스크 크기를 작게 나누어야 작업자 스레드 간의 작업 부화를 비슷한 수준으로 유지할 수 있다.

Last updated