Spring WebFlux — 리액티브 프로그래밍과 논블로킹 I-O

[요청 1] ──── DB 쿼리 ──── (대기 중...) ──── 응답 처리 ──── 완료
[요청 2] ──── API 호출 ──── (대기 중...) ──── 응답 처리 ──── 완료
[요청 3] ──── (스레드 풀 가득 참, 대기 중...)
...
[요청 201] ──── 503 Service Unavailable

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);
    void cancel();
}

public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
}

Subscriber → subscribe() → Publisher
Publisher  → onSubscribe(Subscription) → Subscriber
Subscriber → request(n) → Subscription
Publisher  → onNext(data) × n → Subscriber
Publisher  → onComplete() 또는 onError() → Subscriber

Flux.range(1, 1_000_000)
    .onBackpressureBuffer(1024)  // 버퍼 크기 제한
    .subscribe(
        data -> processSlowly(data),
        error -> log.error("버퍼 초과", error)
    );

// Mono — 하나의 결과
Mono<User> user = userRepository.findById(1L);

// Flux — 여러 개의 결과
Flux<User> users = userRepository.findAll();

// 생성
Mono<String> mono = Mono.just("hello");
Mono<String> empty = Mono.empty();
Mono<String> error = Mono.error(new RuntimeException("fail"));

Flux<Integer> flux = Flux.just(1, 2, 3);
Flux<Integer> range = Flux.range(1, 10);
Flux<Long> interval = Flux.interval(Duration.ofSeconds(1));

Mono<String> name = userMono.map(user -> user.getName());

Mono<Order> order = userMono
    .flatMap(user -> orderRepository.findLatestByUserId(user.getId()));

Mono<Tuple2<User, List<Order>>> result = Mono.zip(
    userRepository.findById(userId),
    orderRepository.findByUserId(userId).collectList()
);

Mono<User> user = userRepository.findById(userId)
    .onErrorResume(ex -> {
        log.warn("DB 조회 실패, 캐시에서 조회", ex);
        return cacheRepository.findById(userId);
    });

Spring Web (spring-web 모듈)
├── Spring MVC     ← 서블릿 기반, 블로킹
└── Spring WebFlux ← 리액티브 기반, 논블로킹

@RestController
@RequestMapping("/api/users")
public class UserController {

    private final UserService userService;

    public UserController(UserService userService) {
        this.userService = userService;
    }

    @GetMapping("/{id}")
    public Mono<ResponseEntity<UserResponse>> getUser(@PathVariable Long id) {
        return userService.findById(id)
            .map(user -> ResponseEntity.ok(UserResponse.from(user)))
            .defaultIfEmpty(ResponseEntity.notFound().build());
    }

    @GetMapping
    public Flux<UserResponse> getAllUsers() {
        return userService.findAll()
            .map(UserResponse::from);
    }

    @PostMapping
    public Mono<ResponseEntity<UserResponse>> createUser(
            @RequestBody Mono<UserRequest> request) {
        return request
            .flatMap(userService::create)
            .map(user -> ResponseEntity
                .created(URI.create("/api/users/" + user.getId()))
                .body(UserResponse.from(user)));
    }
}

@Configuration
public class UserRouter {

    @Bean
    public RouterFunction<ServerResponse> userRoutes(UserHandler handler) {
        return RouterFunctions.route()
            .GET("/api/users/{id}", handler::getUser)
            .GET("/api/users", handler::getAllUsers)
            .POST("/api/users", handler::createUser)
            .build();
    }
}

@Component
public class UserHandler {

    private final UserService userService;

    public UserHandler(UserService userService) {
        this.userService = userService;
    }

    public Mono<ServerResponse> getUser(ServerRequest request) {
        Long id = Long.valueOf(request.pathVariable("id"));
        return userService.findById(id)
            .flatMap(user -> ServerResponse.ok()
                .bodyValue(UserResponse.from(user)))
            .switchIfEmpty(ServerResponse.notFound().build());
    }

    public Mono<ServerResponse> getAllUsers(ServerRequest request) {
        return ServerResponse.ok()
            .body(userService.findAll().map(UserResponse::from), UserResponse.class);
    }

    public Mono<ServerResponse> createUser(ServerRequest request) {
        return request.bodyToMono(UserRequest.class)
            .flatMap(userService::create)
            .flatMap(user -> ServerResponse
                .created(URI.create("/api/users/" + user.getId()))
                .bodyValue(UserResponse.from(user)));
    }
}

@Configuration
public class WebClientConfig {

    @Bean
    public WebClient webClient() {
        return WebClient.builder()
            .baseUrl("https://api.example.com")
            .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
            .build();
    }
}

@Service
public class ExternalApiService {

    private final WebClient webClient;

    public ExternalApiService(WebClient webClient) {
        this.webClient = webClient;
    }

    // GET 요청
    public Mono<Product> getProduct(Long id) {
        return webClient.get()
            .uri("/products/{id}", id)
            .retrieve()
            .bodyToMono(Product.class);
    }

    // POST 요청
    public Mono<Product> createProduct(ProductRequest request) {
        return webClient.post()
            .uri("/products")
            .bodyValue(request)
            .retrieve()
            .onStatus(HttpStatusCode::is4xxClientError, response ->
                response.bodyToMono(String.class)
                    .flatMap(body -> Mono.error(
                        new ClientException("요청 오류: " + body))))
            .bodyToMono(Product.class);
    }

    // 여러 API 병렬 호출
    public Mono<OrderDetail> getOrderDetail(Long orderId) {
        Mono<Order> orderMono = webClient.get()
            .uri("/orders/{id}", orderId)
            .retrieve()
            .bodyToMono(Order.class);

        Mono<User> userMono = webClient.get()
            .uri("/users/{id}", orderId)
            .retrieve()
            .bodyToMono(User.class);

        return Mono.zip(orderMono, userMono)
            .map(tuple -> new OrderDetail(tuple.getT1(), tuple.getT2()));
    }
}

public interface UserRepository extends ReactiveCrudRepository<User, Long> {

    Flux<User> findByNameContaining(String name);

    @Query("SELECT * FROM users WHERE email = :email")
    Mono<User> findByEmail(String email);
}

[이벤트 루프 (소수의 스레드)]
  ├── 요청 1 수신 → 핸들러 실행 → I/O 대기 (스레드 반납) → 완료 콜백
  ├── 요청 2 수신 → 핸들러 실행 → I/O 대기 (스레드 반납) → 완료 콜백
  ├── 요청 3 수신 → ...
  └── ...수만 개 커넥션 처리 가능

Mono<String> result = Mono.fromCallable(() -> blockingOperation())
    .subscribeOn(Schedulers.boundedElastic());

userRepository.findById(userId)
    .checkpoint("사용자 조회 후")
    .flatMap(user -> orderService.findByUserId(user.getId()))
    .checkpoint("주문 조회 후")
    .subscribe();