Java. 스트림이란?

스트림이란 자바 8부터 다량의 데이터 처리 작업을 돕고자 나온 API 이고, 두 가지 핵심적인 추상개념을 제공한다.

 

스트림은 원하는 결과를 생성하기 위해 파이프라인으로 연결될 수 있는 다양한 메서드를 지원하는 개체시퀀스이다.

 

---

 

 

스트림 파이프라인의 개념

 

1. 데이터 소스

: 스트림은 데이터를 처리하는데 사용할 수 있는 다양한 소스로부터 생성됩니다. 이 데이터 소스는 컬렉션, 배열, 파일 등 다양한 형태 일 수 있다.

 

2. 연속된 작업

: 스트림은 중간 연산과 최종 연산으로 구성된다. 중간 연산은 스트림을 반환하며, 여러 중간 연산이 연결될 수 있다. 최종 연산은 스트림 파이프라인을 종료하고 결과를 생성한다.

 

3. 파이프라인

: 스트림에서 제공하는 중간 연산과 최종 연산을 연결하여 데이터를 처리하는 구조를 파이프라인 이라고 한다.

 

4. 지연 연산

: 스트림은 지연 연산을 지원한다. 중간 연산들은 실제로 수행되기 전까지는 스트림 파이프라인이 실행되지 않는다. 이는 필요한 연산만 수행하여 최적의 성능을 얻을 수 있도록 한다. 최종 연산이 호출되면, 그 때 중간 연산들이 수행되며, 최종 결과를 생성한다. 이러한 특징 때문에 무한한 데이터 스트림을 메모리에 로드하지 않고 효율적으로 다룰 수 있다.

 

5. 순차 및 병렬 처리

: 스트림은 순차적으로도 처리될 수 있고, 병렬로도 처리될 수 있다. 병렬 스트림을 사용하면 멀티코어를 활용하여 성능을 향상시킬 수 있다.

 

 

---

 

 

스트림 파이프라인은 소스 스트림에서 시작해 종단 연산으로 끝나고, 그 사이에 중간 연산이 있을 수 있다.

 

1. 중간 연산 ( 지연 연산 )

 

각 중간 연산은 스트림을 어떠한 방식으로 변환하는 역할을 수행하고,

 

한 스트림을 다른 스트림으로 변환하게 하여 메서드 체이닝 ( 순차적 호출 ) 을 가능하게 한다. 

 

예를 들어, 각 원소에 함수를 적용하거나 특정 조건을 만족 못하는 원소를 걸러낼 수 있다.

 

이러한 기법으로 대용량 데이터를 다룰 때, 실제로 필요하지 않는 데이터들을 탐색하는 것을 방지해 속도를 높일 수 있다.

 

즉, 종단 연산에 쓰이지 않는 데이터 원소는 계산 자체에 쓰이지 않는다. 이것을 Short-Circuit 방식이라고 한다.

 

 

- filter(Predicate<T> predicate)

: 주어진 조건에 맞는 요소만 선택하여 새로운 스트림을 생성한다. 주어진 Predicate 함수가 true 를 반환하는 요소만 포함된다.

List<String> fruits = Arrays.asList("apple", "orange", "banana", "kiwi");

List<String> result = fruits.stream()
                            .filter(fruit -> fruit.startsWith("a"))
                            .collect(Collectors.toList());

System.out.println(result); // Output: [apple]

 

 

 

 

 

- map(Function<T, R> mapper)

: 각 요소에 주어진 함수를 적용하여 새로운 값을 생성한다.

List<String> fruits = Arrays.asList("apple", "orange", "banana", "kiwi");

List<Integer> result = fruits.stream()
                            .map(String::length)
                            .collect(Collectors.toList());

// Output: [5, 6, 6, 4]

 

 

 

 

 

- flatMap(Function<T, Stream<R>> mapper)

: 각 요소에 대해 주어진 함수를 적용하여 생성된 스트림을 편면화하여 새로운 스트림을 생성한다. 주로 중첩된 리스트를 단일 리스토로 평면화 할 때 사용된다.

List<List<Integer>> numbers = Arrays.asList(
    Arrays.asList(1, 2),
    Arrays.asList(3, 4),
    Arrays.asList(5, 6)
);

List<Integer> result = numbers.stream()
                              .flatMap(List::stream)
                              .collect(Collectors.toList());

// Output: [1, 2, 3, 4, 5, 6]

 

 

 

 

- distinct()

: 스트림의 요소에서 중복을 제거한 새로운 스트림을 생성한다.

