5. 스트림 활용

스트림 API는 내부적으로 다양한 최적화가 이루어져 내부 반복 뿐 아니라 코드를 병렬로 실행할지 여부도 결정할 수 있다.

• 순차적인 반복을 단일 스레드로 구현하는 외부 반복으로는 스트림 API만큼의 최적화를 달성할 수 없다.

 

필터링

Predicate를 이용해서 스트림의 요소를 선택하는 방법

Stream<T> filter(Predicate<? super T> predicate);

List<Dish> vegetarianMenu = menu.stream()
                                .filter(Dish::isVegetarian) //채식 요리만 필터링
                                .collect(toList());

고유 요소만 필터링 하는 방법(distinct)

List<Integer> numbers = Arrays.asList(1, 2, 1, 3, 3, 2, 4);
numbers.stream()
        .filter(i -> i % 2 ==0)
        .distinct() //2 중복제거
        .forEach(System.out::println); //2,4 출력

• 고유 여부는 스트림에서 만든 객체의 hashCode, equals로 결정된다.

---

5.2 스트림 슬라이싱

java 9은 Predicate를 이용해서 스트림의 요소를 효과적으로 선택할 수 있도록 takeWhile, dropWhile 두 가지 새로운 메서드를 지원한다.

List<Dish> menu = Arrays.asList(
        new Dish("자바커피", false, 150, Dish.Type.MEAT),
        new Dish("하즈", false, 300, Dish.Type.FISH),
        new Dish("프로도", false, 301, Dish.Type.MEAT),
        new Dish("빼그만", true, 500, Dish.Type.OTHER),
        new Dish("uhanuu", true, 700, Dish.Type.OTHER)
);

• 요리 목록

 

TAKEWHILE 활용

위의 요리목록을 보면 칼로리를 오름차순으로 정렬되어 있다.

List<Dish> filteredMenu = menu.stream().filter(dish -> dish.getCalories() > 320)
                                       .collect(toList());

• takeWhile(X)

takeWhile 연산을 이용하면 320칼로리 보다 크거나 같은 요리가 나왔을 때 반복 작업을 중단할 수 있다.

• 데이터⬆️ 차이⬆️

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

• takeWhile(O)

 500을 보는 순간 반복 작업이 중단된 것을 볼 수 있었다.

👻 takeWhile을 이용하면 무한 스트림을 포함한 모든 스트림에 Predicate를 적용해 스트림을 슬라이스할 수 있다.!!

 

DROPWHILE 활용

320칼로리 보다 큰 요소 탐색

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

filter는 takeWhile과 동일하게 동작하지만..!

결과는 takeWhile의 반대의 요소 즉 320칼로리 보다 크거나 같은 요소들인 것을 확인할 수 있다.

dropWhile은 Predicate가 처음으로 거짓이 되는 지점까지 발견된 요소를 버리고 takeWhile은 처음으로 거짓이 발견되는 지점 바로 전까지 요소를 가진다.

• (TAKEWHILE(Predicate: true) ↔️ DROPWHILE(Predicate: false)) : 정반대의 작업을 수행한다.

 

스트림 소개에서 스트림을 축소하는 limit을 공부했기 때문에 넘어간다.

요소 건너뛰기

처음 (n)개 요소를 제외한 스트림 반환 ➡️ skip(n)

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

n개 이하의 요소를 포함하는 스트림에 skip(n)을 호출하면 빈 스트림이 반환된다.

• 혹시라도 menu의 개수보다 skip의 개수가 크거나 같으면 최적화 해주려나 했는데 전부 filter를 수행한다.
• 저번 스트림 소개에서 limit은 더 이상 진행하지 않는것을 알 수 있었다. (short circuit)

List<Dish> dishes = menu.stream()
                        .filter(dish -> {
                            System.out.println(dish.getCalories());
                            return dish.getCalories() > 110;
                        })
                        .skip(1)
                        .limit(2)
                        .collect(toList());

 

 

