[JAVA8] Stream

목차

• Stream 특징 
• Stream 파이프라인
• 중개 오퍼레이션 
• 종료 오퍼레이션 
• 참고 자료 

---

Stream 특징 

Stream은 데이터 저장소가 아니다

Stream은 데이터의 저장소가 아니라 데이터를 처리하는 파이프라인 

컬렉션(List, Set ...)은 데이터를 담고 있지만, Stream은 그 데이터를 연속된 요소(sequence of elements)로서 처리하는 작업을 지원 

List<Integer> numbers = Arrays.asList(1, 2, 3);
Stream<Integer> stream = numbers.stream(); // 데이터를 처리하는 스트림 생성

 

여기서 스트림은 데이터를 담고 있는 게 아니라, numbers 리스트에 있는 데이터를 연산할 때 사용하는 것 

 

Functional in nature

Stream API는 함수형 프로그래밍의 특징을 가진다.

함수형 프로그래밍에서는 데이터를 변경하지 않고, 불변성을 유지하면서 처리한다. 

Stream API 역시 데이터를 직접 수정하지 않고, 데이터 소스에 대한 연산을 수행해 결과를 만든다.

public static void main(String[] args) {
        List<String> names = Arrays.asList("John", "Jane", "Tom");
        // 스트림으로 데이터를 처리하지만, 원본 리스트 names는 변경되지 않는다.
        names
                .stream()
                .map(String::toUpperCase)
                .forEach(System.out::println);
        System.out.println("========================");
        names.forEach(System.out::println);
    }
}

// 출력 
JOHN
JANE
TOM
========================
John
Jane
Tom

 

스트림은 데이터를 한 번만 처리한다

Stream은 데이터를 한 번만 처리할 수 있다. 

한 스트림에 대해 연산을 수행하면, 그 스트림은 더이상 재사용할 수 없고, 새로운 스트림을 만들어야 한다.

Stream<String> stream = names.stream();
stream.forEach(System.out::println);
System.out.println("=====================");
try {
    stream.forEach(System.out::println);
}
catch (IllegalStateException e){
    System.out.println("e = " + e.getMessage());
}

// 출력 

John
Jane
Tom
=====================
e = stream has already been operated upon or closed

 

스트림은 무제한일 수 있다

Stream은 무제한의 데이터 흐름도 처리할 수 있다. 

예를 들어 무한 스트림을 생성해 계속해서 값을 생성하는 경우가 이에 해당한다. 

하지만 이런 경우에는 Short Circuit 메소드를 사용해 제한할 수 있다. 

Stream<Integer> infiniteStream = Stream.iterate(0, n -> n + 2); // 무한 스트림
infiniteStream.limit(10).forEach(System.out::println);

// 출력 

0
2
4
6
8
10
12
14
16
18

이처럼 limit(), findFirst(), anyMatch() 등의 Short Circuit 메소드로 무한 스트림도 제한할 수 있다.

 

중개 오퍼레이션은 Lazy하다

스트림에서 중개 오퍼레이션은 Lazy 동작한다. 

데이터를 바로 처리하지 않고, 최종 오퍼레이션이 호출될 때까지 실행을 미룬다. 

 

중개 오퍼레이션에는 map(), filter(), sorted() 등이 있다. 

이런 메소드들은 호출되더라도 실제로 연산이 수행되지 않는다. 

최종 오퍼레이션인 forEach(), collect() 등이 호출될 때 모든 연산이 한 번에 처리된다. 

System.out.println(names);
System.out.println();
Stream<String> stringStream = names.stream()
        .map(s -> {
            System.out.println(s);
            return s.toUpperCase();
        });

Stream.iterate(1, n -> n + 1).limit(3).forEach(System.out::println);
stringStream.forEach(System.out::println); // 최종 오퍼레이션이 호출될 때 모든 연산이 한 번에 처리

// 출력
[John, Jane, Tom]

1
2
3
John
JOHN
Jane
JANE
Tom
TOM

 

중개 오퍼레이션 Lazy 하다. 

Stream.iterate(1, n -> n + 1).limit(3).forEach(System.out::println);

 

중개 오퍼레이션이 먼저 동작하지만 위 코드가 먼저 출력 

 

출력이 번갈아 나오는 이유!! (중개 오퍼레이션 → 최종 연산 →  중개 오퍼레이션 →  최종연산 ...) 

한 요소를 처리할 때마다 map()과 forEach()가 번갈아 가면서 처리된다.

전체 스트림을 한 번에 처리하는 것이 아니라, 각 요소에 대해 즉시 처리가 일어나기 때문

 

