TIL

3. 주요 연산자

• 연산자(operator)는 이전 Publisher를 감싸 새 Publisher를 만드는 “작업대”(1단계) — 시퀀스를 변환·필터·결합·복구한다
• 전체 목록은 공식 부록 “Which operator do I need?” 참고 — 여기선 자주 쓰는 것 위주

변환 — map / flatMap

• map: 동기 1:1 변환 (값 → 값). 예: 문자열 → 길이. 블로킹/비동기 작업을 넣으면 안 됨
• flatMap: 비동기 1:N — 각 원소를 Publisher로 만든 뒤 평탄화(flatten). 원소마다 비동기 호출(예: WebClient)을 할 때 사용
  • ⚠️ 내부 Publisher들을 동시에 구독 → 결과가 도착 순으로 인터리빙되어 원본 순서 미보장
• concatMap: flatMap과 같지만 순차 구독(하나 끝나야 다음) → 순서 보장, 동시성 없음
• flatMapSequential: 동시에 구독(eager)하되 방출만 원본 순서로 재정렬
• handle: 1:N + 각 원소마다 방출/건너뛰기를 프로그래밍적으로 결정

    Flux.just("a", "bb")
        .map(String::length);          // 동기 1:1 → 1, 2
    flux.flatMap(s -> callAsync(s));   // 각 원소 → Publisher, 동시 구독·인터리빙
    flux.concatMap(s -> callAsync(s)); // 각 원소 → Publisher, 순차·순서 보장
  

연산자	변환	동시성	순서
map	동기 1:1	—	보존
flatMap	비동기 1:N	동시 구독	미보장(인터리빙)
concatMap	비동기 1:N	순차(하나씩)	보존
flatMapSequential	비동기 1:N	동시 구독(eager)	보존(방출만 정렬)

필터

• filter(조건) / filterWhen(비동기 조건) / ofType(타입 제한)
• distinct(전체 중복 제거) / distinctUntilChanged(연속 중복만)
• take(앞 N개) / takeLast(뒤 N개) / skip(앞 N개 버림) / elementAt(특정 위치 1개)

결합 — concat / merge / zip

• concat: 앞 Publisher가 완료된 뒤 다음을 구독 → 순차 연결(순서 보존)
• merge: 여러 소스를 동시에 구독 → 도착 순으로 인터리빙
• zip: 각 소스에서 하나씩 짝지어 결합 — 모든 소스가 방출해야 한 쌍 생성 (여러 비동기 호출 결과를 합칠 때)
• combineLatest: 어느 쪽이든 새 값이 오면 각 소스의 최신값끼리 결합
• 인스턴스 메서드형: startWith/concatWith/mergeWith/zipWith, 그리고 Mono의 완료를 조율하는 then/when/and

연산자	구독 방식	방출
concat	앞 완료 후 다음	순차(순서 보존)
merge	동시 구독	도착 순 인터리빙
zip	각 소스 대기	소스별 1개씩 짝지어 결합

곁다리 동작 — doOnXxx

• 신호를 소비하지 않고 부수 효과(로깅·메트릭 등)만 수행하고 그대로 흘려보냄
• doOnNext/doOnComplete/doOnError/doOnSubscribe, log(내부 신호 로깅)
• doFinally: 완료·에러·취소 어떤 종료에도 실행 (SignalType을 받아 종류 구분)

에러 처리

• onError는 터미널 신호 — 복구 연산자로 잡지 않으면 시퀀스가 그대로 종료된다
• 명령형 try/catch ↔ 연산자 매핑

명령형	연산자
정적 기본값 반환	onErrorReturn
대체 경로 실행(동적 fallback)	onErrorResume
예외 래핑 후 재전파	onErrorMap
에러 로깅 후 전파	doOnError
finally 정리	doFinally
try-with-resources	using

.onErrorReturn(DEFAULT);                    // 에러 → 정적 기본값 방출 후 완료
.onErrorResume(e -> fallbackPublisher(e));  // 에러 → 대체 Publisher로 전환(동적)
.onErrorMap(e -> new BusinessException(e)); // 예외를 다른 타입으로 래핑해 재전파
.onErrorComplete();                         // 에러를 삼키고 정상 완료로 전환
.doOnError(e -> log.warn("failed", e));     // 부수효과만 — 에러는 계속 전파됨

재시도 — retry / retryWhen

• retry(n): 에러 시 소스를 n번 재구독(re-subscribe) → cold 소스는 처음부터 재시작
• retryWhen(Retry): 동반 신호(RetrySignal)로 조건부·백오프 재시도

    .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))); // 최대 3회, 지수 백오프
    // Retry.fixedDelay(n, dur) / Retry.max(n) 등도 제공
  

• ⚠️ 재구독이므로 부작용 있는 소스엔 주의. 재시도·백오프 전략 자체는 재시도(Retry)·서킷 브레이커 참고

Context — 리액티브의 ThreadLocal 대체

• 왜 필요한가: 리액티브 체인은 스레드를 넘나들어 ThreadLocal이 불안정 → 스레드가 아니라 구독(subscription)에 매달린 불변 key-value 저장소가 필요 (3.1.0+)
• 전파 방향: 아래(구독자) → 위(업스트림) — 그래서 contextWrite는 그 값을 읽는 연산자보다 체인상 아래(downstream) 에 놓아야 위 연산자가 읽을 수 있다
• 쓰기: contextWrite(...), Context.of(k, v) / 읽기: Mono.deferContextual·Flux.deferContextual, transformDeferredContextual (ContextView는 읽기 전용, 불변)

    Mono.deferContextual(ctx -> Mono.just("Hello " + ctx.get(key)))  // (위) Context 읽기
        .contextWrite(ctx -> ctx.put(key, "World"));                 // (아래) Context 쓰기 → "Hello World"
  

• 보안 주체·추적 ID처럼 요청 범위 값을 스레드 경계 너머로 나를 때 사용