limit과 skip은 상호 보완적인 연산을 수행하는데 순서가 바뀌면 어떻게 될까?

 

skip -> limit

• skip(1)으로 150 넘기고 limit(2)개 질의를 보고 (short circuit) 300과, 301요소만 남게 된다.

limit -> skip

• limit(2)개 질의를 보고 (short circuit) 150과, 300만 남았는데 skip(1)으로 150 넘기고 300만 남게된다.

---

5.3 매핑

특정 객체에서 특정 데이터를 선택하는 작업 ➡️ map,flatMap

 

스트림의 각 요소에 함수 적용하기

기존의 값을 고친다(modify) 보다는 새로운 버전을 만든다 개념에 가까워 변환(transforming)에 가까운 매핑(mapping)이라는 단어를 사용한다.

List<String> dishNames = menu.stream()
                           .map(Dish::getName)
                           .collect(toList());

List<Dish>  ➡️  List<String> (getName은 문자열 반환)

 

각 요리명의 길이를 알기 위해 다른 map메서드를 연결(chaining)할 수 있다.

List<Integer> dishNameLengths = menu.stream() //stream<Dish>
                                    .map(Dish::getName) //stream<String>
                                    .map(String::length) //stream<Integer>
                                    .collect(toList());

 

스트림 평면화

map을 이용해서 리스트에서 고유 문자로 이루어진 리스트를 반환하기(List<String>)

["Hello","world"] ➡️ ["H", "e", "l", "o", "w", "r", "d"] 

List<String> words = List.of("Hello", "World");
List<Stream<String>> uniqueCharacters = words.stream()//Stream<String>
                                    .map(str -> str.split(""))//Stream<String[]>
                                    .map(Arrays::stream)//Stream<Stream<String>>
                                    .distinct()
                                    .collect(toList());

Stream<String[]> 이부분을 Arrays.Stream을 통해서 각 배열을 받아 Stream으로 어떻게든 처리를 하려고 했지만...

public static <T> Stream<T> stream(T[] array) {
    return stream(array, 0, array.length);
}

반환값이 List<Stream<String>>로 원하는 List<String>이 아니다

 

flatMap을 사용하면 해결할 수 있다. (1:다 반환값일 때 String -> String[], Object -> Object[])

쉽게 생각하면 Stream<Stream<Obejct[]>> 이런 형태로 반환이 될 때 Stream<Object>로 평탄화

List<String> uniqueCharacters = words.stream()//Stream<String>
                                      .map(str -> str.split(""))//Stream<String[]>
                                      .flatMap(Arrays::stream)//Stream<String>
                                      .distinct()
                                      .collect(toList()); //List<String>

List<int[]> collect = numbers1.stream()
        .flatMap(i -> numbers2.stream().map(j -> new int[]{i, j})) //map사용하면 stream<stream<int[]>>
        .collect(toList());

map과 flatMap을 봐보자

• mapper is a stateless function 참조

<R> Stream<R> map(Function<? super T, ? extends R> mapper);

 

<R> Stream<R> flatMap(Function<? super T, ? extends Stream<? extends R>> mapper);

 

둘다 새로운 Stream을 반환하지만 flatMap은 스트림의 각 값을 다른 스트림으로 만든 다음에 모든 스트림에 하나의 스트림으로 연결하는 기능을 수행한다. 

 

---

5.4 검색과 매칭

특정 속성이 데이터 집합에 있는지 여부를 검색하는 데이터 처리 ➡️ allMatch, anyMatch, noneMatch, findFirst, findAny

 

Predicate가 적어도 한 요소와 일치하는지 확인

if (menu.stream().anyMatch(Dish::isVegetarian)){
    System.out.println("이 메뉴는 채식주의자를 위한 요리입니다!");
}

anyMatch를 이용하며 불리언을 반환하는 최종연산이다. 

 

Predicate가 모든 요소와 일치하는지 확인

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

allMatch를 이용하며 하나라도 일치하지 않으면 false를 반환한다.

 

NONEMATCH

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