손쉽게 병렬 처리할 수 있다

Stream API는 병렬 처리를 쉽게 지원한다. parrelStream()을 사용하거나 parallel() 메소드를 통해 순차 스트림을 병렬 

스트림으로 변환할 수 있다. 

이를 통해 여러 CPU 코어에서 데이터를 병렬로 처리하여 성능을 향상시킬 수 있다.

List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
numbers.stream().forEach(System.out::println);          // 병렬 처리 x (출력 순서 규칙적)
System.out.println("==============================");
numbers.parallelStream().forEach(System.out::println);  // 병렬 처리 o (출력 순서 불규칙)

// 출력 

1
2
3
4
5
6
7
8
9
10
==============================
7
6
9
10
8
3
5
4
2
1

 

하지만 언제나 병렬 처리가 베스트는 아니다. 

병렬 처리를 하면 스레드 컨텍스트 스위칭 비용과 CPU 스케줄링 비용이 오버헤드로 작용한다. 

데이터량이 적다면 오히려 이런 추가 오버헤드가 더 크게 반영되어 더 느려질 수 있다. 

항상 성능 테스트를 통해 적용하자. 

 

스레드 이름 출력 

List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
numbers
        .stream()
        .map(c -> {
            System.out.println(c + " " + Thread.currentThread().getName());
            return c;
        })
        .forEach(System.out::println);   // 병렬 처리 x (출력 순서 규칙적)
System.out.println("==============================");
numbers
        .parallelStream()
        .map(c -> {
            System.out.println(c + " " + Thread.currentThread().getName());
            return c;
        })
        .forEach(System.out::println);  // 병렬 처리 o (출력 순서 불규칙)

1 main
1
2 main
2
3 main
3
4 main
4
5 main
5
6 main
6
7 main
7
8 main
8
9 main
9
10 main
10
==============================
7 main
7
6 main
6
9 main
9
10 main
10
8 main
8
3 ForkJoinPool.commonPool-worker-1
3
2 main
2
1 main
1
4 main
4
5 ForkJoinPool.commonPool-worker-1
5

---

Stream 파이프라인

스트림 파이프라인(Stream Pipeline)은 Java Stream API의 기본 구성 방식으로, 데이터를 처리하는 흐름을 

중개 오퍼레이션과 종료 오퍼레이션의 조합으로 이룬다. 

이 과정에서 데이터 소스는 종료 오퍼레이션이 호출되기 전까지는 실제로 처리되지 않으며, Lazy evaluation(지연 처리)

방식을 따른다. 

 

파이프라인 구성

 

○ 데이터 소스

스트림을 생성하는 원천 (컬렉션, 배열, I/O 채널 등...)

 

○ 중개 오퍼레이션

데이터를 변환하거나 필터링하는 연산 단계.

이 오퍼레이션은 Lazy 처리

ex) map(), filter(), sorted(), distinct() 

 

○ 종료 오퍼레이션

데이터를 처리하고 결과를 출력하거나 수집하는 최종 연산 

스트림이 실제로 동작하는 시점 

ex) forEach(), collect(), count(), reduce()

 

스트림 파이프라인의 동작 원리

• 0개 또는 다수의 중개 오퍼레이션과 1개의 종료 오퍼레이션으로 구성
• 중개 오퍼레이션은 데이터를 필터링, 매핑, 정렬 등 다양한 방식으로 변환하는 연산을 수행, 데이터는 실제 처리되지 않고 파이프라인으로 전달만 된다.
• 종료 오퍼레이션이 호출되면 비로소 스트림이 실행되며, 이전에 설정한 중개 오퍼레이션들이 차례로 적용

---

중개 오퍼레이션

스트림에서 중개 오퍼레이션은 연산 결과로 또 다른 스트림은 반환하며

여러 개의 중개 오퍼레이션을 연결하여 파이프라인을 형성할 수 있다. 

중개 오퍼레이션은 Stateless와 Stateful로 구분할 수 있으며, 각 유형에 따라 성능과 메모리 사용 방식이 다르다.

 

Stateless 오퍼레이션 

Stateless 오퍼레이션은 이전 요소의 상태와 상관없이 각 요소를 독립적으로 처리

성능과 메모리 사용량이 효율적이며, 일반적으로 더 가볍고 빠르다. 

 

• filter(): 조건을 만족하는 요소를 필터링.
• map(): 각 요소를 매핑하여 변환.
• flatMap(): 각 요소를 스트림으로 변환하고, 이를 하나의 스트림으로 합침.
• peek(): 요소를 소비하지 않고 중간에 어떤 동작을 수행 (디버깅용).
• skip(): 처음 N개의 요소를 건너뜀.
• limit(): 처음 N개의 요소까지만 스트림을 제한.

