비동기 Web Flux

WebFlux란?

Spring WebFlux는 Spring 5에서 도입된 Reactive Programming을 지원하는 비동기, 논블로킹(non-blocking) 웹 프레임워크입니다. 기존의 Spring MVC가 Servlet 기반의 동기적, 블로킹 방식으로 작동하는 것과는 달리, WebFlux는 Reactive Streams API를 기반으로 하여 더 높은 확장성과 비동기 처리를 제공합니다.

WebFlux의 주요 특징

1. 논블로킹 I/O (Non-blocking I/O)
    WebFlux는 Netty, Undertow와 같은 비동기 I/O 처리 엔진과 통합되어 동작합니다. 요청과 응답 처리 과정에서 I/O 작업이 블로킹되지 않아 많은 요청을 동시에 처리할 수 있습니다. Spring WebFlux는 비동기 I/O를 기반으로 설계된 프레임워크입니다. 이는 요청(request)과 응답(response)을 처리할 때 스레드가 특정 작업(I/O 작업) 때문에 멈추지 않고 다른 작업을 계속 수행할 수 있도록 설계된 방식입니다. 
   • Netty: 고성능 비동기 네트워크 프레임워크. 주로 비동기 I/O를 처리하며, WebFlux의 기본 실행 엔진으로 자주 사용됨.
   • Undertow: 가볍고 유연한 Java 기반 웹 서버로, Netty와 마찬가지로 비동기 I/O를 지원.
   • division of code flow from the main (or current) thread of execution. This concept is the foundation of asynchronous programming, where tasks can execute independently of the main thread without blocking it.
2.  Reactive Streams 기반
   WebFlux는 Reactive Streams 명세를 따릅니다. 이를 통해 데이터를 스트리밍 방식으로 처리하며, Publisher와 Subscriber 패턴을 활용합니다. 
   • 대표적으로 Mono (0 또는 1개의 데이터), Flux (0개 이상의 데이터) 타입을 사용합니다.
3. 비동기 체인 내의 값을 획득할 수 있음.

While traditional asynchronous "fire-and-forget" operations are focused on delegation without worrying about the result, WebFlux builds on this model to provide a structured way to retrieve and process values asynchronously by utilizing Mono and Flux.

 

WebFlux의 주요 디자인 패턴

Callback 패턴

The callback pattern is a way of managing asynchronous operations by passing a function (the "callback") as an argument to another function. The callback is executed after the asynchronous task completes.  A callback is a way to designate what to do when a certain event occurs or when data becomes available. a callback is "notified" (called) when the operation is complete or when input is provided.

 

public class AsyncProcessor {

    public void processDataAsync(String input, Callback callback) {
        new Thread(() -> {
            try {
                // Simulate a time-consuming task
                Thread.sleep(2000);
                String processedData = input.toUpperCase(); // Example processing
                callback.onComplete(processedData); // Invoke the callback
            } catch (InterruptedException e) {
                callback.onComplete("Error: " + e.getMessage());
            }
        }).start(); // Run in a new thread to simulate async behavior
    }
}

 

 

EventLoop 패턴

An event loop is a design pattern used in asynchronous programming. It acts as a central controller, continuously checking for tasks, events, or I/O operations that are ready to execute and dispatching them for processing. At its core, the event loop continuously polls for new events or completed tasks and dispatches them to their appropriate handlers

 

. Event Loop Architecture

Netty uses an event loop to manage I/O operations.

• How the Event Loop Works:
  • A single thread runs an event loop that listens for I/O readiness events, such as:
    • A socket being ready to read data.
    • A network connection being ready to accept a write.
    • A timeout or scheduled task being ready to execute.

Why It’s Good for I/O:

• The event loop is optimized for I/O-bound tasks by continuously polling for events and efficiently dispatching them to handlers.
• Since I/O events are often quick and lightweight (e.g., reading a small chunk of data), this model minimizes overhead.

 

 

WebFlux의 구성 요소

1. Handler & Router
   WebFlux는 두 가지 스타일로 요청 처리를 정의합니다.
   • Annotation-based (Spring MVC와 유사): @RestController, @GetMapping, @PostMapping 등
   • Functional-style: 함수형 라우터 (RouterFunction과 HandlerFunction 사용)

@RestController
public class ExampleController {

    @GetMapping("/hello")
    public Mono<String> sayHello() {
        return Mono.just("Hello, WebFlux!");
    }
}

 

@Configuration
public class RouterConfig {

    @Bean
    public RouterFunction<ServerResponse> route(ExampleHandler handler) {
        return RouterFunctions
            .route(GET("/hello"), handler::sayHello);
    }
}

@Component
public class ExampleHandler {

    public Mono<ServerResponse> sayHello(ServerRequest request) {
        return ServerResponse.ok().body(Mono.just("Hello, WebFlux!"), String.class);
    }
}

 

2. WebClient
기존의 RestTemplate을 대체하는 비동기 HTTP 클라이언트입니다.

• 비동기로 HTTP 요청을 보내고 응답을 처리합니다.
• 논블로킹 방식으로 데이터를 스트리밍합니다.