List<Integer> numbers = Arrays.asList(1, 2, 2, 3, 4, 4, 5);

List<Integer> result = numbers.stream()
                              .distinct()
                              .collect(Collectors.toList());

// Output: [1, 2, 3, 4, 5]

 

 

 

 

- sorted(), sorted(Comparator<T> comparator)

: 스트림의 요소를 정렬하여 새로운 스트림을 생성한다. 기본적으로 요소의 자연 순서에 따라 정렬되고, sorted(..)를 사용하여 비교자를 지정할 수도 있다.

List<String> fruits = Arrays.asList("apple", "orange", "banana", "kiwi");

List<String> result = fruits.stream()
                            .sorted()
                            .collect(Collectors.toList());

// Output: [apple, banana, kiwi, orange]

 

 

 

 

- peek(Consumer<T> action)

: 각 요소에 대해 주어진 동작을 수행하며, 새로운 스트림을 생성한다. 주로 디버깅이나 로깅용으로 사용된다.

List<String> fruits = Arrays.asList("apple", "orange", "banana", "kiwi");

List<String> result = fruits.stream()
                            .peek(fruit -> System.out.println("Processing: " + fruit))
                            .collect(Collectors.toList());

// Output:
// Processing: apple
// Processing: orange
// Processing: banana
// Processing: kiwi

 

예외 사항 : Stateful operations 

 

이러한 특성으로 인해 무한 스트림을 다룰 수 있게 된다.

 

무한 스트림은 애초에 크기가 정해져 있지 않은 만큼, 중복 제거를 할 수 없지만

 

limit()와 같은 short-circuit 연산을 통해 유한 스트림으로 변환 할 수 있다.

 

중복을 제거하는 distinct() 나 전체 데이터를 정렬하는 sort() 연산들을 Stateful 연산이라 한다.

 

이는 지연 연산을 무효화시키고, 결과를 생성하기 전에 전체 데이터를 탐색해버리는 결과를 초래하기 때문에 사용에 주의해야 한다.

 

 

 

 

 

 

2. 최종 연산

 

스트림 파이프라인을 종료하고 최종 결가를 생성하는 역할을 한다. 

 

최종 연산이 호출되기 전까지는 중간 연산이 실행되지 않는다. 

 

 

 

List<String> fruits = Arrays.asList("apple", "orange", "banana", "kiwi");

 

 

 

1. forEach(Consumer<T> action) 

: 각 요소에 대해 주어진 동작을 수행한다. 병렬 스트림에서는 각 요소에 대해 동시에 동작할 수 있다.

        // 최종 연산: forEach
        fruits.stream()
              .forEach(System.out::println);
        // Output: apple, orange, banana, kiwi

 

 

 

 

2. collect(Collector<T, A, R> collector)

: 스트림의 요소를 수집하여 지정된 컬렉터에 따라 최종 결과를 생성한다.

        // 최종 연산: collect
        List<String> resultCollect = fruits.stream()
                                          .filter(fruit -> fruit.startsWith("a"))
                                          .collect(Collectors.toList());
        System.out.println("Collect Result: " + resultCollect);
        // Output: Collect Result: [apple]

 

 

 

 

3. count()

: 스트림의 요소 개수를 반환한다.

        // 최종 연산: count
        long count = fruits.stream()
                          .filter(fruit -> fruit.startsWith("a"))
                          .count();
        System.out.println("Count Result: " + count);
        // Output: Count Result: 1

 

 

 

 

4. anyMatch(Predicate<T> predicate)

: 주어진 조건을 만족하는 요소가 스트림에 하나라도 있는지 여부를 반환한다.

        // 최종 연산: anyMatch
        boolean anyStartsWithA = fruits.stream()
                                       .anyMatch(fruit -> fruit.startsWith("a"));
        System.out.println("Any Match Result: " + anyStartsWithA);
        // Output: Any Match Result: true

 

 

 

 

5. allMatch(Predicate<T> predicate)

: 모든 요소가 주어진 조건을 만족하는지 여부를 반환한다.

        // 최종 연산: allMatch
        boolean allStartsWithA = fruits.stream()
                                       .allMatch(fruit -> fruit.startsWith("a"));
        System.out.println("All Match Result: " + allStartsWithA);
        // Output: All Match Result: false

 

 

 

 

6.noneMatch(Predicate<T> predicate)

: 모든 요소가 주어진 조건을 만족하지 않는지 여부를 반환한다.

        // 최종 연산: noneMatch
        boolean noneStartsWithZ = fruits.stream()
                                        .noneMatch(fruit -> fruit.startsWith("z"));
        System.out.println("None Match Result: " + noneStartsWithZ);
        // Output: None Match Result: true

 

 

 

 

