CS | Stream에 대하여

나경호·2023년 8월 7일

Backend CS 스터디

CS 스터디 정리

목록 보기

2/2

Stream에 대해서 설명해주세요.

스트림(Stream)

자바8에 새롭게 추가된 기능으로, 선언형으로 데이터(컬렉션, 배열, 파일, iterate...)를 처리할 수 있습니다.
Stream을 사용하면 데이터를 쉽게 필터링, 변환, 집계할 수 있습니다.
Stream은 병렬처리가 가능하도록 설계되었으므로 멀티 코어 프로세서를 활용하여 처리 속도를 높일 수 있습니다.

Stream의 특징

Stream은 데이터 구조가 아닙니다.

Stream은 데이터를 저장하지 않습니다.
Stream에서 요소를 추가하거나 제거할 수 없습니다.

Stream은 생성, 중간, 최종 작업으로 나뉩니다.

아래에서 더 자세히 알아보겠지만, 대부분의 Stream 작업은 또 다른 새 Stream을 반환하며 함께 연결되어 작업 파이프 라인을 형성합니다.
Stream 자체를 반환하는 작업을 중간 작업(Intermediate Operations)이라 합니다.(예: filter(), distinct(), sorted() 등)
Stream 이외의 것을 반환하는 작업을 최종 작업(Terminal Operations)이라고 합니다.(예: count(), min(), max(), collect() 등)

병렬처리(Parallelism)

병렬처리는 멀티 코어에서 멀티 쓰레드를 동작시키는 방식입니다.
한 개 이상의 쓰레드를 포함하는 각 코어들이 동시에 실행되는 성질을 말합니다.
많은 양의 데이터를 처리하면서 작업 속력을 높이기 위해서는 병렬로 처리하고 멀티 코어 아키텍처를 사용해야 합니다.
Stream은 다중 스레드 코드를 작성하지 않고 병렬로 처리할 수 있습니다.

병렬 스트림

병렬 스트림이란, 각각의 스레드에서 처리할 수 있도록 스트림 요소를 여러 청크로 분할한 스트림입니다.(컬렉션에 parallelStream을 호출하기만 하면 병렬 스트림이 생성됩니다.)
병렬 스트림은 전체 데이터를 서브 데이터들로 나눈 후 서브 데이터들을 병렬 처리하여 작업을 빠르게 수행하는 것을 구현한 것입니다.

💡 그럼 모든 Stream을 병렬 스트림으로 사용하면 되는가?

여기서 주의할 점은 모든 병렬 Stream이 동일한 ThreadPool에서 thread를 가져와 사용한다는 것입니다.
Tread Pool을 이미 모두 점유하고 있다면 더이상 요청이 처리되지 않는 문제가 발생할 수 있습니다.
이러한 문제는 ForkJoinPool을 커스텀하게 제작함으로써 해결할 수 있습니다.
- Fork/Join Framework : Java 7에 추가된 병렬 Stream의 내부 로직입니다.
- Fork / Join Framework은 작업을 분할가능할 만큼 쪼개고, 쪼개진 작업을 별도의 work thread를 통해 작업 후 결과를 합치는 과정을 거쳐 결과를 만들어냅니다.

Lazy Evaluation

직역하면 "게으른 연산"으로, 불필요한 연산을 피하기 위해 연산을 지연시키는 것을 말합니다.
필요할 때만 Stream의 요소를 평가하여 메모리 사용량을 줄이고 성능을 향상시킬 수 있습니다.

Stream은 한 번만 통과할 수 있습니다.

Stream은 두 번 이상 탐색할 수 없습니다.

List<String> nameList = Arrays.asList("Dinesh", "Ross", "Kagiso", "Steyn");          
Stream<String> stream = nameList.stream();
stream.forEach(System.out::println);
stream.forEach(System.out::println);    
//Exception in thread "main" java.lang.IllegalStateException: stream has already been operated upon or closed

Stream은 내부 반복 방식으로 작동합니다.

