[Spring WebFlux] 2. Reactive Core

• spring-web 기반 구조
• ✅ reactive web application을 위한 저수준 기반 기능을 제공함
• ✅ 서버: HTTP Server → HttpHandler → WebHandler API → Controller / Functional Endpoint
• ✅ 클라이언트: HTTP Client → ClientHttpConnector → WebClient

 

1. HttpHandler

• 서로 다른 HTTP Server를 WebFlux의 공통 Reactive 처리 구조에 연결하는 최소 추상화
• ✅ HTTP Request를 받아 Response를 Non-Blocking으로 처리하고, 처리 완료 여부를 Mono<Void>로 반환
• ✅ Netty, Tomcat, Jetty 처럼 서로 다른 서버 API를 Webflux가 동일한 방식으로 사용할 수 있도록 추상화

 

코드) handle()

Mono<Void> handle(
    ServerHttpRequest request,
    ServerHttpResponse response
);

• request → 들어온 HTTP 요청
• response → 여기에 HTTP 응답을 작성
• Mono<Void> → 반환할 데이터는 없고, 응답 작업의 완료/실패만 비동기로 표현

 

서버별 Adapter

• 서버마다 내부 API가 다르므로 Webflux가 Adapter를 사이에 둠
• ✅ Server → XXXHttpHandlerAdapter → HttpHandler

 

표) 서버별 Adapter

Server	Adapter
Reactor Netty	ReactorHttpHandlerAdapter
Tomcat	TomcatHttpHandlerAdapter
Jetty	JettyCoreHttpHandlerAdapter
Servlet Container	ServletHttpHandlerAdapter

 

예시) Reactor Netty

HttpHandler handler = ...;

ReactorHttpHandlerAdapter adapter =
    new ReactorHttpHandlerAdapter(handler);

HttpServer.create()
    .host(host)
    .port(port)
    .handle(adapter)
    .bindNow();

 

2. WebHandler API

• Spring WebFlux에서 웹 애플리케이션 수준의 요청 처리 기능을 제공하는 API
• ✅ HttpHandler 위에서 실제 웹 애플리케이션에 필요한 기능을 제공함 
• ex) 세션, 필터, 예외 처리 등

 

Special bean types

• WebHttpHandlerBuilder는 다음 Bean들을 자동으로 찾아 WebHandler 처리 체인을 구성함

Bean	개수	역할
WebExceptionHandler	0..N	Filter 또는 WebHandler에서 발생한 예외 처리
WebFilter	0..N	요청 전·후 공통 처리
WebHandler	1	실제 요청 처리
WebSessionManager	0..1	WebSession 관리
ServerCodecConfigurer	0..1	Form / Multipart 등의 요청 데이터 파싱
LocaleContextResolver	0..1	요청의 Locale 결정
ForwardedHeaderTransformer	0..1	Forwarded Header 처리

 

Form Data

Multipart Data

Forwarded Headers

• 프록시나 로드밸런서를 거치기 전, 클라이언트가 원래 요청한 정보를 전달하기 위한 HTTP 헤더
• ✅ RFC 7239에서 표준 Forwarded 헤더를 정의함
• ✅ Client 요청 → Proxy가 원래 요청 정보 기록 → Forwarded Header 추가 → Application 전달
• ➡️ 요청이 프록시를 거치면서 바뀐 정보 때문에, 백엔드가 원래 클라이언트가 어떤 주소로 요청했는지 알 수 있도록 전달하는 정보

 

예시) Forwarded Headers

X-Forwarded-Proto: https
X-Forwarded-Host: example.com
X-Forwarded-Port: 443

Forwarded: proto=https;host=example.com

• 위 두 형태로 제공됨

 

Non-standard Headers

ForwardedHeaderTransformer

• Forwarded Header 애플리케이션이 보는 요청의 host, port, scheme을 원래 클라이언트 요청 기준으로 수정하는 컴포넌트
• ⚠️ Forwarded 헤더를 신뢰할 수 있는 프록시가 넣었는지 구별할 수 없음
• ➡️ Proxy에서 외부사용자가 임의로 넣은 값을 제거하도록 구성해야 함

 

