TIL

2. Mono와 Flux

Reactor가 리액티브 스트림의 Publisher를 구현한 두 핵심 타입 — 1단계의 Publisher/Subscriber·배압·”구독 전엔 실행 안 됨” 위에서 동작
둘 다 연산자 체인을 구성하는 출발점이며, WebFlux의 요청/응답 타입도 이 둘

Flux — 0..N 시퀀스

Flux<T>는 표준 Publisher<T> — 0~N개 아이템을 비동기로 방출하고, 선택적으로 완료(onComplete) 또는 에러(onError) 로 종료
이 세 신호가 그대로 다운스트림 Subscriber의 onNext/onComplete/onError 호출로 이어짐
예: 스트리밍 응답, 목록 조회 결과, 이벤트 스트림

Mono — 0..1 결과

Mono<T>는 특화된 Publisher<T> — onNext로 최대 1개를 방출한 뒤 곧바로 onComplete(값이 있든 없든 성공), 또는 onError 1회(실패)
- 대부분의 구현은 onNext 직후 즉시 onComplete를 호출
- onNext + onError 조합은 금지
Flux 연산자의 부분집합만 제공 — 0..1에 의미 있는 것만 (예: count처럼 다개수 전제 연산은 없음)
- 일부 연산자는 반환 타입이 바뀜: Mono.concatWith(Publisher) → Flux, Mono.then(Mono) → Mono
값 없는 완료를 표현할 때 Mono<Void> — Runnable처럼 “끝났다”만 알림
HTTP 요청은 응답이 하나뿐 → Flux<HttpResponse>보다 Mono<HttpResponse>가 적합

	Flux	Mono
아이템 수	0..N	0..1
정상 종료	onNext* 후 onComplete	onNext 0~1회 후 onComplete
연산자	전체	0..1에 맞는 부분집합
대표 용도	목록·스트림·이벤트	단건 조회·저장 결과·완료 신호(`Mono<Void>`)

마블 다이어그램

리액티브 시퀀스와 연산자의 동작을 시각화하는 표기 — 공식 문서·연산자 설명 전반에서 사용
읽는 법
- 시간은 왼쪽 → 오른쪽으로 흐름
- 각 도형 = onNext 값
- 세로 막대 | = onComplete(정상 종료), X = onError(에러 종료)
- 연산자 다이어그램은 위쪽 입력 시퀀스 → (변환 박스) → 아래쪽 출력 시퀀스 형태로, 연산자가 흐름을 어떻게 바꾸는지 보여줌

생성 — 팩토리 메서드

이미 값이 있을 때는 정적 팩토리로 간단히 만든다

  Flux.just("foo", "bar", "foobar");     // 알려진 값들을 나열
  Flux.fromIterable(list);               // Iterable(List 등)에서
  Flux.fromArray(array);                 // 배열에서
  Flux.range(5, 3);                      // 5, 6, 7 — (시작값, 개수)
  Flux.empty();                          // 값 없이 즉시 완료
  Flux.error(new RuntimeException());    // 즉시 에러로 종료

  Mono.just("foo");
  Mono.justOrEmpty(nullable);            // null이면 빈 Mono
  Mono.empty();                          // 값 없이 완료 (Mono<Void> 등)
  Mono.error(new IllegalStateException());

값을 프로그래밍적으로 만들어야 하면 generate/create/push (3단계 이후)

1단계 원칙대로 subscribe() 호출 전엔 아무 일도 일어나지 않는다 → subscribe가 시퀀스를 트리거

콜백 조합에 따라 5가지 오버로드 (뒤로 갈수록 더 많은 신호를 처리)

  subscribe();                                          // 구독만 하고 시퀀스 트리거
  subscribe(onNext);                                    // 각 값 처리
  subscribe(onNext, onError);                           // + 에러에 반응
  subscribe(onNext, onError, onComplete);               // + 정상 완료 시 실행 (onComplete은 Runnable — 값 없음)
  subscribe(onNext, onError, onComplete, subscription); // + Subscription을 받아 수요(request) 직접 제어

예: 값·에러·완료를 각각 처리

  Flux.range(1, 4)
      .map(i -> {
          if (i <= 3) return i;
          throw new RuntimeException("Got to 4");
      })
      .subscribe(
          i   -> System.out.println(i),                 // onNext: 1, 2, 3
          err -> System.err.println("Error: " + err),   // onError: i=4에서 예외 → 시퀀스 종료
          ()  -> System.out.println("Done"));           // onComplete: 에러로 끝나면 호출되지 않음

세밀한 배압 제어가 필요하면 마지막 변형의 람다 대신 BaseSubscriber를 확장해 hookOnSubscribe에서 request(n)을 직접 호출하는 편이 안전 (1단계 request(n)과 연결)

구독 취소 — Disposable

람다형 subscribe는 모두 Disposable을 반환 → dispose()를 호출하면 구독이 취소됨

취소는 소스에 “원소 생산을 멈추라“는 신호 = 리액티브 스트림의 Subscription.cancel()과 연결

  Disposable disposable = source.subscribe(...);
  disposable.dispose();   // 정리 시점에 구독 취소

Cold과 Hot

1단계에서 미룬 구분 — 시퀀스가 언제 데이터를 만드는가의 차이
- cold: 구독하는 순간 데이터를 새로 생성 → 구독자마다 처음부터 다시 흐름 (just/fromIterable/range 등 기본 팩토리는 cold)
- hot: 구독과 무관하게 진행 → 늦게 구독한 쪽은 구독 이후 신호만 수신
cold 시퀀스를 여러 구독자가 공유(hot화)하려면 share()/publish() 등 별도 연산자 필요 (3단계 이후)

TIL

2. Mono와 Flux

Flux — 0..N 시퀀스

Mono — 0..1 결과

마블 다이어그램

생성 — 팩토리 메서드

구독 — subscribe 람다 변형

구독 취소 — Disposable

Cold과 Hot