7. findFirst(), findAny()

: findFirst는 스트림의 첫 번째 요소를 반환하고, findAny는 병렬 스트림에서 불특정 요소를 반환한다.

주로 Optinal로 감싸져 있어 값이 존재하는지 확인할 때 사용된다.

        // 최종 연산: findFirst
        Optional<String> first = fruits.stream()
                                      .findFirst();
        System.out.println("Find First Result: " + first.orElse("No element"));
        // Output: Find First Result: apple

 

 

 

 

8. min(Comparator<T> comparator), max(Comparator<T> comparator)

: 주어진 비교자에 따라 최솟값 또는 최댓값을 반환한다.

        // 최종 연산: min and max
        List<Integer> numbers = Arrays.asList(3, 1, 4, 1, 5, 9, 2, 6, 5);
        Optional<Integer> min = numbers.stream()
                                       .min(Integer::compareTo);
        Optional<Integer> max = numbers.stream()
                                       .max(Integer::compareTo);
        System.out.println("Min Result: " + min.orElse(0));
        System.out.println("Max Result: " + max.orElse(0));
        // Output: Min Result: 1, Max Result: 9

 

 

 

 

9. reduce()

: 스트림의 요소를 하나의 값으로 결합한다.

       // 최종 연산: reduce
        Optional<Integer> sum = numbers.stream()
                                       .reduce(Integer::sum);
        System.out.println("Reduce Sum Result: " + sum.orElse(0));
        // Output: Reduce Sum Result: 36

 

 

 

 

10. toArray()

: 스트림의 요소를 배열로 반환한다.

        // 최종 연산: toArray
        String[] array = fruits.stream()
                              .toArray(String[]::new);
        System.out.println("Array Result: " + Arrays.toString(array));
        // Output: Array Result: [apple, orange, banana, kiwi]

 

 

 

---

 

스트림은 왜 사용하는 걸까?

 

스트림의 메서드를 보다보면 굳이 스트림을 쓰지 않고서도 구현 할 수 있을 것 같다는 생각이 들게 된다.

List<String> fruits = Arrays.asList("apple", "orange", "banana", "kiwi");

// 스트림을 사용하지 않은 경우
List<String> resultWithoutStream = new ArrayList<>();
for (String fruit : fruits) {
    if (fruit.startsWith("a")) {
        resultWithoutStream.add(fruit.toUpperCase());
    }
}
System.out.println(resultWithoutStream);

 

 

// 스트림을 사용한 경우
List<String> resultWithStream = fruits.stream()
                                     .filter(fruit -> fruit.startsWith("a"))
                                     .map(String::toUpperCase)
                                     .collect(Collectors.toList());
System.out.println(resultWithStream);

 

 

 

스트림을 제대로 사용하면 프로그램이 짧고 깔끔해지지만, 잘못 사용하면 읽기 어렵고 유지보수가 힘들다.

 

때문에, 모든 반복문을 스트림으로 바꾸기 보다, 반복문과 스트림을 적절히 조합해야 한다.

 

public class Anagrams {
	  public static void main(String[] args) throws IOException {
      File dictionary = new File(args[0]);
      int minGroupSize = Interger.parseInt(args[1]);
      
      Map<String, Set<String>> groups = new HashMap<>();
      try(Scanner s = new Scanner(dictionary)){
      	while(s.hasNext()){
        	String word = s.next();
            groups.computeIfAbsent(alphabetize(word), (unused) -> new TreeSet<>()).add(word);
        }
      }
      
      for(Set<String> group : groups.values())
      	if(group.size() >= minGroupSize)
        	System.out.println(group.size() + ":" + group);
    }
    
    public static String alphabetie(String s){
    	char[] a = s.toCharArray();
        Arrays.sort(a);
        return new String(a);
    }
}

 

해당 코드는 입력받은 단어 중 사용자가 지정한 값보다 원소 수가 많은 아나그램 그룹을 출력한다.

 

 

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

 

위 코드를 스트림을 활용하여, 프로그램 전체가 단 하나의 표현식으로 처리되게 만들었다.

 

오히려 기존보다 코드 가독성이 떨어졌다.

 

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

 

이와 같이 세부 구현은 주 프로그램 로직 밖으로 빼내 가독성을 높여주자.

 

이러한 도우미 메서드를 적절히 활용하는 것은 일반 반복 코드에서보다 스트림 파이프 라인을 사용할 때 매우 중요해진다.

 

