리액티브 프로그래밍 — Reactor와 Reactive Streams 이해하기

[요청 1] → [스레드 1] → DB 쿼리 (200ms 대기) → 응답
[요청 2] → [스레드 2] → 외부 API (500ms 대기) → 응답
...
[요청 201] → 스레드 풀 고갈 → 대기열에서 기다림

public interface Publisher<T> {
    void subscribe(Subscriber<? super T> s);  // 구독 등록
}

public interface Subscriber<T> {
    void onSubscribe(Subscription s);  // 구독 시작
    void onNext(T t);                  // 데이터 수신
    void onError(Throwable t);         // 에러 발생
    void onComplete();                 // 완료 신호
}

public interface Subscription {
    void request(long n);   // n개의 데이터를 요청 (배압의 핵심)
    void cancel();          // 구독 취소
}

// Publisher이자 Subscriber — 중간 처리 단계
public interface Processor<T, R> extends Subscriber<T>, Publisher<R> { }

Publisher                    Subscriber
   │◄── subscribe() ────────────│
   │── onSubscribe(Subscription)──►│
   │◄── request(3) ─────────────│  ← 3개 요청
   │── onNext(data1) ──────────►│
   │── onNext(data2) ──────────►│
   │── onNext(data3) ──────────►│
   │◄── request(2) ─────────────│  ← 추가 2개 요청
   │── onNext(data4) ──────────►│
   │── onNext(data5) ──────────►│
   │── onComplete() ───────────►│

Flux.range(1, 1000)
    .onBackpressureBuffer(100)  // 최대 100개까지 버퍼링
    .subscribe(data -> slowProcess(data));

Mono<String> mono = Mono.just("안녕하세요");
// 여기서는 아무 일도 안 일어남

mono.subscribe(System.out::println);
// 이제 "안녕하세요" 출력

// 생성
Mono<String> mono = Mono.just("데이터");
Mono<String> empty = Mono.empty();
Mono<String> error = Mono.error(new RuntimeException("실패"));

// 지연 생성 — 구독 시점에 실행
Mono<String> deferred = Mono.defer(() ->
    Mono.just("현재 시간: " + System.currentTimeMillis())
);

// 블로킹 호출을 감싸기
Mono<String> fromCallable = Mono.fromCallable(() -> fetchFromDatabase());

Mono.just("hello")
    .map(s -> s.toUpperCase())       // 동기 변환: "HELLO"
    .flatMap(s -> saveToDb(s))       // 비동기 변환: Mono<Result> 반환
    .onErrorResume(e -> {            // 에러 시 대체 Mono
        log.warn("실패: {}", e.getMessage());
        return fetchFromCache();
    })
    .subscribe(result -> log.info("완료: {}", result));

public Mono<User> findOrCreate(String email) {
    return userRepository.findByEmail(email)
        .switchIfEmpty(Mono.defer(() -> {    // 비어있으면 실행
            User newUser = new User(email);
            return userRepository.save(newUser);
        }));
}

Flux<String> flux = Flux.just("A", "B", "C");
Flux<Integer> range = Flux.range(1, 10);
Flux<Long> interval = Flux.interval(Duration.ofSeconds(1));

// 필터와 변환
Flux.range(1, 100)
    .filter(n -> n % 2 == 0)           // 짝수만
    .map(n -> n * 10)                  // 10배
    .take(5)                           // 처음 5개만
    .subscribe(System.out::println);   // 20, 40, 60, 80, 100

// 여러 Flux 결합
Flux.zip(flux1, flux2, (s, n) -> s + n);  // 쌍으로 묶기: A1, B2, C3
Flux.concat(flux1, flux2);                 // 순서대로 이어 붙이기
Flux.merge(flux1, flux2);                  // 도착 순서대로 (순서 미보장)

// map: 동기 1:1 변환 (T → R)
Flux.just(1, 2, 3)
    .map(n -> n * 2)
    .subscribe(System.out::println);  // 2, 4, 6

// flatMap: 비동기 1:N 변환 (T → Publisher<R>)
Flux.just(1, 2, 3)
    .flatMap(n -> fetchData(n))       // 각 값마다 비동기 호출
    .subscribe(System.out::println);  // 결과 순서 보장 안 됨