List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
numbers.stream()
       .filter(n -> n % 2 == 0)        // 짝수만 필터링
       .map(n -> n * n)                // 제곱값으로 변환
       .forEach(System.out::println);  // 출력

 

 

이 예시에서 filter()와 map()은 각 요소를 독립적으로 처리하므로 Stateless 오퍼레이션

 

Stateful 오퍼레이션

Stateful 오퍼레이션은 이전 요소들의 상태나 전체 스트림의 상태에 의존하여 요소를 처리한다.

즉, 일부 연산은 모든 데이터를 확인한 후에야 다음 단계로 넘어갈 수 있다.

이 연산들은 성능에 영향을 줄 수 있으며, 메모리 사용량도 증가할 수 있다.

 

 

 

• distinct(): 중복을 제거. 모든 데이터를 확인하고 중복을 제거해야 하므로 상태 의존적.
• sorted(): 요소를 정렬. 정렬하려면 모든 요소를 미리 확인해야 하므로 상태 의존적.
• limit()와 skip()도 일부 상황에서 상태 의존적으로 동작할 수 있다.

 

List<String> names = Arrays.asList("John", "Jane", "Tom", "Jane");
names.stream()
     .distinct()       // 중복 제거
     .sorted()         // 알파벳순으로 정렬
     .forEach(System.out::println);

 

이 예시에서 distinct()와 sorted()는 Stateful 오퍼레이션이다.

중복을 제거하려면 이전의 요소들을 모두 저장하고 비교해야 하고, 정렬도 전체 스트림을 모두 확인한 후에야 실행할 수 있다.

---

종료 오퍼레이션

스트림에서 종료 오퍼레이션은 스트림 파이프라인을 마무리하고, 연산 결과를 반환하는 마지막 단계이다.

종료 오퍼레이션은 스트림을 소비하며, 데이터 소스를 처리하여 특정 값을 반환하거나, 사이드 이펙트를 발생시키는 작업을 수행한다.

종료 오퍼레이션을 수행하면 더 이상 스트림을 사용할 수 없으며, 새로운 스트림을 생성해야 다시 데이터를 처리할 수 있다.

종료 오퍼레이션은 스트림의 데이터를 실제로 처리하기 때문에, 이전까지의 중개 오퍼레이션들이 게으르게(Lazy) 설정된 연산들이 실행된다.

종료 오퍼레이션은 Stream을 리턴하지 않고, 보통 값을 반환하거나 사이드 이펙트(예: 출력)를 발생시킨다.

 

• collect()
  • 스트림의 요소를 수집하여 최종 결과를 반환.
  • 일반적으로 List, Map, Set 등으로 변환할 때 많이 사용
  • Collectors 클래스의 여러 유틸리티 메서드와 함께 사용

List<String> list = Stream.of("John", "Jane", "Tom")
                          .map(String::toUpperCase)
                          .collect(Collectors.toList());
System.out.println(list);  // [JOHN, JANE, TOM]

 

• forEach()
  • 스트림의 각 요소에 대해 지정된 작업을 수행한다.
  • 보통 데이터를 처리할 때 사용되며, 스트림을 소비하면서 각 요소에 대해 동작을 실행한다.

Stream.of("John", "Jane", "Tom")
      .forEach(System.out::println);

// 출력 
John
Jane
Tom

• count()
  • 스트림의 요소 개수를 반환

long count = Stream.of("John", "Jane", "Tom")
                   .filter(s -> s.startsWith("J"))
                   .count();
System.out.println(count);  // 2

• min() / max()
  • 스트림의 요소 중에서 가장 작은/큰 값을 반환.
  • 요소 간 비교는 Comparator를 사용

Optional<Integer> min = Stream.of(5, 2, 8, 3)
                              .min(Integer::compareTo);
System.out.println(min.get());  // 2

• allMatch() / anyMatch() / noneMatch()
  • 스트림의 모든 요소가 조건을 만족하는지, 하나라도 조건을 만족하는지, 또는 아무것도 조건을 만족하지 않는지 boolean 값을 반환

boolean allMatch = Stream.of(5, 10, 15)
                         .allMatch(n -> n > 0);
System.out.println(allMatch);  // true

• findFirst() / findAny()
  • 스트림에서 첫 번째 요소 또는 임의의 요소(병렬 스트림에서 주로 사용)를 반환

Optional<String> first = Stream.of("John", "Jane", "Tom")
                               .findFirst();
System.out.println(first.get());  // John