Stream은 내부 반복을 사용하면서 작업을 병렬적으로 처리하거나 최적화된 다양한 순서로 처리가 가능합니다.
반면에 for 문과 같은 외부 반복은 병렬성을 스스로 관리(synchronized 키워드 사용)해야 합니다.

Stream 3가지 단계

생성하기: 배열, 컬렉션, 임의의 수, 파일 등 거의 모든 것을 가지고 Stream을 생성할 수 있습니다.
가공하기: 필터링(filtering) 및 맵핑(mapping) 등 원하는 결과를 만들어가는 중간 작업(intermediate operations)을 말합니다.
결과 만들기: 최종적으로 결과를 만들어내는 작업(terminal operations)입니다.

Stream 생성

Stream.of()로 Stream 생성 : 특정 객체를 요소로 갖는 Stream을 생성하고 싶을 때 Stream.of() 를 사용할 수 있습니다.
```
Stream<String> stream = Stream.of("code", "chacha", "blog", "example");
```
Stream.empty()로 비어있는 Stream 생성 : Stream.empty() 는 어떤 요소도 갖고 있지 않는 Stream 객체를 생성합니다.
```
Stream<String> stream = Stream.empty();
```
컬렉션 타입(Collection, List, Set) Stream 생성 : 컬렉션 타입의 경우 인터페이스에 추가된 default 메서드 stream을 통해 스트림을 생성합니다.
```
List<String> list = Arrays.asList("a1", "a2", "b1", "b2", "c2", "c1");
Stream<String> stream = list.stream();
```

Arrays.stream()으로 Stream 생성 : Arrays.stream() 을 이용하여 Stream을 생성할 수 있습니다.

String[] array = new String[]{"a1", "a2", "b1", "b2", "c2", "c1"};
Stream<String> stream = Arrays.stream(array);

Stream.generate()로 Stream 생성
- generate 메소드를 이용하면 Supplier<T> 에 해당하는 람다로 값을 넣을 수 있습니다.
- Supplier : 인자는 없고 리턴값만 있는 함수형 인터페이스.
```
public static<T> Stream<T> generate(Supplier<T> s) { ... }
```
- 이 때 생성되는 스트림은 크기가 정해져있지 않고 무한하기 때문에 특정 사이즈로 최대 크기를 제한해야 합니다.
```
Stream<String> generatedStream = Stream.generate(() -> "gen").limit(5); // 5개의 “gen” 이 들어간 스트림이 생성됩니다.
```
Stream.iterate()로 Stream 생성
- Stream.iterate() 도 generate() 와 유사합니다. 하지만 Stream.iterate() 은 두 개의 인자를 받습니다. 첫번째 인자는 초기값, 두번째 인자는 함수입니다. 이 함수는 1개의 인자를 받고 리턴 값이 있습니다.
```
Stream<Integer> stream = Stream.iterate(0, n -> n + 2).limit(5);
```

Stream 중간 연산

filter(Predicate\<T>) : Predicate를 인자로 받아 true인 요소를 포함한 스트림 반환
- Predicate\<T>: T에 대한 조건에 대해서 true/false를 반환하는 Functional Interface
distinct() : 중복 값을 필터링

코드 예제

Stream<String> stream2 = Stream.of("1", "1", "1", "2", "3", "4");
stream2.distinct().filter(i -> i.equals("4")).forEach(System.out::print); //4

limit(n) : 주어진 사이즈 이하 크기를 갖는 스트림 반환
skip(n) : 처음 요소 n개 제외한 스트림 반환

코드 예제

Stream<String> stream1 = Stream.of("1", "2", "3", "4", "5", "6","7","8","9","10");
stream1.skip(3).limit(5).forEach(System.out::print); //45678

map(Function) : 매핑 함수의 result로 구성된 스트림 반환

코드 예제

Member[] members = {
	new Member("김"),
	new Member("최"),
	new Member("박"),
	new Member("이")
};

Stream<Member> memberStream = Stream.of(members);

memberStream.map(Member::getLastName)
	.forEach(System.out::print); //김최박이
}

flatMap() :매핑된 각 배열을 스트림이 아니라 스트림의 콘텐츠로 매핑함. map과 달리 평면화(1차원)된 스트림 반환