위 예제와 같은 의미이며, allMatch와 반대 연산으로 하나라도 일치하면 false를 반환한다.

 

allMatch, anyMatch, noneMatch 세 메서드는 쇼트서킷 기법으로 java의 &&, ||와 같은 연산을 활용한다. 

---

요소 검색

findAny,findFirst도 쇼트서킷 기법이지만 boolean을 반환하지 않고 Optional을 반환한다.

 

filter와 findAny를 이용해서 채식 요리를 선택할 수 있다.

stream pipe-line은 내부적으로 단일 과정으로 실행할 수 있도록 최적화 되어 쇼트서킷을 이용해서 결과를 찾는 즉시 실행을 종료한다.

public void ifPresent(Consumer<? super T> action) {
  	if (value != null) {
       action.accept(value);
    }
}

menu.stream()
        .filter(Dish::isVegetarian)
        .findAny()
        .ifPresent(dish -> System.out.println(dish.getName()));

Optional<T>를 반환하기 때문에 값이 있을 때만 람다가 실행되도록 NullPointerException을 방지할 수있다.

 

findFirst와 FindAny는 언제 사용할까?

stream은 순서를 보장하기 때문에(list에 대한 순서와 동일) findAny도 findFirst와 동일한 요소가 들어있을 것이다.

👍🏻 둘 중에서 하나만 있으면 될거 같지만 병렬 실행에서 첫 번째 요소를 찾기 위해 findFirst 메서드가 있으며 순서가 상관없다면 병렬 스트림에서는 제약이 적은 findAny를 사용한다.

---

5.5 리듀싱

리듀싱 연산 : 모든 스트림 요소를 처리해서 값으로 도출 (최종 연산)

• 함수형 프로그래밍 언어 용어로는 이 과정이 마치 종이(우리의 스트림)를 작은 조각이 될 때까지 반복해서 접는 것과 비슷하다는 의미로 폴드(fold)라고 부른다. 

List<Integer> numbers = List.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
Integer sum = numbers.stream().reduce(0, (a, b) -> a + b);

 

• reduce에 인자 2개(초기값, BinaryOperator<T>)를 통해 for-each로 인한 반복된 패턴과 람다 표현식을 통해서 추상화 가능하다.

https://yongho1037.tistory.com/703

초기값을 첫 번째 파라미터(a)에 사용하고 스트림에서 4를 소비해 두번 째 파라미터(b)로 사용했다.

이후는 누적값을 a에 다음값을 b에 넣는 식으로 스트림이 하나의 값으로 줄어들 때 까지 반복하여 마지막 요소를 반환한다.

 

초기값 없음

초기값을 받지 않도록 오버라이드 된 reduce도 있지만 이 reduce는 Optional을 반환한다.

Integer integer = numbers.stream().reduce(0, Integer::sum); //a+b 메서드 참조
Optional<Integer> optional = numbers.stream().reduce(Integer::sum);

스트림에 아무 요소가 없다면(numbers.isEmpty() == true) 초기값이 없어 reduce는 합계를 반환할 수 없다. (있으면 초기값 반환)

 

스트림의 모든 요소를 소비할 때까지 람다를 반복 수행하면서 최대값, 최소값을 생산할 수 있다.

Optional<Integer> max = numbers.stream().reduce(Integer::max);//(x,y)->x<y?y:x
Optional<Integer> min = numbers.stream().reduce(Integer::min);//(x,y)->x<y?x:y

 

스트림의 특정 요소의 개수를 구할 수 있다.

long count1 = numbers.stream().count();
Integer count2 = numbers.stream()
        .map(i -> 1)
        .reduce(0, Integer::sum);

 

 

map과 reduce를 연결하는 기법을 맵 리듀스(map-reduce)패턴 이라고 한다.

• 쉽게 병렬화하는 특정 덕분에 구글이 웹 검색에 적용하면서 유명해진 패턴

 

reduce 메서드의 장점과 병렬화

reduce를 이용하면 내부 반복이 추상화되면서 내부 구현에서 병렬로 reduce를 실행할 수 있게된다.

