[Java] Stream

Stream이란?

Java의 Stream API는 일련의 데이터 흐름을 표준화된 방법으로 쉽게 처리할 수 있도록 지원하는 클래스의 집합(패키지)이다. 오라클 공식 문서에서는 Stream 패키지를 요소들의 Stream에 함수형 연산을 지원하는 클래스라고 정의하고 있다. 

 

즉, Java의 Stream을 이용하면 일련의 데이터를 함수형 연산을 통해 표준화된 방법으로 쉽게 가공, 처리할 수 있다.

 

• `Stream<T>`는 데이터 요소의 연속된 시퀀스로서, 데이터 소스를 기반으로 필터링, 변환, 집계 등의 작업을 처리할 수 있다.
• 데이터 소스 변경 금지: Stream은 데이터 소스를 직접 수정하지 않는다.
• 목적: 컬렉션과 달리 데이터 요소를 직접 관리하지 않고, 선언전 데이터 처리를 수행하기 위함이다.

---

Stream의 구성 요소

Stream은 파이프라인으로 구성되며, 다음의 세 가지 단계로 이루어진다.

1. 소스(Source): 데이터의 출처(컬렉션, 배열, I/O 채널 등)
2. 중간 연산(Intermediate Operations): Stream을 변환하거나 필터링하는 연산(`filter`, `map`, `flatMap` 등). Lazy(지연 처리)로 작동
3. 최종 연산(Terminal Operations): Stream 파이프라인을 종료하고 결과를 반환하거나 실행(`collect`, `forEach`, `reduce` 등)

---

Stream의 주요 특징

• Lazy Evaluation (지연 평가)
  • 중간 연산은 최종 연산이 호출될 때까지 실행되지 않는다.
  • 성능 최적화를 위해 필요한 데이터만 계산
• 한 번만 소비
  • Stream은 한 번만 사용 가능. 재사용하려면 새 Stream을 생성해야 한다.
  • 동일한 Stream으로 여러 파이프라인을 만들 수 없음.
• 병렬 처리 지원
  • Stream은 순차적(`sequential`) 또는 병렬적으로(`parallel`) 실행 가능
  • `parallelStream()` 또는 `BaseStream.parallel()` 사용
• Stateless 및 Non-Interfering
  • 람다 함수나 메서드 참조를 사용할 때, 상태를 변경하거나 데이터 소스를 수정하지 않아야 함.

---

Stream 생성

컬렉션

List<String> list = List.of("a", "b");
Stream<String> stream = list.stream();

 

배열

String[] array = {"a", "b"};
Stream<String> stream = Arrays.stream(array);

 

`Stream.of()`

Stream<String> stream = Stream.of("a", "b", "c");

 

`Stream.builder()`

Stream<String> stream = Stream.<String>builder()
                              .add("One")
                              .add("Two")
                              .add("Three")
                              .build();

 

무한 스트림

• 요소가 실시간으로 생성되므로, 미리 정의된 데이터 구조(`list`, `map` 등)와는 다르다.
• 종료 조건을 명시하지 않으면 무한 루프가 될 수 있으므로, `limit()`와 같은 조건을 설정해야 한다.
• 일부 무한 스트림은 병렬 스트림에서 제대로 작동하지 않을 수 있다.

`generate`

• `Supplier`를 인수로 받아 스트림 요소를 생성한다.

Stream.generate(Math::random) 
      .limit(5)               
      .forEach(System.out::println);

 

`iterate`

• 초기값(시드 값)과 다음 값을 생성하는 함수(람다)를 인수로 받는다.
• 이전 값이 다음 값의 입력으로 사용되며, 상태를 유지한다.

Stream<Integer> infiniteStream = Stream.iterate(0, n -> n + 1);

---

주요 메서드

중간 연산 (Intermediate Operations)

• `filter(Predicate<T>)`: 조건에 맞는 요소만 걸러낸다.

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

numbers.stream()
       .filter(n -> n % 2 == 0)  // 짝수 필터링
       .forEach(System.out::println); // 출력: 2, 4

Integer[] empIds = { 1, 2, 3, 4 };
    