코드 예제(map과 flatMap 차이 비교)

flatMap은 1차원 각각의 요소가, map은 stream 자체가 호출되는 것을 알 수 있습니다.

Stream<String[]> stream1 = Stream.of(
		new String[]{"Python", "Java", "C"},
		new String[]{"PHP", "JavaScript", "Kotlin"}
);
Stream<String[]> stream2 = Stream.of(
		new String[]{"Python", "Java", "C"},
		new String[]{"PHP", "JavaScript", "Kotlin"}
);

System.out.println("flatMap 예시");
stream1.flatMap(Arrays::stream)
		.forEach(System.out::println);

System.out.println("map 예시");
stream2.map(Arrays::stream)
		.forEach(System.out::println);
}
/**
 * flatMap 예시
 * Python
 * Java
 * C
 * PHP
 * JavaScript
 * Kotlin
 * 
 * map 예시
 * java.util.stream.ReferencePipeline$Head@4517d9a3
 * java.util.stream.ReferencePipeline$Head@372f7a8d
 */

중간 연산은 모두 스트림을 반환한다.

Stream 최종 연산

(boolean) allMatch(Predicate) : 모든 스트림 요소가 Predicate와 일치하는지 검사
(boolean) anyMatch(Predicate) : 하나라도 일치하는 요소가 있는지 검사
(boolean) noneMatch(Predicate) : 매치되는 요소가 없는지 검사

코드 예제

IntStream stream1 = IntStream.of(30, 90, 70, 10);
IntStream stream2 = IntStream.of(30, 90, 70, 10);
IntStream stream3 = IntStream.of(30, 90, 70, 10);

System.out.println(stream1.allMatch(n -> n > 80)); // false
System.out.println(stream2.anyMatch(n -> n > 80)); // true
System.out.println(stream3.noneMatch(n -> n > 80)); // false

(Optional) findAny() : 현재 스트림에서 임의의 요소 반환
(Optional) findFirst() : 스트림의 첫번째 요소

코드 예제

IntStream stream1 = IntStream.of(30, 90, 70, 10);
IntStream stream2 = IntStream.of(30, 90, 70, 10);

System.out.println(stream1.findAny().getAsInt()); // 30
System.out.println(stream2.findFirst().getAsInt()); // 30

reduce() : 모든 스트림 요소를 처리해 값을 도출. 두 개의 인자를 가짐

코드 예제

Stream<String> stream1 = Stream.of("일","이","삼","사");
Stream<String> stream2 = Stream.of("일","이","삼","사");
Optional<String> result1 = stream1.reduce((a1, a2) -> a1 + " + " + a2); 
String result2 = stream2.reduce("영",(a1,a2) ->  a1 + " + " + a2);

System.out.println(result1.get()); // ((일 + 이) + 삼) + 사
System.out.println(result2); // 영 + 일 + 이 + 삼 + 사

collect() : 스트림을 reduce하여 list, map, 정수 형식 컬렉션을 만듬

Stream<String> stream = Stream.of("넷", "둘", "하나", "셋");
List<String> list = stream.collect(Collectors.toList());

System.out.println(list); //[넷, 둘, 하나, 셋]

(void) forEach() : 스트림 각 요소를 소비하며 람다 적용
(Long) count : 스트림 요소 개수 반환

Optional 클래스란?

Optional 클래스는 값의 존재나 여부를 표현하는 컨테이너 클래스입니다.

optional을 이용해 null확인 관련 버그를 피할 수 있습니다.
Optional은 값이 존재하는지 확인하고싶거나, 값이 없을 때 어떻게 처리할 것인지 강제하는 기능을 제공합니다.
isPresent() 는 Optional이 값을 포함하면 참을 반환합니다.

스트림 사용 시 주의사항

스트림을 사용할 때 흔히하는 실수

스트림 재사용

스트림은 한 번만 사용할 수 있습니다.

"무한" 스트림 생성

iterate나 generate를 이용하여 스트림을 생성할 때 제한을 두지 않으면 무한 스트림이 생성됩니다.