• 반복적인 합계에서는 sum 변수를 공유해야 하므로 쉽게 병렬화하기 어렵다.

병렬화를 위해서는 입력을 분할 하고, 분할된 입력을 더한 다음에, 더한 값을 합쳐야 한다.

• 강제적으로 동기화시킨다 하더라도 결국 병렬화로 얻어야 할 이득이 스레드 간의 소모적인 경쟁 때문에 상쇄되어 버린다.

가변 누적자 패턴(mutable accumulator pattern) ➡️ 우리가 sum 변수 초기화해서 누적하는 방법

• 병렬화와 거리가 너무 먼 기법이다.

컬렉션으로 스트림을 만드는 stream 메서드를 parallelStream로 바꾸는 것만으로도 병렬성을 얻을 수 있다.

Integer integer = numbers.parallelStream().reduce(0, Integer::sum);

• 제약 조건: reduce에 넘겨준 람다의 상태(인스턴스 변수 같은)가 바뀌지 말아야 하며, 연산이 어떤 순서로 실행되더라도 결과가 바뀌지 않는 구조여야 한다.

 

스트림 연산 : 상태 없음과 상태 있음

각각의 연산은 다양한 연산을 수행하기 때문에 내부적인 상태를 고려해야 한다.

 

map, filter 등은 입력 스트림에서 각 요소를 받아 0 또는 결과를 출력 스트림으로 보낸다.

• 사용자가 제공한 람다나 메서드 레퍼런스가 내부적인 가변 상태를 갖지 않는 다는 가정 하에 내부 상태(stateless operation)를 갖지 않는 연산이다.

reduce, sum, max 같은 연산은 결과를 누적할 내부 상태가 필요하다.

• 스트림에서 처리하는 요소 수와 관계 없이 내부 상태의 크기는 한정(bounded) 되어 있다.
  • int나 double과 같은 내부 상태 사용.

sorted나 distinct 같은 연산은 filter나 map과는 달리 스트림의 요소를 정렬하거나 중복을 제거하려면 과거의 이력을 알고 있어야 한다.

• 어떤 요소를 출력 스트림으로 추가하려면 모든 요소가 버퍼에 추가되어 있어야 한다.
• 연산을 수행하는 데 필요한 저장소 크기는 한정(bounded)되어 있지 않다.
  • 데이터 스트림의 크기가 크거나 무한이라면 문제가 생길 수 있다. (쇼트 서킷을 이용해야 되는걸까..?)
• 이러한 연산을 내부 상태를 갖는 연산(stateful operation)이라 한다.

---

5.7 숫자형 스트림

이 값을 int로 반환하고 싶다면 박싱 비용을 피할 수 없고 .reduce(Integer::sum) 보다는 .sum 이런식으로 질의를 하고 싶다.

Optional<Integer> calories = menu.stream() //Stream<Dish>
                                 .map(Dish::getCalories) //Stream<Integer>
                                 .reduce(Integer::sum); //Optional<Integer>

스트림의 요소가 Integer임에도 다른 요소가 올 수 있기 때문에 인터페이스에서 sum 메서드를 제공하지 않는다.

숫자 스트림을 효율적으로 처리할 수 있도록 기본형 특화 스트림(primitive stream specialization)을 제공한다.

 

각각의 요소에 특화된 IntStream, DoubleStream, LongStream을 제공한다.

• 각각의 인터페이스는 sum,max,min 같이 자주 사용하는 숫자 관련 리듀싱 연산 수행 메서드를 제공한다.
• 기본형 특화 스트림 ➡️ 객체 스트림으로 복원하는 기능 boxed()메서드도 제공한다.

 

숫자 스트림 ↔️ 객체 스트림

스트림을 특화 스트림으로 변환할 때는 mapToInt, mapToDouble, mapToLong 세 가지 메서드를 가장 많이 사용한다.

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

• 객체 스트림 Stream<Dish> ➡️ IntStream

