> For the complete documentation index, see [llms.txt](https://jihunparkme.gitbook.io/docs/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://jihunparkme.gitbook.io/docs/book/modern-java-in-action/02.md).

# 02.함수형 데이터 처리

## PART II. 함수형 데이터 처리

### Stream

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

* 선언형으로 컬렉션 데이터를 처리
* 멀티스레드 코드를 구현하지 않아도 데이터를 투명하게 병렬로 처리
* 복잡한 루프, 조건문 등이 필요 없이 선언형 코드와 동작 파라미터화를 활용하면 변하는 요구사항에 쉽에 대응할 수 있다.
* 스트림 API 를 통해 데이터 처리를 병렬화로 진행하면서 스레드와 락을 걱정할 필요가 없다.

**특징**

* 선언형 : 간결성 + 가독성 향상
* 조립 : 유연성 향상
* 병렬화 : 성능 향상

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

* 상태 있는 연산(stateful operation) : 값을 계산하는 데 필요한 상태를 저장하는 연산 (`reduce`, `sorted`, `distinct`)
* 상태 없는 연산(stateless operation) : 상태를 저장하지 않는 연산 (`filter`, `map`)

**Java7**

```java
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**

```java
//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**

```java
//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 를 인수로 받아 일치하는 모든 요소를 포함하는 스트림 반환

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

#### Slicing

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

  ```java
  List<Dish> sliceMenu1 = specialMenu.stream()
                                      .takeWhile(dish -> dish.getCalories() < 320)
                                      .collect(toList());
  ```
* 슬라이싱 후 나머지 요소 선택

  ```java
  List<Dish> sliceMenu2 = specialMenu.stream()
                                      .dropWhile(dish -> dish.getCalories() < 320)
                                      .collect(toList());
  ```
* `limit` : 스트림 축소

  ```java
  List<Dish> dishes = specialMenu.stream()
                                  .filter(dish -> dish.getCalories() > 300)
                                  .limit(3)
                                  .collect(toList());
  ```
* `skip` : 건너뛰기(처음 n개 요소 제외)

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

#### Mapping

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

  ```java
  List<String> dishNames = menu.stream()
                              .map(Dish::getName) //요리명 추출
                              .map(String::length) //요리명 글자 길이 추출
                              .collect(toList());
  ```
* `flatMap` : 생성된 스트림을 하나의 스트림으로 평명화

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

#### Serching & Maching

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

  ```java
  boolean isVegetarian = menu.stream()
      						.anyMatch(Dish::isVegetarian)
  ```
* `anyMatch` : 스트림의 모든 요소가 일치하는지 검사

  ```java
  boolean isHealthy = menu.stream()
      					.allMatch(dish -> dish.getCalories() < 1000);
  ```
* `nonMatch` : 스트림의 요소 중 일치하는 요소가 없는지 확인

  ```java
  boolean isHealthy = menu.stream()
      					.noneMatch(dish -> dish.getCalories() >= 1000);
  ```
* `findFirst` : 스트림에서 첫 번째 요소 찾기

  ```java
  Optional<Integer> firstSquaredDivisibleByThree = someNumbers.stream()
      													.map(n -> n * n)
      													.filter(n -> n % 3 == 0)
      													.findFirst();
  ```
* `findAny` : 스트림에서 임의의 요소 반환

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

#### Reducing

* `reduce` : 스트림의 모든 요소를 처리해서 값으로 도출 (초기값, BinaryOperator)
* 누적자를 초깃값으로 설정한 다음 BinaryOperator 로 스트림의 각 요소를 반복적으로 누적자와 합쳐 스트림을 하나의 값으로 리듀싱

  ```java
  //덧셈
  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 대신 특화된 스트림을 반환

  ```java
  int calories = menu.stream() //Stream<Dish> 반환
      				.mapToInt(Dish:getCalories) //IntStream 반환
      				.sum();
  ```
* `boxed` : 특화 스트림을 일반 스트림으로 변환하기

  ```java
  IntStream intStream = menu.stream().mapToInt(Dish::getCalories);
  Stream<Integer> stream = intStream.boxed();
  ```
* `OptionalInt`, `OptionalDouble`, `OptionalLong` 으로 이전 값의 존재 여부를 확인할 수 있다.

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

  int max = maxCalories.orElse(1);
  ```
* `range`(인수 미포함), `rangeClosed`(인수 포함) 로 숫자를 생성할 수 있다.

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

### 스트림 생성

**값으로 스트림 생성**

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

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

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

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

**배열로 스트림 생성**

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

**파일로 스트림 생성**

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

```java
//고유 단어의 수 계산
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` : 생산된 각 값을 연속적으로 계산

  ```java
  //(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` : 생산된 각 값을 연속적으로 계산하지 않음 (상태가 없는 메서드에 주로 사용)

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

## 스트림으로 데이터 수집

* Collectors Class 에서 제공하는 메서드
  * 리듀싱과 요약
  * 요소 그룹화
  * 요소 분할

### Reducing

```java
// 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(", "));
```

**범용 리듀싱 요약 연산**

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

```java
// 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)을 기준으로 스트림의 항목을 그룹화

```java
// 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;
    }));