List<Employee> employees = Stream.of(empIds)
    .map(employeeRepository::findById)
    .filter(e -> e != null)
    .filter(e -> e.getSalary() > 200000)
    .collect(Collectors.toList());

 

• `map(Function<T, R>)`: 각 요소를 다른 값으로 변환

List<String> names = List.of("John", "Jane");
names.stream()
     .map(String::toUpperCase)  // 대문자로 변환
     .forEach(System.out::println); // 출력: JOHN, JANE

Integer[] empIds = { 1, 2, 3 };
   
List<Employee> employees = Stream.of(empIds)
  .map(employeeRepository::findById)
  .collect(Collectors.toList());

 

• `flatMap(Function<T, Stream<R>>)`: 여러 스트림을 하나로 병합

List<List<String>> nested = List.of(List.of("a", "b"), List.of("c"));
nested.stream()
      .flatMap(List::stream)  // 중첩된 리스트를 병합
      .forEach(System.out::println); // 출력: a, b, c

 

• `peek(Consumer<? super T> action)`: 각 요소에 대해 특정 동작 수행

empList.stream()
  .peek(e -> e.salaryIncrement(10.0))
  .peek(System.out::println)
  .collect(Collectors.toList());

 

• `sorted()`: 정렬(기본 또는 사용자 지정)

List<Employee> employees = empList.stream()
  .sorted((e1, e2) -> e1.getName().compareTo(e2.getName()))
  // .sorted(Comparator.comparing(Employee::getName))
  .collect(Collectors.toList());

 

• `distinct()`: 중복 제거

List<Integer> intList = Arrays.asList(2, 5, 3, 2, 4, 3);
List<Integer> distinctIntList = intList.stream().distinct().collect(Collectors.toList());

 

• `limit(long n)`: 최대 n개의 요소만 유지
• `skip(long n)`: 처음 n개의 요소를 건너뜀

---

최종 연산 (Terminal Operations)

최종 연산이 완료된 후, 스트림 파이프라인은 소비된 것으로 간주되어 더 이상 사용할 수 없다.

 

• `collect(Collector)`: Stream 결과를 컬렉션 등으로 변환

List<String> upperCaseNames = names.stream()
                                   .map(String::toUpperCase)
                                   .collect(Collectors.toList());

@Test
public void whenCollectByJoining_thenGetJoinedString() {
    String empNames = empList.stream()
      .map(Employee::getName)
      .collect(Collectors.joining(", "));

    assertEquals(empNames, "Jeff Bezos, Bill Gates, Mark Zuckerberg");
}

 

• `forEach(Consumer)`: 각 요소에 대해 작업 수행

empList.stream().forEach(e -> e.salaryIncrement(10.0));

 

• `reduce()`: Stream의 요소를 하나의 값으로 축소 

int sum = List.of(1, 2, 3).stream()
                          .reduce(0, Integer::sum); // 결과: 6

• `count()`: 요소의 개수 반환
• `min(), max()`: 요소의 최소값, 최대값 반환
• `sum(), average()`: 합계, 평균 반환
• `findFirst()`: 첫 번째 요소를 반환
• `findAny()`: (병렬 Stream)에서 아무 요소나 반환
• `allMatch(Predicate)`: 모든 요소가 조건에 만족하는지 확인
• `anyMatch(Predicate)`: 하나라도 조건을 만족하는 지 확인
• `noneMatch(Predicate)`:  모든 요소가 조건을 만족하지 않는지 확인
• `takeWhile(Predicate)`: 조건이 참인 동안 요소를 가져옴. 조건이 처음으로 거짓이 되면 처리 중단
• `dropWhile(Predicate)`: 조건이 참인 동안 요소를 건너 뛰고, 거짓이 되면 이후의 요소를 처리
• `ofNullable`: Null 값이 있을 경우 빈 스트림을 반환

---

Advanced collect

joining

@Test
public void whenCollectByJoining_thenGetJoinedString() {
    String empNames = empList.stream()
      .map(Employee::getName)
      .collect(Collectors.joining(", "));

    assertEquals(empNames, "Jeff Bezos, Bill Gates, Mark Zuckerberg");
}

 

 

toSet