boxed 메서드를 통해서 기본형 특화 스트림 ➡️ 객체 스트림으로 복원

IntStream map(IntUnaryOperator mapper);

intStream의 map 연산은 int를 인수로 받아서 int를 반환하는 람다 IntUnaryOperator을 받는다.

@FunctionalInterface
public interface IntUnaryOperator {
    int applyAsInt(int operand);
    ...
}

정수가 아닌 다른 객체타입의 값을 반환하기 위해 다시 일반스트림으로 변환해야 한다.

 

기본형 특화 스트림 ➡️ 다른 특화 스트림으로 변경 가능하다.

 

기본형 특화 스트림을 사용하면 sum, average, max, min등을 이용할 수 있는데 sum은 스트림이 비어있으면 기본값 0을 반환한다.

int sum = menu.stream()
              .mapToInt(Dish::getCalories) //IntStream
              .sum();

OptionalInt, OptionalDouble, OptionalLong

 

 

OptionalDouble average = menu.stream()
                      .mapToInt(Dish::getCalories) //IntStream
                      .average();

OptionalLong min1 = menu.stream()
                       .mapToLong(Dish::getCalories)
                       .min();
                       
OptionalInt min2 = menu.stream()
                      .mapToInt(Dish::getCalories) //IntStream
                      .min();

 

• 합계(sum) 같은 경우는 0이라는 기본값이 있었으면 문제가 발생하지 않지만 최대값, 최소값, 평균값 같은 경우는 기본값을 도출할 수 없다.

Optional를 이용해서 최대값이 없는 상황에 사용할 기본값을 명시적으로 정의할 수 있다.

int max = menu.stream()
                .mapToInt(Dish::getCalories)
                .max()
                .orElse(1);

 

숫자 범위

자바 8의 IntStream과 LongStream에서는 range와 rangeClosed라는 두가지 정적 메서드를 제공한다.

IntStream range = IntStream.range(1, 100)
                            .filter(n -> n%2==0); //49개 (1~99)
IntStream rangeClose = IntStream.rangeClosed(1, 100)
                            .filter(n -> n%2==0); //50개 (1~100)

두 메서드 둘다 첫 번째 인수로 시작값, 두 번째 인수로 종료값을 가지나 rangeClosed는 종료값이 결과에 포함이 된다.

 

---

5.8 스트림 만들기

 

값으로 스트림 만들기

Stream의 static 메서드를 이용해서 스트림을 만들 수 있다.

Stream<String> stream = Stream.of("Modern", "Java", "In", "Action");
Stream<Object> empty = Stream.empty(); //비어있는 스트림

• 값을 채울수도 비어있는 스트림을 만들 수도 있다.

 

null이 될 수 있는 객체로 스트림 만들기

Stream의 ofNullable을 이용해서 null이라면 비어있는 스트림을 값이 있다면 값을 채운 stream을 반환할 수 있다.

public static<T> Stream<T> ofNullable(T t) {
    return t == null ? Stream.empty()
                     : StreamSupport.stream(new Streams.StreamBuilderImpl<>(t), false);
}

ofNullable에서 값이 있어도 없어도 Stream을 반환하기 때문에 flatMap을 이용해서 평탄화하여 유용하게 사용할 수 있다.

HashMap<String, Object> map = new HashMap<>();
List<Object> value = Stream.of("key")
                           .flatMap(s -> Stream.ofNullable(map.get(s)))
                           .collect(toList());

 

배열로 스트림 만들기

int sum = Arrays.stream(new int[]{1, 2, 3, 4, 5}).sum(); //15

파일로 스트림 만들기

java.nio.file.Files의 많은 정적 메서드가 스트림을 반환한다.

long uniqueWords = 0;
try (Stream<String> lines = Files.lines(Paths.get("data.txt"), StandardCharsets.UTF_8)) {
    uniqueWords = lines.flatMap(line -> Arrays.stream(line.split(" ")))
            .distinct()
            .count();
}

