본 포스팅은 패스트캠퍼스 환급 챌린지 참여를 위해 작성하였습니다.
강의 요약
오늘 강의에서는 Project Reactor의 핵심 구성 요소인 Flux와 Mono의 실습 위주의 내용을 다루었다. StepVerifier를 통한 테스트를 작성 하며 진행했다. map, filter, take, flatMap, concatMap 등 주요 Operator의 실습을 진행했고, Schedulers의 종류에 대해서도 간단하게 알게 되었다. 오늘은 Flux와 Mono의 차이점, flatMap과 concatMap의 비동기 처리 특성, 그리고 Schedulers의 실행 모델에 대해 정리해보자.
Flux와 Mono
Reactor는 Reactive Streams 명세를 구현하며, 데이터의 개수(카디널리티)에 따라 두 가지 Publisher 타입을 제공한다.
Mono: 0-1개 결과를 위한 Publisher
- Mono는 최대 한 개의 값을 방출하는 특수한 Publisher다.
- onNext() 호출 후 즉시 onComplete() 시그널을 보내거나, 값 없이 onComplete()만 호출하거나, onError()를 통해 실패를 알린다.
- 데이터베이스의 단일 레코드 조회나 HTTP 요청처럼 단일 결과를 다루는 비동기 작업에 적합하며, Java의 Optional과 개념적으로 유사하다.
Mono<String> mono = Mono.just("Hello");
// onSubscribe -> onNext(Hello) -> onComplete
Flux: 0-N개 시퀀스를 위한 Publisher
- Flux는 0개 이상의 값을 방출할 수 있는 표준 Publisher다.
- 여러 onNext() 호출 후 onComplete() 또는 onError()로 종료된다.
- 컬렉션 처리, 스트림 데이터, 실시간 이벤트처럼 연속된 데이터를 다루는 경우 사용한다.
- Java 8 Stream과 유사해 보이지만, Stream은 pull 모델인 반면 Reactive는 push 모델이라는 근본적 차이가 있다.
Flux<Integer> flux = Flux.just(1, 2, 3, 4);
// onSubscribe -> request(unbounded) -> onNext(1) -> onNext(2) -> onNext(3) -> onNext(4) -> onComplete
두 타입의 선택은 반환값의 카디널리티로 결정된다. 단일 값이라면 Mono를, 여러 값이라면 Flux를 사용한다. 이러한 타입 구분은 API 설계의 의도를 명확히 표현하며, count()나 next() 같은 연산자를 통해 타입 간 전환도 가능하다.
flatMap의 비동기 변환 메커니즘
- flatMap은 각 원본 요소를 Publisher로 변환한 뒤, 이들을 평탄화하여 단일 스트림으로 병합하는 연산자다.
- map이 동기적인 1:1 변환을 수행하는 것과 달리, flatMap은 각 요소에 대해 비동기 작업을 수행하고 그 결과를 하나의 Flux로 통합한다.
동작 원리
Flux.just("user1", "user2", "user3")
.flatMap(userId -> getUserDetails(userId)) // 각 ID마다 비동기 호출
.subscribe(System.out::println);
- flatMap은 내부적으로 여러 Publisher를 동시에 구독할 수 있다.
- 위 예제에서 getUserDetails는 각 userId에 대해 개별 Mono를 반환하며, flatMap은 이 세 개의 Mono를 동시에 구독한다.
- 결과적으로 어떤 호출이 먼저 완료되느냐에 따라 출력 순서가 달라질 수 있다.
concurrency 파라미터
- flatMap은 동시성 수준을 제어할 수 있는 concurrency 파라미터를 제공한다.
- 기본값은 Queues.SMALL_BUFFER_SIZE(256)이며, 이 값을 조정하여 동시 처리 수를 제한할 수 있다.
- concurrency를 1로 설정하면 순차 처리되지만, 여전히 각 내부 Publisher는 비동기적으로 실행된다.
.flatMap(id -> callExternalApi(id), 3) // 최대 3개까지 동시 처리
flatMap vs concatMap: 순서 보장과 성능
두 연산자의 핵심 차이는 순서 보장 여부와 동시성 처리 방식에 있다.
| 특성 | flatMap | concatMap |
| 동시 구독 | 여러 Publisher 동시 구독 가능 | 하나씩 순차적으로 구독 |
| 순서 보장 | 완료 순서대로 출력 (순서 무보장) | 원본 순서 유지 |
| 처리량 | 높음 (병렬 처리) | 낮음 (순차 처리) |
| 적용 시나리오 | 순서 무관한 I/O 작업 | 순서 의존적 작업 |
비동기 처리에서의 차이점
// flatMap: 비동기 호출이 완료되는 즉시 방출
Flux.range(1, 3)
.flatMap(i -> Mono.delay(Duration.ofMillis(100 - i * 10))
.map(d -> "flatMap-" + i))
.subscribe(System.out::println);
// 출력: flatMap-3, flatMap-2, flatMap-1 (완료 순서)
// concatMap: 원본 순서를 유지하며 방출
Flux.range(1, 3)
.concatMap(i -> Mono.delay(Duration.ofMillis(100 - i * 10))
.map(d -> "concatMap-" + i))
.subscribe(System.out::println);
// 출력: concatMap-1, concatMap-2, concatMap-3 (원본 순서)
- flatMap은 각 Mono가 완료되는 즉시 결과를 방출하므로 지연 시간이 짧은 작업이 먼저 출력된다.
- 반면 concatMap은 첫 번째 작업이 완료될 때까지 대기한 후, 순서대로 결과를 방출한다.
적용 기준
flatMap 선택 시나리오
- 독립적인 외부 API 호출 (각 호출 결과가 서로 영향을 주지 않음)
- 대량의 데이터베이스 조회 (순서 무관)
- 처리량이 중요한 스트리밍 데이터 처리
concatMap 선택 시나리오
- 순차적 작업이 필요한 파일 처리 (이전 단계 결과가 다음 단계에 영향)
- 트랜잭션 순서가 중요한 금융 시스템
- 로그 순서 보장이 필요한 감사 시스템
Schedulers와 스레드 실행 모델
Reactor는 Schedulers를 통해 실행 컨텍스트를 추상화하며, 각 Scheduler는 서로 다른 스레드 풀 전략을 제공한다.
Schedulers.immediate()
- 현재 스레드에서 즉시 실행
- 스케줄링 오버헤드 없음
- 테스트나 동기 실행이 명확히 필요한 경우 사용
Schedulers.single()
- 단일 재사용 스레드
- 모든 작업이 동일한 스레드에서 순차 실행
- 경량 작업의 순서 보장이 필요할 때 적합
Schedulers.parallel()
- CPU 코어 수만큼 스레드 생성
- CPU 집약적 작업에 최적화
- 논블로킹 연산에 적합
Schedulers.boundedElastic()
- I/O 작업을 위한 탄력적 스레드 풀
- 최대 스레드 수 제한 (CPU 코어 * 10)
- 블로킹 호출을 격리할 때 사용
publishOn vs subscribeOn
두 연산자는 스레드 실행 위치를 제어하지만 영향 범위가 다르다.
publishOn
- 체인 내 특정 지점 이후의 연산자만 영향
Flux.range(1, 3)
.map(i -> {
System.out.println("map1: " + Thread.currentThread().getName());
return i * 2;
})
.publishOn(Schedulers.parallel())
.map(i -> {
System.out.println("map2: " + Thread.currentThread().getName());
return i + 1;
})
.subscribe();
// map1: main, map2: parallel-1
subscribeOn
- 전체 체인의 구독부터 특정 Scheduler로 이동
Flux.range(1, 3)
.map(i -> i * 2)
.subscribeOn(Schedulers.parallel())
.subscribe(System.out::println);
// [main] onSubscribe
// [parallel-1] request(unbounded)
// [parallel-1] onNext(2), onNext(4), onNext(6)
기본적으로 Reactor의 모든 연산은 subscribe()를 호출한 스레드에서 실행된다. subscribeOn을 사용하면 구독 시점부터 모든 상위 연산자가 지정된 Scheduler의 스레드에서 실행되며, 이는 로그를 통해 확인할 수 있다.
참고 출처
- https://www.baeldung.com/reactor-core
- https://www.baeldung.com/java-reactor-flux-vs-mono
- https://projectreactor.io/docs/core/3.5.15/reference/index.html#core-features
