## Stream API vs Flux: 아키텍처 차이점과 실전 적용
비동기 시스템을 다루는 Java 개발자에게 Stream API와 Flux의 차이점을 이해하는 것은 매우 중요합니다. filter, map, limit 등의 메서드에서 유사점이 있음에도 불구하고, 이 두 도구는 근본적으로 다른 문제를 해결합니다. Stream API는 유한 컬렉션을 동기적으로 처리하기 위해 설계되었으며, Project Reactor의 Flux는 백프레셔와 비동기 지원을 갖춘 스트리밍 데이터의 리액티브 처리를 위해 만들어졌습니다. 이 글에서는 주요 아키텍처 차이점을 분석하고 적합한 도구를 선택하는 데 도움을 줍니다.
Stream API Basics: Synchronous Collection Processing
Stream API는 Java의 내장된 선언적 컬렉션 처리 메커니즘입니다. 핵심 특징: 유한 데이터 세트를 단일 실행 스레드 내에서 처리합니다. .stream()을 호출하면 시작된 스레드(예: 메인 스레드)에서 요소를 순차적으로 처리하는 순차 스트림이 생성됩니다.
작업 분류를 기억하세요:
- 중간 작업 (filter, map, flatMap, limit, skip) — 즉시 실행되지 않고 변환 체인을 구축합니다. 새로운 스트림을 반환하며 무한정 체이닝할 수 있습니다.
- 최종 작업 (collect, forEach, reduce) — 전체 체인을 실행하고 스트림을 종료합니다. 호출 후 스트림은 재사용할 수 없습니다.
데이터 필터링 및 변환 예시:
List<Integer> numbers = List.of(1, 2, 3, 4, 5);
numbers.stream()
.filter(n -> n % 2 == 0)
.map(n -> n * 2)
.forEach(System.out::println); // 4, 8여기서:
filter는 짝수를 유지합니다map은 각 요소를 두 배로 만듭니다forEach는 실행을 트리거하고 결과를 출력합니다
Stream API의 치명적 제한은 일회성 사용입니다. 다른 최종 작업을 시도하면 IllegalStateException이 발생합니다. 이는 리액티브 스트림과의 근본적 차이입니다.
Flux: Reactive Streaming Data Processing
Project Reactor의 Flux는 무한 데이터 스트림의 비동기 처리를 위한 Reactive Streams 사양을 구현합니다. Stream API와 달리 Flux는 다음을 지원합니다:
- 논블로킹 실행
- 백프레셔 (데이터 소비 속도 제어)
- 리액티브 연산자를 통한 오류 처리
- 모든 스레드에서의 이벤트 구독
주요 아키텍처 차이점:
- 최종 작업 — 처리를 시작하려면
.subscribe()가 필수입니다. 없으면 Flux는 "cold" 상태로 유지되며 데이터를 생성하지 않습니다. - 지연 초기화 — 구독 후에만 데이터가 생성됩니다.
- 무한 스트림 지원 — Flux는 무한 소스(예: 실시간 센서 데이터)를 처리합니다.
Flux에서의 동등 작업:
Flux.just(1, 2, 3, 4, 5)
.filter(n -> n % 2 == 0)
.map(n -> n * 2)
.subscribe(System.out::println); // 4, 8여기서 .subscribe()는 처리를 시작할 뿐만 아니라 스트림 생명주기를 관리합니다. Flux는 .onErrorResume()으로 오류를 처리하고 .limitRate()으로 흐름을 조절합니다.
Critical Differences: Comparison Table
명확성을 위해 주요 차이점을 구조화하여 정리했습니다:
- 실행 모델
- Stream API: 동기적, 블로킹 처리
- Flux: 비동기적, 논블로킹 처리와 백프레셔
- 데이터 유형
- Stream API: 유한 컬렉션만
- Flux: 유한 및 무한 스트림
- 오류 처리
- Stream API: 예외 직접 발생 (try/catch 필요)
- Flux: 연산자를 통한 내장 오류 처리 (onErrorResume, retry)
- 재사용성
- Stream API: 최종 작업 후 일회성
- Flux: 적절한 설정 시 하나의 스트림에 여러 구독 가능
- 병렬성
- Stream API: .parallelStream()으로 CPU 코어 분산
- Flux: .publishOn() 및 .subscribeOn()으로 실행 컨텍스트 제어
Practical Use Case: Data Transformation in Real Scenarios
HashMap에 저장된 직원과 그들의 등급을 업데이트하는 상황을 고려해보세요. 요구사항:
- Junior 직원의 등급을 Middle로 승격
- "(promoted)" 태그 추가
- 업데이트된 구조 반환
Stream API 구현:
Map<String, String> updatedEmployees = employees.entrySet().stream()
.map(entry -> {
String grade = entry.getValue();
if ("Junior".equals(grade)) {
return Map.entry(entry.getKey(), "Middle (increased)");
}
return entry;
})
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));Flux로 동일 시나리오:
Map<String, String> updatedEmployees = Flux.fromStream(employees.entrySet().stream())
.map(entry -> {
if ("Junior".equals(entry.getValue())) {
return Map.entry(entry.getKey(), "Middle (increased)");
}
return entry;
})
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))
.block();주요 구현 차이점:
- Flux는 결과를 얻기 위해 명시적
.block()필요 (리액티브 원칙 위반) - Stream API는 완료될 때까지 실행 스레드를 블록합니다
- 리액티브 시스템에서는
.subscribe()를 통한 논블로킹을 선호하세요
Key Takeaways
- Stream API — 단일 스레드에서 동기적 유한 컬렉션 처리에 이상적. ETL 작업이나 비즈니스 로직의 데이터 변환에 사용.
- Flux — 비동기 스트리밍 데이터 처리. 네트워크 요청, 센서, 이벤트 버스에 완벽.
- 백프레셔 — Stream API에 없는 Flux의 핵심 기능. 과부하 시 시스템 안정 유지.
- 오류 처리 — Flux는 리액티브, Stream API는 전통적 try/catch.
- 성능 — 짧은 동기 작업은 Stream API 우위; 긴 비동기 프로세스는 Flux.
— Editorial Team
아직 댓글이 없습니다.