• Stream 인터페이트가 AutoCloseable 인터페이스를 구현해 try-with-resource를 활용할 수 있다.

public interface Stream<T> extends BaseStream<T, Stream<T>> {
	...
}

public interface Stream<T> extends BaseStream<T, Stream<T>> {
	...
     /**
     * Closes this stream, causing all close handlers for this stream pipeline
     * to be called.
     *
     * @see AutoCloseable#close()
     */
    @Override
    void close();
}

---

함수로 무한 스트림 만들기

Stream API는 두 개의 정적메서드 Iterate, generate를 이용해 크기가 고정되지 않은 무한 스트림(infinite stream)을 만들 수 있다.

• 두 메서드를 이용해서 만든 스트림은 요청할 때마다 주어진 함수를 이용해서 값을 만든다. 
• 요청할 때마다 값을 생산하는 끝이 없는 스트림을 언바운드 스트림(unbounded stream)이라고 표현한다.
• 무한한 값을 출력하지 않도록 limit(n) 함수와 같이 사용해야 한다.
• 무한 스트림의 요소는 무한적으로 계산이 반복되므로 정렬하거나 리듀스할 수 없다. (파이프라인으로 들어오기전에 limit 사용)

iterate

첫 번째 인자에 seed값인 초기값을 넣어주어야 한다.

Stream.iterate(0,n -> n +2)
        .limit(10)
        .forEach(System.out::println);
        
        
//파보나치수열
Stream.iterate(new int[]{0,1}, t-> new int[]{t[1],t[0]+t[1]})
                .limit(10)
                .map(t -> t[0])
                .forEach(System.out::println);

자바 9의 iterate 메서드는 Predicate를 지원한다.

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

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

• filter를 사용하면 언제 종료해야 될지 모르기 때문에 스트림 쇼트서킷을 지원하는 takeWhle, dropWhile을 이용해야 한다.

generate

iterate와 다르게 generate는 생산된 각 값을 연속적으로 계산하지 않는다.

Stream.generate(() -> 1)
        .limit(10)
        .forEach(System.out::println); //1을 10번 출력

• generate는 인자로 Supplier<? extends T> s를 받는다.  ()->T
•  Math.random()와 같은 발행자(supplier)도 상태가 없는 메서드로 나중에 계산에 사용할 어떤 값도 저장하지 않는다.

generate를 이용해서 파보나치수열을 작성해보자.

• 이전 값을 알기 위해서는 발행자에 상태가 있어야 될거 같다.

 

박싱 연산 문제를 피하기 위해서 IntStream을 사용하자

 

익명 클래스는 람다와 다르게 previous, current처럼 가변(mutable)상태 객체를 이용해 getAsInt 메서드의 연산을 커스터마이즈 할 수 있는 상태 필드를 정의할 수 있다.

• 람다는 상태를 바꾸지 않는다.

IntSupplier fib = new IntSupplier() {
    private int previous = 0;
    private int current = 1;

    @Override
    public int getAsInt() {
        int oldPrevious = this.previous;
        int nextValue = this.previous + this.current;
        this.previous = this.current;
        this.current = nextValue;
        return oldPrevious;
    }
};
IntStream.generate(fib).limit(10).forEach(System.out::println);

클로저를 이용해서 작성할 수 있긴 할것이다.

public class Main {
    private int previous = 0;
    private int current = 1;

    public void fib(){
        IntStream.generate(() -> {
                    int oldPrevious = previous;
                    int nextValue = previous + current;
                    previous = current;
                    current = nextValue;
                    return oldPrevious;
                })
                .limit(10)
                .forEach(System.out::println);
    }

iterate를 사용했을 때는 각 과정에서 새로운 값을 생성하면서도 기존 상태를 바꾸지 않는 순수한 불변(immutable) 상태를 유지했다.

 

스트림을 병렬로 처리하면서 올바른 결과를 얻기 위해서는 불변 상태 기법을 고수해야 한다.

• 병렬 코드에서는 발행자에 상태가 있으면 안전하지 않다.

---