또한, 람다에서는 타입 이름이 생략되므로, 매개변수의 이름을 타입을 유추할 수 있도록 적절히 지어주도록 하자.

 

 

---

 

이제 스트림의 병렬 프로그래밍에 대해 알아보자.

 

개발자를 시작하고 서버를 만들 기회가 생길 때 마다 문제가 생겼던 곳은 주로 동시성을 생각해야 하는 부분이였다.

 

 

public static void main(String[] args) {
   primes().map(p -> TWO.pos(p.intValueExact()).subtract(ONE))
      .fileter(mersenne -> mersenne.isProbablePrime(50))
      .limit(20)
      .forEach(System.out::println);
}

static Stream<BigInteger> primes() {
   return Stream.iterate(TWO, BigInteger::nextProbablePrime);
}

 

해당 프로그램을 실행하면 문제 없이 실행된다.

 

하지만 속도를 높이고 싶어 스트림 파이프라인의 parallel()를 호출하면 아무것도 출력하지 못하면서 CPU가 과부화되는 상태가 무한히 지속된다.

---

Stream의 parallel란?

 

Java8부터는 parallelStream(), parallel()만을 사용하고도 stream을 병렬 처리할 수 있게 한다. ForkJoinPool 관리 방식을 사용해서 복잡하던 스레드 관리 방식을 Fork와 Join을 통해서 작업들을 분할 정복(Divide and Conquer) 기법으로 처리한다. ParallelStream을 사용할 때 몇 가지 특징에 대해 알고 넘어가는 것이 좋다.

 

• 병렬 처리이기 때문에, 순서를 보장하지 않는다.
• 별도의 설정이 없으면 해당 어플리케이션이 구동되는 스펙에 따라 스레드 수가 결정된다.
• 스레드가 N개 생성되었을 때, 하나는 메인 스레드로 스트림을 처리하는 기본 스레드, 나머지 N-1개의 스레드가 ForkJoinPool 스레드이다.
• 가장 중요한 부분인데, 여기서 사용하는 ForkJoinPool을 Custom Pool을 별도로 생성해서 활용하지 않으면, JVM 인스턴스에서 공통적으로 사용하는 Pool을 사용한다.

---

 

스트림 라이브러리가 이 파이프라인을 병렬화하는 방법을 찾아내지 못했기 때문이다.

 

스트림 파이프라인을 아무 생각 없이 병렬스트림으로 돌리면 안된다.

 

대체로 스트림의 소스가 ArrayList, HashMap, HashSet, ConcurrentHashMap의 인스턴스거나

 

배열, int 범위, long 범위일 때 병렬화의 효과가 가장 좋다.

 

하지만 참조들이 가리키는 실제 객체가 메모리에서 서로 떨어져 있을 수 있는데, 그러면 참조 지역성이 나빠진다.

 

참조 지역성이 낮으면 스레드는 데이터가 주 메모리에서 캐시 메모리로 전송되어 오기를 기다리며 대부분 시간을 멍하니 보내게 된다.

 

따라서 참조 지역성은 병렬화 할 때 아주 중요한 요소로 작용하고, 참조 지역성이 가장 뛰어난 자료구조는 기본 타입의 배열이다.

 

스트림 파이프라인의 종단 연산의 동작 방식 역시 병렬 수행에 효율을 준다.

 

종단 연산에서 수행하는 작업량이 파이프라인 전체작업에서 상당 비중을 차지하면서 순차적인 연산이라면 파이프라인 병렬 수행의 효과는 제한될 수 밖에 없다.

 

종단 연산 중 병렬화에 가장 적합한 것은 축소다.

 

reduce, min, max, count, sum 가 속한다.

 

anyMatch, allMatch, noneMatch 처럼 조건에 맞으면 바로 반환되는 메서드도 적합하다.

 

반면 collect() 등 가변 축소를 수행하는 메서드들은 합치는 부담이 너무 크기 때문에 병렬화에 적합하지 않다.

 

직접 구현한 Stream, Iterable, Collection이 병렬화의 이점을 제대로 누릴려면 spliterator 메서드를 반드시 재정의하고 스트림 병렬화 성능을 강도 높게 테스트하도록 하자.

 

계산도 올바로 수행하고 성능도 빨라질 거라는 확신 없이는 스트림 파이프라인 병렬화는 시도조차 하지 말라.  잘못되면 오동작하게하거나 성능을 급격히 떨어뜨린다. 테스트 후에, 계산도 정확하고 성능도 좋아졌음이 확실해졌을 때, 오직 그럴 때만 병렬화 버전 코드를 운영 코드에 반영하라.