예시) 처리 과정

원래 요청이 아래와 같다면

http://10.0.0.5:8080
Forwarded: proto=https;host=example.com

• 요청 변환 담당: "http://10.0.0.5:8080 → ForwardedHeaderTransformer → https://example.com"
• 변환에 성공한 Forwarded 헤더는 제거함

 

코드) 빈 등록

@Bean
ForwardedHeaderTransformer forwardedHeaderTransformer() {
    ForwardedHeaderTransformer transformer = new ForwardedHeaderTransformer();
	transformer.setRemoveOnly(true); // Forwarded Header 신뢰하지 않고 제거하기
    
    return transformer;
}

 

3. Filters

• WebHandler가 요청을 처리하기 전/후에 공통 로직을 적용하는 필터
• 여러 WebFilter가 체인 형태로 실행된 뒤 최종적으로 WebHandler가 요청을 처리함
• ✅ 공통 처리에 사용됨
• ✅ @Order or Ordered 구현으로 우선순위 지정 가능
• ✅ Bean 등록 시 자동 적용됨
• ex) 인증, 로깅, CORS, 요청/응답 가공 등

 

CORS

• Controller에 Annotation을 사용해 CORS를 세부적으로 설정할 수 있음
• ✅ Spring Security와 함께 사용할 경우 CorsWebFilter 빈 등록이 권장됨 (Security보다 CORS가 먼저 처리되야 함)

 

예제) Controller에 직접 적용

@RestController
@RequestMapping("/api")
@CrossOrigin(origins = "http://localhost:3000")
public class UserController {

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

 

예제) CorsWebFilter 전체 적용

@Configuration
public class CorsConfig {

    @Bean
    public CorsWebFilter corsWebFilter() {
        CorsConfiguration config = new CorsConfiguration();

        config.addAllowedOrigin("http://localhost:3000");
        config.addAllowedMethod("*");
        config.addAllowedHeader("*");
        config.setAllowCredentials(true);

        UrlBasedCorsConfigurationSource source =
                new UrlBasedCorsConfigurationSource();

        source.registerCorsConfiguration("/**", config);

        return new CorsWebFilter(source);
    }
}

 

예제) Spring Security 사용

@Bean
SecurityWebFilterChain securityWebFilterChain(ServerHttpSecurity http) {

    return http
            .cors(cors -> {})
            .csrf(ServerHttpSecurity.CsrfSpec::disable)
            .authorizeExchange(exchange -> exchange
                    .anyExchange().authenticated()
            )
            .build();
}

• Request → CORS 처리 → Spring Security Filter Chain → WebHandler
• SecurityWebFilterChain.cors() → Spring Security에서 CORS 처리를 활성화함
• 이후 CorsWebFilter 빈 등록하면 적용됨 

 

URL Handler

 

4. Exceptions

• WebExceptionHandler는 WebFilter 체인이나 최종 WebHandler에서 발생한 예외를 처리하는 컴포넌트
• Request → WebFilter → WebHandler → Exception → WebExceptionHandler
• ✅ Spring Bean으로 등록하면 WebFlux 예외 처리 체인에 포함됨
• ✅ 여러 개 등록 가능
• ✅ @Order 또는 Ordered를 통해 실행 우선순위 지정 가능

 

구현체

구현체	역할
ResponseStatusExceptionHandler	ResponseStatusException의 상태 코드를 HTTP Response에 설정
WebFluxResponseStatusExceptionHandler	ResponseStatusExceptionHandler 확장 + 예외 클래스의 @ResponseStatus도 인식

 

코드) ResponseStatusExceptionHandler

@GetMapping("/users/{id}")
public Mono<User> getUser(@PathVariable Long id) {
    return userService.findById(id)
            .switchIfEmpty(Mono.error(
                    new ResponseStatusException(
                            HttpStatus.NOT_FOUND,
                            "User not found"
                    )
            ));
}

• ResponseStatusExceptionHandler → 404 Not Found

 

