Stream 파이프라인

Stream은 데이터 소스에서 원소를 가져와 중간 연산을 연결하고 최종 연산에서 결과를 만듭니다.

중간 연산은 지연되어 최종 연산이 호출될 때 필요한 만큼만 실행됩니다.

파이프라인 내부에서 외부 가변 상태를 수정하면 순서와 병렬 실행에 취약해집니다.

map, filter, flatMap, collect가 각각 변환, 선택, 평탄화, 수집의 역할을 갖습니다.

이 절에서는 루프를 무조건 바꾸기보다 데이터 흐름이 더 선명해지는 경우를 찾습니다.

Stream은 자료를 보관하는 컬렉션이 아니라 한 번 소비하는 계산 계획입니다. 소스, 중간 연산, 최종 연산이 연결돼도 findFirst 같은 단축 최종 연산은 필요한 원소까지만 요청할 수 있습니다. 이 지연성과 한 번 소비 규칙이 일반 List와 가장 크게 다른 지점입니다.

소스는 파이프라인의 원소를 제공한다

컬렉션, 배열, 파일 줄, 생성 함수에서 스트림을 만들며 원본 자료구조를 자동으로 변경하지 않습니다.

중간 연산은 새 스트림을 지연 구성한다

filter와 map을 호출한 시점에는 보통 원소가 처리되지 않습니다.

연산 순서가 처리량과 가독성에 영향을 줍니다.

두 파이프라인은 같은 원본을 다시 스트림으로 열지만 결과의 의미가 다릅니다. 하나는 조건을 통과한 판매를 그룹 집계하고, 다른 하나는 모든 판매를 표시 문자열로 변환합니다.

import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

public class StreamStages {
    record Sale(String category, int amount) { }

    public static void main(String[] args) {
        List<Sale> sales = List.of(
                new Sale("book", 20), new Sale("tool", 5), new Sale("book", 30));

        Map<String, Integer> totals = sales.stream()
                .filter(sale -> sale.amount() >= 10)
                .collect(Collectors.groupingBy(
                        Sale::category,
                        Collectors.summingInt(Sale::amount)));
        System.out.println(totals);

        List<String> labels = sales.stream()
                .map(sale -> sale.category() + ":" + sale.amount())
                .sorted()
                .toList();
        System.out.println(labels);
    }
}

금액 5인 tool은 첫 파이프라인의 filter에서 제외되어 합계 맵에 없지만, 두 번째 파이프라인에는 필터가 없어 tool:5가 남습니다. Stream이 원본 sales를 수정한 것이 아니라 각 최종 연산이 독립 결과를 만든 것입니다.

{book=50}
[book:20, book:30, tool:5]

첫 파이프라인은 필터 뒤 그룹 합계를 만들고 두 번째는 각 원소를 독립적으로 변환해 정렬합니다.

최종 연산이 소비를 시작한다

toList, reduce, collect, forEach, findFirst가 결과를 만들고 스트림을 닫힌 상태로 만듭니다.

다음 코드는 한 번 소비 규칙, 중첩 목록의 평탄화, 외부 부작용 회피를 각각 보여 줍니다.

var stream = List.of(1, 2, 3).stream();
System.out.println(stream.mapToInt(Integer::intValue).sum());
// stream.count(); // 실행 시 IllegalStateException: 이미 소비됨

List<List<String>> nested = List.of(List.of("A", "B"), List.of("C"));
List<String> flat = nested.stream().flatMap(List::stream).toList();

// 외부 mutable List에 forEach로 add하기보다 toList로 결과를 반환한다.
// parallelStream은 측정과 공유 상태 검토 없이 켜지 않는다.

부작용 없는 변환을 선호한다

외부 리스트에 add하는 forEach보다 map(...).toList()처럼 결과를 반환하면 실행 순서와 병렬 여부에 덜 민감합니다.

스트림은 한 번 소비하면 다시 쓸 수 없다

최종 연산 뒤 같은 Stream을 재사용하면 IllegalStateException이 납니다.

다시 순회하려면 소스에서 새 스트림을 만듭니다.

병렬화는 순서와 비용을 측정한 뒤 선택한다

parallelStream은 자동 성능 향상 스위치가 아닙니다.

작업 크기, 분할 비용, 공유 상태, 순서 요구를 측정합니다.

filter 앞뒤에 peek를 잠시 넣고 findFirst로 최종 연산을 바꿔 보세요. 모든 원소가 처리되지 않는 모습을 관찰한 뒤 peek는 진단용으로만 제거합니다. 실제 상태 변경을 peek나 forEach에 숨기면 단축 평가와 병렬 실행에서 처리 횟수를 추론하기 어려워집니다.

parallelStream은 공용 ForkJoinPool, 분할 가능한 소스, 충분히 큰 CPU 작업이라는 조건의 영향을 받습니다. 파일 I/O나 작은 목록에 습관적으로 적용하지 말고, 순차 버전을 기준으로 실제 환경에서 측정한 뒤 선택합니다. 요청 서버에서는 공용 풀 사용이 다른 작업의 지연에도 영향을 줄 수 있습니다.

Stream은 컬렉션 문법의 대체품이 아니라 원소가 어떤 단계로 걸러지고 변환되어 결과가 되는지 표현하는 파이프라인입니다.