WebClient webClient = WebClient.create("https://example.com");

webClient.get()
    .uri("/api/resource")
    .retrieve()
    .bodyToMono(String.class)
    .subscribe(System.out::println);

 

 

 

WebFlux 프레임워크 동작방식

 

 

• When a request is received, Netty, the default server in Spring WebFlux, runs its event loop on a limited number of I/O threads. I/O thread in Netty's event loop subscribes to the Mono or Flux returned by your controller or service layer.
• WebFlux is non-blocking, meaning when a request encounters an I/O task (e.g., database query, external API call): The current thread (e.g., Netty's I/O thread) does not block while waiting for the I/O operation to complete. Instead, it registers a callback with the I/O system to be invoked when the operation completes.
• While waiting for the I/O operation, the thread that was handling the request is released to handle other tasks or requests. 
• When the I/O operation completes (e.g., database query returns data, or HTTP response arrives): A callback is triggered. The result is fed back into the reactive pipeline.

 

WebFlux 프레임워크 구성요소

Netty threads are designed to handle non-blocking operations efficiently. If a task is blocking or CPU-intensive, it should not run on Netty threads to avoid degrading performance. In such cases, the logic is offloaded to a different thread pool.

 

Netty is built on top of Java NIO, which allows handling I/O operations (e.g., reading/writing to sockets) without blocking the thread.

• How It Works:
  • Instead of waiting (blocking) for data to be available (e.g., an incoming HTTP request or a database response), Netty registers the I/O operation with the operating system and continues processing other tasks. The thread is not idle while waiting; it can handle other I/O events.

 

 

 

 

MONO와 FLUX

1. Mono

개념

• Mono는 최대 1개의 데이터(0개 또는 1개)를 비동기적(데이터를 즉시 반환하지 않고, 준비되었을 때 제공)으로 처리하는 데이터 컨테이너입니다. 즉, 데이터를 담고 있다가 필요할 때 꺼내서 사용할 수 있습니다.
• Mono는 미래에 받을 수 있는 값을 미리 준비해놓는 상자입니다.
• 주로 단일 값 또는 오류 처리를 위한 작업에 사용됩니다.

생성

Mono는 데이터를 감싸서 비동기적으로 처리할 준비를 합니다.

• just(T data) : 단일 데이터를 포함하는 Mono 생성.
• empty() : 아무 데이터도 포함하지 않는 빈 Mono 생성.
• error(Throwable e) : 오류를 반환하는 Mono 생성.
• fromCallable(Callable<T>) : Callable로부터 데이터를 생성. It wraps the execution result of the Callable into a Mono, 

데이터 꺼내기

Mono는 내부 데이터를 처리하기 위해 **구독(subscribe)**이라는 과정을 거칩니다. subscribe()가 호출되기 전까지는 실제로 데이터 스트림이 흐르지 않습니다. subscribe() 호출 이후, 메인 스레드는 즉시 다음 코드로 진행합니다. Reactive Streams의 처리 (onNext, onComplete, onError 이벤트)는 별도의 스레드(주로 스케줄러에 의해 관리되는 스레드)에서 비동기적으로 실행됩니다. subscribe() 이후의 Reactive Streams 처리는 비동기적으로 이루어집니다.이로 인해 메인 스레드는 블로킹되지 않고 다른 작업을 계속할 수 있습니다.

• subscribe() : 는 데이터를 꺼내 실제로 처리합니다. Mono는 기본적으로 아무 작업도 하지 않다가 subscribe가 호출될 때 실행됩니다. 
• WebFlux 프로그래밍에서 subscribe()의 실행은 코드 흐름을 Mono 내부의 체인에서 사용하는 쓰레드와 메인 쓰레드로 분리합니다. 완전한 WebFlux 프로그래밍 방식은 컨트롤러에서 Mono를 반환하고, WebFlux 프레임워크가 이를 구독(subscribe())하여 Mono 내부 값을 비동기적으로 처리한 후 클라이언트에게 응답으로 반환하는 구조를 취합니다."

 

Mono<String> mono = Mono.just("Hello, Mono!"); // 단일 값 생성
mono.subscribe(data -> System.out.println("Received: " + data)); // 데이터를 꺼내 사용

Mono의 상태

Mono는 다음 상태 중 하나로 동작합니다:

• onNext : Mono가 데이터를 방출하면 실행됩니다.
• onSubscribe : 구독이 시작될 때(onSubscribe 이벤트 발생 시) 호출됩니다.
• onComplete :  Mono가 데이터를 방출하고, 오류 없이 작업이 정상적으로 완료되면 호출됩니다.
• onError :  Mono 실행 중 오류가 발생하면 호출됩니다. 오류가 발생하면 onNext와 onCompletem, onSubscribe는 호출되지 않음.

Mono<String> mono = Mono.just("Hello, WebFlux!")
    .doOnNext(data -> System.out.println("onNext: " + data)) // 데이터 방출 시 실행
    .doOnComplete(() -> System.out.println("onComplete"))   // 완료 시 실행
    .doOnError(error -> System.out.println("onError: " + error)); // 오류 발생 시 실행

mono.subscribe();

Mono의 다양한 메서드

• map(Function<T, R>) : 데이터를 변환.
• flatMap(Function<T, Mono<R>>): Mono<data>로 return 된 값이 Mono<Mono<data>구조로 중첩되는 것을 막기위하여 Mono 안의 데이터를 빼서 return 값을 Mono<data>로 유지하는 것. Mono를 return 하는 메소드는 주로 
• onErrorReturn : 오류 처리 

WebClient로 비동기 서비스 짜기

WebClient가 진정한 비동기 처리를 하려면 다음 조건이 만족되어야 합니다:

1. Mono 또는 Flux 타입으로 데이터를 반환.
2. 반환된 Mono/Flux를 구독(subscribe)하지 않고, WebFlux나 Reactor가 자체적으로 처리하게 위임.
3. 논블로킹 환경에서 실행(Spring WebFlux 사용).

public Mono<ResponseDepositPayDto> createDepositOrderAndInitiatePay(
    long postId, long participationId, String userId, PGMethod pgMethod) {
    log.info("readyDepositPay");

    return depositService.validateDepositCreateRequest(postId, participationId, userId)
        .then(depositService.createAndSaveDepositOrder(postId, participationId, userId))
        .flatMap(depositEntity ->
            kakaopayApi.sendPayReadySign(postId, depositEntity.getDepositId(), userId)
                .flatMap(payReadyApiDto ->
                    paymentService.createPaymentAndSave(
                            depositEntity.getDepositId(), userId, payReadyApiDto.getTid(), pgMethod)
                        .map(payment -> buildResponseDepositPayDto(depositEntity, payment, payReadyApiDto))
                )
        );
}

 

 

WebClient를 사용한 흐름: 메인 스레드와 I/O 스레드 관점

 

In WebClient, notifications about the completion of an API call rely on the underlying non-blocking I/O model implemented by the Netty framework, which uses event-driven programming and the Reactor Netty runtime.

 

When you use WebClient to make a request, the HTTP call is initiated through Netty, the default non-blocking HTTP client in Spring WebFlux. Netty handles I/O operations (sending the HTTP request and waiting for the response) using an event loop.

 

In the event loop model:

• A small number of threads are used to manage a large number of connections.
• Instead of blocking, the event loop registers a callback that will be triggered when the HTTP response is received.

Registering a Callback:

• When the request is sent, a callback is registered to handle the response once it arrives.
• This callback is part of the Reactor Netty runtime, which integrates with the Reactor framework used by WebClient.

 

Response Received (Callback Triggered):

• When the HTTP response is received, the event loop triggers the registered callback. This callback:
  • Reads the HTTP response headers and body in chunks (non-blocking).
  • Emits the data to the reactive stream.

 

 

 

 

2. Reactive Streams의 스레드 흐름

• 구독 시작 : subscribe() 메서드가 호출되면, 일반적으로 Netty의 이벤트 루프 스레드(예: reactor-http-nio-1)가 구독을 시작합니다.
• 연산자 체인 실행 : 구독이 시작되면, Mono나 Flux의 연산자 체인(map, flatMap 등)이 실행됩니다. 이 과정은 대개 다른 Netty 스레드(예: reactor-http-nio-2)에서 수행됩니다.
• 비동기 작업 처리 : 만약 체인 내에 WebClient를 통한 API 호출 같은 비동기 작업이 있다면, 이는 별도의 스레드(예: reactor-http-nio-3)에서 실행될 수 있습니다. 이는, mono로 이루어진 request 내에 webclient가 다시 mono를 만들기 때문입니다.
• 결과 전달: 모든 처리가 완료되면, 결과는 다시 Netty의 이벤트 루프 스레드로 전달됩니다.

 

2. Mono와 Flux

Mono

• Definition: 하나의 이벤트를 통해 발생한 데이터의 컨테이너
• Use Case: Use Mono when you expect a single result or no result (i.e., void) from an operation.

Flux

Definition: 복수의 이벤트(스트림)를 통해 발생한 데이터의 컨테이너

• Use Case: Use Flux when you expect multiple results or a stream of data, such as processing a list of items.

 

 

사용 예

Mono Examples:

• Fetching a single user by ID from a database.
• Returning a success or error response from an API call.

 

Flux Examples:

• Fetching a list of users from a database.
• Streaming real-time events (e.g., WebSocket, live updates).

 

주요 메서드

• just(T... data) : 여러 데이터를 포함하는 Flux 생성.
• fromIterable(Iterable<T>) : 리스트 또는 컬렉션에서 Flux 생성.
• range(int start, int count) : 범위의 숫자를 포함하는 Flux 생성.
• interval(Duration) : 일정 시간 간격으로 데이터를 방출.
• filter(Predicate<T>) : 데이터를 필터링.
• map(Function<T, R>) : 데이터를 변환.
• flatMap(Function<T, Flux<R>>): Flux로 매핑.