코드) WebFluxResponseStatusExceptionHandler

@ResponseStatus(HttpStatus.NOT_FOUND)
public class UserNotFoundException extends RuntimeException {

    public UserNotFoundException() {
        super("User not found");
    }
}

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

• UserNotFoundException
• → @ResponseStatus(HttpStatus.NOT_FOUND)
• → WebFluxResponseStatusExceptionHandler
• → 404 Not Found

 

5. Codecs

• HTTP에서 전달되는 byte 데이터와 Java 객체 사이를 변환하는 기능
• ✅ Non-Blocking I/O + Reactive Streams Backpressure 기반
• ✅ HTTP Body(byte) → Decoder / HttpMessageReader → Java Object
• ✅ Java Object → Encoder / HttpMessageWriter → HTTP Body(byte)

 

Jackson JSON

• Jackson 라이브러리가 존재하면 WebFlux에서 JSON ↔ Java 객체 변환을 지원함
• ✅ Mono와 Flux, Media Type에 따라 처리 방식이 달라짐

 

표) JacksonJsonEncoder

대상	처리 방식	핵심
Mono<T>	JSON 객체 하나를 완성한 뒤 JsonMapper로 Java 객체 생성	단일 객체 Decode
Flux<T>	JSON 객체 하나가 완성될 때마다 바로 Java 객체로 변환	다중 객체 Streaming Decode 가능
지원 형태	JSON Array, NDJSON, JSON Lines, JSON Text Sequences	객체 단위 처리 가능

• JSON byte stream → TokenBuffer → JsonMapper → Java Object

 

표) JacksonJsonEncoder

대상	처리 방식	핵심
Mono<T>	객체 하나를 JsonMapper로 JSON 직렬화	단일 객체 Encode
Flux<T> + application/json	여러 값을 모아 Collection으로 만든 뒤 JSON 배열로 직렬화	일반 JSON 응답
Flux<T> + Streaming Media Type	객체 하나씩 Encode → Write → Flush	실시간 Streaming
SSE	Event마다 JSON 인코딩 후 즉시 Flush	지연 없이 이벤트 전달

• Java Object → JsonMapper → JSON

 

표) String 처리

대상	처리 방식	핵심
String	기본적으로 Jackson JSON Codec이 처리하지 않음	CharSequenceEncoder 사용
Flux<String>	문자열 스트림으로 처리	JSON 배열로 자동 변환되지 않음
Flux<String> → JSON Array 필요	collectToList() → Mono<List<String>>로 변환 후 JSON 인코딩	JSON 배열 형태로 반환 가능

• Flux<String> → collectToList() → Mono<List<String>> → JSON Array

Form Data

Multipart 

Protocol Buffers

• Google이 만든 바이너리 직렬화 형식
• ✅ Java Object → ProtobufEncoder → Binary Data → HTTP
• ✅ HTTP Binary Data → ProtobufDecoder → Protobuf Message
• ✅ application/x-protobuf, application/octet-stream, application/vnd.google.protobuf를 지원

Codec	역할
ProtobufDecoder	Protobuf → 객체
ProtobufEncoder	객체 → Protobuf
ProtobufJsonDecoder	JSON → Protobuf Message
ProtobufJsonEncoder	Protobuf Message → JSON

 

Google Gson

Limits

• 일부 Decoder는 데이터를 객체로 만들기 위해 메모리에 일정량을 버퍼링함
• ✅ maxInMemorySize를 이용해 서버 측 Codec의 버퍼 최대 한도를 설정함
• ✅ HTTP Body → Buffer → maxInMemorySize 검사 → Decode

 

설정) limits

@Configuration
public class WebFluxConfig implements WebFluxConfigurer {

    @Override
    public void configureHttpMessageCodecs(ServerCodecConfigurer configurer) {
        configurer.defaultCodecs()
                .maxInMemorySize(2 * 1024 * 1024); // 2MB
    }
}

 

Streaming