```

**그룹화된 요소 조작**

```java
// 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())));
```

**다수준 그룹화**

```java
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 는 적용할 컬렉터와 변환 함수를 인수로 받아 다른 컬렉터를 반환

```java
//요리 수를 종류별로 연산
//{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)
  * 분할 함수는 참, 거짓 두 가지 요소의 스트림 리스트를 모두 유지하는 장점이 있다.
  * 프레디케이트를 스트림의 각 항목에 적용한 결과로 항목 분할

```java
//{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**

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

```java
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

```java
/* 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()

```java
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));
    }
}
```

* 사용하기

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

## 병렬 데이터 처리와 성능

### 병렬 스트림

* `parallelStream()` : 각각의 스레드에서 처리할 수 있도록 스트림 요소를 여러 청크로 분할한 스트림
* 내부적으로 `ForkJoinPool` 을 사용

```java
//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://www.baeldung.com/java-microbenchmark-harness)

  ```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'
  ```
* 함수 성능 측정

  * 올바른 자료구조를 선택해야 병렬 실행도 최적의 성능을 발휘할 수 있다.
  * 함수형 프로그래밍을 올바르게 사용하여 병렬 실행의 힘을 이용해보자.

  ```java
  @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();
      }
  }
  ```

**병렬 스트림 주의점**

* 병렬화를 이용하려면, `스트림을 재귀적으로 분할`해야 하고,
* 각 서브 스트림을 `서로 다른 스레드의 리듀싱 연산으로 할당`해야 하고,
* 이들 `결과를 하나의 값으로 합쳐`야 한다.
* 멀티 코어 간의 데이터 이동은 생각보다 비싸므로, 코어 간 데이터 전송 시간보다 훨씬 오래 걸리는 작업만 병렬로 처리하자.
* 또한, 병렬 스트림과 병렬 계산에서는 공유된 가변 상태를 피하자. `상태 공유에 따른 부작용`

  ```java
  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 를 호출하자.
  * 두 서브 태스크의 한 태스크에는 같은 스레드를 재사용할 수 있다.
* 포크/조인 프레임워크를 이용하는 병렬 계산은 디버깅이 어렵다.
* 멀티코어에 포크/조인 프레임워크를 사용하는 것이 순차 처리보다 무조건 빠른 것은 아니다.

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

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

* 풀에 있는 작업자 스레드의 태스크를 재분배하고 균형을 맞출 때 `작업 훔치기 알고리즘`을 사용
  * `작업 훔치기 알고리즘` : 스레드는 자신에게 할당된 태스크를 포함하는 `이중 연결 리스트를 참조`하며 작업이 끝날 때마다 큐의 헤드에서 다른 태스크를 가져와 작업을 처리 -> 할 일이 없어진 스레드는 유휴 상태로 바뀌는 것이 아니라 (모든 큐가 빌 때까지) `다른 스레드 큐의 꼬리에서 작업을 훔쳐`온다.
* 따라서, 태스크 크기를 작게 나누어야 작업자 스레드 간의 작업 부화를 비슷한 수준으로 유지할 수 있다.