// flatMap: 병렬 실행, 순서 미보장 (빠름)
Flux.just(3, 1, 2)
    .flatMap(n -> Mono.just(n).delayElement(Duration.ofMillis(n * 100)))
    .subscribe(System.out::println);  // 1, 2, 3

// concatMap: 순차 실행, 순서 보장 (느림)
Flux.just(3, 1, 2)
    .concatMap(n -> Mono.just(n).delayElement(Duration.ofMillis(n * 100)))
    .subscribe(System.out::println);  // 3, 1, 2

Mono.empty().defaultIfEmpty("기본값");             // 고정 값 반환
Mono.empty().switchIfEmpty(fetchDefaultFromDb());  // 대체 Publisher 실행

Mono<User> user = userService.findById(userId);
Mono<List<Order>> orders = orderService.findByUserId(userId);
Mono<Point> point = pointService.getBalance(userId);

Mono.zip(user, orders, point)
    .map(tuple -> new UserProfile(tuple.getT1(), tuple.getT2(), tuple.getT3()))
    .subscribe(profile -> log.info("프로필 조회 완료"));

Flux.range(1, 3)
    .map(i -> {
        log.info("[map1] {} on {}", i, Thread.currentThread().getName());
        return i * 10;
    })
    .publishOn(Schedulers.parallel())              // 여기부터 parallel 스레드
    .map(i -> {
        log.info("[map2] {} on {}", i, Thread.currentThread().getName());
        return i + 1;
    })
    .subscribeOn(Schedulers.boundedElastic())      // 소스 구독은 elastic 스레드
    .subscribe();

// [map1] 1 on boundedElastic-1    ← subscribeOn의 영향
// [map2] 10 on parallel-1         ← publishOn의 영향

// 블로킹 DB 호출을 리액티브로 감싸기
Mono<User> user = Mono.fromCallable(() ->
        jdbcTemplate.queryForObject("SELECT * FROM users WHERE id = ?", userRowMapper, userId)
    )
    .subscribeOn(Schedulers.boundedElastic());

// onErrorReturn — 기본값 반환
Mono.just("abc")
    .map(s -> Integer.parseInt(s))
    .onErrorReturn(0)
    .subscribe(System.out::println);     // 0

// onErrorResume — 대체 로직 실행 (특정 예외 타입 지정 가능)
Mono.just("abc")
    .map(s -> Integer.parseInt(s))
    .onErrorResume(NumberFormatException.class, e -> Mono.just(-1))
    .subscribe(System.out::println);     // -1

// retry — 재시도
webClient.get().uri("/api/data").retrieve()
    .bodyToMono(String.class)
    .retry(3)                            // 최대 3회 재시도
    .subscribe(System.out::println);

// retryWhen — 지수 백오프로 세밀한 재시도 제어
webClient.get().uri("/api/data").retrieve()
    .bodyToMono(String.class)
    .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
        .maxBackoff(Duration.ofSeconds(10)))
    .subscribe(System.out::println);

Spring MVC (동기 블로킹)
  요청 → 스레드 → DB 쿼리 ── 대기 ── 응답 (스레드가 대기하는 동안 아무것도 못 함)

Spring WebFlux (비동기 논블로킹)
  요청 → 이벤트 루프 → DB 쿼리 ── 콜백 등록 (스레드가 다른 요청 처리)

@GetMapping("/users/{id}")
public Mono<User> getUser(@PathVariable Long id) {
    return userService.findById(id);
}

@GetMapping("/users")
public Flux<User> getAllUsers() {
    return userService.findAll();
}

// 가상 스레드: 기존 동기 코드를 그대로 쓰면서 확장성 확보
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    for (int i = 0; i < 10_000; i++) {
        executor.submit(() -> {
            // 블로킹 코드도 OK — 가상 스레드가 알아서 양보
            return httpClient.send(request, BodyHandlers.ofString()).body();
        });
    }
}

// log()로 파이프라인 흐름 확인
Flux.range(1, 5)
    .log("range")           // 구독, request, onNext 등 로그 출력
    .map(n -> n * 2)
    .log("mapped")
    .subscribe();