@Test
public void whenCollectBySet_thenGetSet() {
    Set<String> empNames = empList.stream()
      .map(Employee::getName)
      .collect(Collectors.toSet());

    assertEquals(empNames.size(), 3);
}

 

 

`summarizingDouble`

@Test
public void whenApplySummarizing_thenGetBasicStats() {
    DoubleSummaryStatistics stats = empList.stream()
      .collect(Collectors.summarizingDouble(Employee::getSalary));

    assertEquals(stats.getCount(), 3);
    assertEquals(stats.getSum(), 600000.0, 0);
    assertEquals(stats.getAverage(), 200000.0, 0);
}

 

 

`summaryStatistics()`

@Test
public void whenApplySummaryStatistics_thenGetBasicStats() {
    DoubleSummaryStatistics stats = empList.stream()
      .mapToDouble(Employee::getSalary)
      .summaryStatistics();

    assertEquals(stats.getSum(), 600000.0, 0);
}

 

 

`partitioningBy`

조건에 따라 스트림을 두 그룹으로 나누다.

@Test
public void whenStreamPartition_thenGetMap() {
    Map<Boolean, List<Integer>> isEven = Arrays.asList(2, 4, 5, 6, 8).stream()
      .collect(Collectors.partitioningBy(i -> i % 2 == 0));

    assertEquals(isEven.get(true).size(), 4);
    assertEquals(isEven.get(false).size(), 1);
}

 

 

`groupingBy`

스트림 요소를 여러 그룹으로 분류한다.

@Test
public void whenStreamGroupingBy_thenGetMap() {
    Map<Character, List<Employee>> groupByAlphabet = empList.stream()
      .collect(Collectors.groupingBy(e -> e.getName().charAt(0)));

    assertEquals(groupByAlphabet.get('B').get(0).getName(), "Bill Gates");
}

 

 

`reducing`

`reducing`은 `reduce`와 `reduce`와 유사하지만, Collector로 사용할 수 있다.

@Test
public void whenStreamReducing_thenGetValue() {
    Double percentage = 10.0;
    Double salIncrOverhead = empList.stream()
      .collect(Collectors.reducing(
          0.0, e -> e.getSalary() * percentage / 100, Double::sum));

    assertEquals(salIncrOverhead, 60000.0, 0);
}

---

병렬 스트림

병렬로 데이터를 처리한다.

List<Integer> numbers = List.of(1, 2, 3, 4);
numbers.parallelStream()
       .map(n -> n * 2)
       .forEach(System.out::println);

---

File Operation

File Write

@Test
public void whenStreamToFile_thenGetFile() throws IOException {
    String[] words = {
      "hello", 
      "refer",
      "world",
      "level"
    };
    
    try (PrintWriter pw = new PrintWriter(
      Files.newBufferedWriter(Paths.get("output.txt")))) {
        Stream.of(words).forEach(pw::println); // 각 단어를 파일에 씀
    }
}

 

 

File Read

private List<String> getPalindrome(Stream<String> stream, int length) {
    return stream.filter(s -> s.length() == length)  // 길이가 length인 문자열 필터
      .filter(s -> s.compareToIgnoreCase(
        new StringBuilder(s).reverse().toString()) == 0) // 회문 필터
      .collect(Collectors.toList());
}

@Test
public void whenFileToStream_thenGetStream() throws IOException {
    List<String> str = getPalindrome(Files.lines(Paths.get("output.txt")), 5);
    assertThat(str, contains("refer", "level"));
}

 

@Test
public void whenStreamTransformed_thenWriteToFile() throws IOException {
    List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);

    try (PrintWriter pw = new PrintWriter(Files.newBufferedWriter(Paths.get("numbers.txt")))) {
        numbers.stream()
            .map(n -> "Number: " + n) // 데이터 변환
            .forEach(pw::println);    // 변환된 데이터 기록
    }
}

---

주의사항

• Stream은 데이터 소스 변경 없이 읽기 전용으로 처리해야 한다.
• IO 리소스가 소스인 경우, 반드시 `try-with-resources`로 닫아야 한다.

---

 

 

 

참조

https://docs.oracle.com/javase/8/docs/api/?java/util/stream/Stream.html

https://stackify.com/streams-guide-java-8/

https://mangkyu.tistory.com/114