• heartbeat를 주기적으로 보내 연결을 감지함
• ✅ text/event-stream, application/x-ndjson 지원
• ✅ 연결이 끊겼는데 아무 데이터를 보내지 않으면 서버가 클라이언트 연결 종료를 늦게 인식할 수 있음
• ➡️ 주기적으로 데이터를 보내는 것이 중요

 

DataBuffer

• WebFlux가 사용하는 byte buffer 추상화
• ✅ Codec의 입력 또는 출력으로 사용됨
• ✅ Netty ByteBuf / ByteBuffer → DataBuffer → Codec
• ✅ 서버마다 버퍼 Buffer 구현이 다를 수 있음
• ⚠️ Netty처럼 pooled buffer를 사용하는 환경에서 DataBuffer를 직접 다루는 경우, 사용 후 release 필수

 

설명) buffer pool

• netty는 네트워크 요청을 매우 많이 처리하기 때문에, 매 요청마다 새 byte buffer를 만들면 비용이 커질 수 있음
• ✅ 매번 새 메모리를 만드는 대신 기존 메모리를 재활용함
• ✅ Buffer Pool → ByteBuf 빌림 → 사용 → Pool에 반환 → 다음 요청에서 재사용
• ➡️ 소비한 후 release해야 메모리 누수를 피할 수 있음

 

6. Logging

• 비동기/멀티스레드 환경에서도 요청 단위로 로그를 추적할 수 있도록 설계됨
• ✅ DEBUG: 핵심 정보 위주로 간결하게 출력
• ✅ TRACE: DEBUG보다 더 상세한 정보를 제공하지만 불필요하게 과도한 로그는 지양
• ➡️  WebFlux에서는 하나의 요청이 여러 Thread에서 처리될 수 있으므로 Thread ID 대신 요청별 Log ID를 사용

 

Log Id

• ⚠️ WebFlux에서는 하나의 요청이 여러 Thread를 오갈 수 있기 때문에 Thread ID 만으로 같은 요청의 로그를 묶기 어려움
• ➡️ Thread가 아니라 Request 기준으로 로그를 추적함
• Server: ServerWebExchange에 'LOG_ID_ATTRIBUTE'로 저장됨
• Client: ClientRequest에 'LOG_ID_ATTRIBUTE'로 저장됨

 

코드) Server LogId 읽기

@Component
public class LoggingFilter implements WebFilter {

    private static final Logger log = LoggerFactory.getLogger(LoggingFilter.class);

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
        log.info("{} request: {} {}",
                exchange.getLogPrefix(),
                exchange.getRequest().getMethod(),
                exchange.getRequest().getURI());

        return chain.filter(exchange);
    }
}

• ServerWebExchange#getLogPrefix(): 서버의 Log ID Prefix 가져옴

 

코드) Client LogId 읽기

WebClient webClient = WebClient.builder()
        .filter((request, next) -> {
            log.info("{} request: {} {}",
                    request.logPrefix(),
                    request.method(),
                    request.url());

            return next.exchange(request);
        })
        .build();

• ClientRequest#logPrefix(): WebClient가 보내는 요청의 Log ID Prefix 가져옴

 

Sensitive Data

• Spring WebFlux는 기본적으로 Form Parameter/Header 마스킹 처리함.
• ⚠️ DEBUG나 TRACE 로그에는 민감한 정보가 포함될 수 있음 (Request Header, Form Parameter 등)
• ❗️ 전체 Request 정보를 로그에 출력하려면 명시적으로 활성화해야 함

 

설정) Server 상세 Logging

@Configuration
class MyConfig implements WebFluxConfigurer {

    @Override
    public void configureHttpMessageCodecs(ServerCodecConfigurer configurer) {

        configurer.defaultCodecs()
                .enableLoggingRequestDetails(true);
    }
}

 

설정) Client 상세 Logging

Consumer<ClientCodecConfigurer> consumer = configurer ->
        configurer.defaultCodecs()
                .enableLoggingRequestDetails(true);

WebClient webClient = WebClient.builder()
        .exchangeStrategies(strategies -> strategies.codecs(consumer))
        .build();

 

Appenders

Custom codecs

 

출처

• Spring WebFlux - Reactive Core