또는 코드의 의도에 맞지 않는 순서로 스트림을 작성할 경우에도 무한 스트림이 생성될 수 있습니다.

// 의도치 않게 생성된 무한 스트림
IntStream.iterate(0, i -> ( i + 1 ) % 2)
         .distinct()
         .limit(10)
         .forEach(System.out::println);

System.out.println("complete");
// limit(10)을 했으나, distinct가 먼저 실행되어 0과 1만 무한히 반복된다.

과도한 스트림의 사용은 오히려 가독성을 떨어트린다.

스트림을 과용하여 코드 가독성을 떨어트리고 유지 보수 비용을 늘리는 케이스 예시

public class StreamAnagrams {
    public static void main(String[] args) throws IOException {
        Path dictionary = Paths.get(args[0]);
        int minGroupSize = Integer.parseInt(args[1]);

        try (Stream<String> words = Files.lines(dictionary)) {
            words.collect(
                    groupingBy**(word -> word.chars().sorted()
                            .collect(StringBuilder::new,
                                    (sb, c) -> sb.append((char) c),
                                    StringBuilder::append).toString()))**
                    .values().stream()
                    .filter(group -> group.size() >= minGroupSize)
                    **.map(group -> group.size() + ": " + group)**
                    .forEach(System.out::println);
        }
    }
}

사전 하나를 훑어 원소 수가 많은 아나그램(알파벳이 같고 순서만 다른 단어) 그룹들 출력

import static java.util.stream.Collectors.groupingBy;

// 코드 45-3 스트림을 적절히 활용하면 깔끔하고 명료해진다. (Effective Java 271쪽)
public class HybridAnagrams {
    public static void main(String[] args) throws IOException {
				
        Path dictionary = Paths.get(args[0]);
        int minGroupSize = Integer.parseInt(args[1]);
				//파일 내용은 java 문자열 stream 으로 생성
        try (Stream<String> words = Files.lines(dictionary)) {
            words.collect(groupingBy(word -> **alphabetize(word))**) 
                    .values().stream()
                    .filter(group -> group.size() >= minGroupSize)
                    .forEach(g -> System.out.println(g.size() + ": " + g));
        }
    }

    private static String **alphabetize**(String s) {
        char[] a = s.toCharArray();
        Arrays.sort(a);
        return new String(a);
    }
}

질문 정리

Q1. ForkJoinPool을 사용해 커스텀 쓰레드를 만들었을 때 단점은 없을까요? 실제로 병렬처리가 어떤 식으로 일어나는 지도 조사가 된다면 좋을 것 같아요

A1.

ForkJoinPool의 단점으로는
1. ForkJoinPool은 recursive하게 작업되기 때문에 너무 많이 재귀적으로 분할되는 경우, 스레드 간의 작업 스케줄링이 불필요하게 발생하여 성능이 저하될 수 있습니다.
2. ForkJoinPool은 작업이 동시에 실행되기 때문에 작업의 순서를 제어하기 어렵습니다.
3. ForkJoinPool은 작업이 완료될 때까지 기다려야 할 경우, 작업들 간의 동기화 오버헤드가 발생할 수 있습니다.

그러나 포크/조인 프레임워크에서는 작업 훔치기(workstealing) 이라는 기법으로 ForkJoinPool의 모든 스레드를 거의 공정하게 분할한다.
할 일이 없어진 스레드가 유휴 상태로 바뀌는 것이 아니라 다른 스레드 큐의 꼬리에서 작업을 훔쳐온다.

병렬 스트림은 포크/조인 프레임워크를 아래의 그림과 같은 방식으로 활용하게 됩니다. 포크/조인 프레임워크에서는 서브 테스크를 스레드 풀 (ForkJoinPool)의 작업자 스레드에 분산 할당하는 ExecutorService 인터페이스를 구현합니다.

포크/조인 프레임워크는 Spliterator를 통해 데이터를 포함하는 스트림을 병렬화합니다.