• reduce()
  • 스트림의 요소를 누적 연산을 통해 하나의 값으로 결합한다.
  • 주로 요소들을 합치거나 특정 연산을 수행하여 단일 결과를 도출할 때 사용한다.

int sum = Stream.of(1, 2, 3, 4)
                .reduce(0, Integer::sum);  // 0은 초기값
System.out.println(sum);  // 10

 

여러가지 Stream API 테스트 

public class OnlineClass {
    private Integer id;
    private String title;
    private boolean closed;
    public OnlineClass(Integer id, String title, boolean closed) {
        this.id = id;
        this.title = title;
        this.closed = closed;
    }
    public Integer getId() {
        return id;
    }
    public void setId(Integer id) {
        this.id = id;
    }
    public String getTitle() {
        return title;
    }
    public void setTitle(String title) {
        this.title = title;
    }
    public boolean isClosed() {
        return closed;
    }
    public void setClosed(boolean closed) {
        this.closed = closed;
    }
}

 

/**
 * Stream API Practice
 */
public class App {
    public static void main(String[] args) {
        List<OnlineClass> springClasses = new ArrayList<>();
        springClasses.add(new OnlineClass(1, "spring boot", true));
        springClasses.add(new OnlineClass(2, "spring data jpa", true));
        springClasses.add(new OnlineClass(3, "spring mvc", false));
        springClasses.add(new OnlineClass(4, "spring core", false));
        springClasses.add(new OnlineClass(5, "rest api development", false));

        System.out.println("spring 으로 시작하는 수업");
        springClasses
        .stream()
        .filter(c -> c.getTitle().startsWith("spring"))
                .forEach(oc -> System.out.println(oc.getTitle()));

        System.out.println();
        System.out.println("close 되지 않은 수업");
        springClasses
        .stream()
        .filter(Predicate.not(OnlineClass::isClosed)) // !OnlineClass::isClosed 불가능 ㅋㅋ
        .forEach(oc -> System.out.println(oc.getTitle()));

        System.out.println();
        System.out.println("수업 이름만 모아서 출력");
        springClasses
                .stream()
                .map(OnlineClass::getTitle)
                .forEach(System.out::println);

        List<OnlineClass> javaClasses = new ArrayList<>();
        javaClasses.add(new OnlineClass(6, "The Java, Test", true));
        javaClasses.add(new OnlineClass(7, "The Java, Code manipulation", true));
        javaClasses.add(new OnlineClass(8, "The Java, 8 to 11", false));

        List<List<OnlineClass>> keesunEvents = new ArrayList<>();
        keesunEvents.add(springClasses);
        keesunEvents.add(javaClasses);

        System.out.println();
        System.out.println("두 수업 목록에 들어있는 모든 수업 아이디 출력");
        keesunEvents
                .stream()
                .flatMap(Collection::stream)
                .forEach(oc -> System.out.println(oc.getId()));

        System.out.println();
        System.out.println("10부터 1씩 증가하는 무제한 스트림 중에서 앞에 10개 빼고 최대 10개 까지만");
        Stream.iterate(10, n -> n + 1) // 무제한 스트림
                .skip(10)
                .limit(10)
                .forEach(System.out::println);

        System.out.println();
        System.out.println("자바 수업 중에 Test가 들어있는 수업이 있는지 확인");

        System.out.println(javaClasses
                .stream()
                .anyMatch(jc -> jc.getTitle().contains("Test")));

        System.out.println();
        System.out.println("스프링 수업 중에 제목에 spring이 들어간 것만 제목만 모아서 List로 만들기");
        List<String> springTitle = springClasses
                .stream()
                .filter(sc -> sc.getTitle().contains("spring"))
                .map(sc -> sc.getTitle())
                .collect(Collectors.toList());

        springTitle.forEach(System.out::println);

        // flatMap
        List<String> strings1 = new ArrayList<>();
        List<String> strings2 = new ArrayList<>();
        Stream.iterate("o", s -> s + s).limit(10)
                .forEach(s -> {
                        strings1.add(s);
                        strings2.add(s);
                });

        List<List<String>> strings = new ArrayList<>();
        strings.add(strings1);
        strings.add(strings2);

        List<String> collect = strings.stream()
                .flatMap(Collection::stream)
                .collect(Collectors.toList());
        collect.forEach(System.out::println);
    }
}

 

---

참고 자료 

인프런 백기선님 더 자바, Java 8

더 자바, Java 8 강의 | 백기선 - 인프런

 

JAVA8 공식 문서 

• https://docs.oracle.com/javase/8/docs/api/java/util/stream/package-summary.html
• https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html