병렬 스트림에서 Spliterator를 이용하기 때문에 분할 로직을 개발하지 않고도 분할 기능을 사용할 수 있습니다.

public interface Spliterator<T> {
  boolean tryAdvance(Consumer<? super T> action);
  Spliterator<T> trySplit();
  long estimateSize();
  int characteristics();
}
// T는 Spliterator에서 탐색하는 요소의 형식을 가리킨다.
// tryAdvance 메서드는 Spilterator의 요소를 하나씩 순차적으로 소비하며 탐색해야할 요소가 남아 있으면 True를 리턴한다.
// trySplit 메서드는 Spliterator의 일부 요소를 분할해서 두번째 Spilterator를 생성하는 메서드이다.
// estimateSize로 탐색해야 할 요소의 갯수를 구할수 있다.

Q2. Lazy Evaluation의 구체적인 예시가 함께 있으면 이해하기 좋을 것 같아요.

A2. 1~10까지의 정수를 갖는 List에서 6보다 작고, 짝수인 요소를 찾아 10배 시킨 리스트를 출력하는 코드를 예시로 생각해보면 좋습니다.

final List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
 
System.out.println(
    list.stream()
        .filter(i -> {
            System.out.println("i < 6");
            return i<6;
        })
        .filter(i -> {
            System.out.println("i%2 == 0");
            return i%2==0;
        })
        .map(i -> {
            System.out.println("i = i*10");
            return i*10;
        })
        .collect(Collectors.toList())
);

위의 코드를 실행한 결과는 아래와 같습니다.

i < 6
i%2 == 0
i < 6
i%2 == 0
i = i*10
i < 6
i%2 == 0
i < 6
i%2 == 0
i = i*10
i < 6
i%2 == 0
i < 6
i < 6
i < 6
i < 6
i < 6
[20, 40]

위의 결과로 알 수 있는 점은 Lazy Evaluation은 Eager Evaluation(조급한 연산)과 달리 순차적으로 모두 연산하는 것이 아닌 필요할 때만 Stream의 요소를 평가하여 연산하게 됩니다. 위의 예시를 요약하면

6보다 작은지 검사한다. ( false 이면 2, 3번 과정 패스하고 다음 요소 진행 )
짝수인지 검사한다. ( false 이면 3번 과정 패스하고 다음 요소 진행 )
요소에 10을 곱해준다.

이와 같이 정리할 수 있습니다.

Q3. 내부 반복과 외부 반복이 어떻게 다른지 조금 더 자세히 설명해주면 좋을 것 같아요.

A3.

외부반복: 컬렉션의 요소를 반복문으로 직접 탐색하는 것

예)

List<String> names = new ArrayList<>();

for(Dish dish: menu){
	names.add(dish.getName());
}
//메뉴 리스트를 명시적으로 순차 반복한다.

외부반복은 명시적으로 컬렉션 요소를 하나씩 가져와 처리합니다.

내부반복: 람다 표현식을 인수로 받아, 어떤 작업을 수행할지만 지정하면 모든 것이 알아서 처리됨.

예)

List<String> names = menu.stream()
	.map(Dish::getName)
	.collect(toList());
    
//메뉴리스트를 내부적으로 순차 반복한다

내부반복은 작업을 투명하게 병렬로 처리하거나, 더 최적화된 다양한 순서로 내부로직을 통해 알아서 처리해줍니다.

Q4. 빈 Stream 객체를 어떤 경우에 사용되는지 궁금합니다.

A4. 제가 찾아본 바로는 실무에서 데이터베이스에서 조회한 결과를 스트림으로 처리하는 상황에 사용할 수 있다고 합니다. 예를 들어, 데이터베이스 테이블에서 특정 조건을 만족하는 데이터를 조회하여 스트림으로 처리하고 싶을 때 빈 스트림을 활용할 수 있다고 합니다.

Reference

나경호

기억창고👩‍🌾

이전 포스트

CS | interface와 abstract class차이에 대해서 설명해주세요.

1개의 댓글

happy

2023년 8월 7일

개발자로서 배울 점이 많은 글이었습니다. 감사합니다.

답글 달기