[리액티브 프로그래밍] Spring WebFlux의 요청 처리 흐름
- Coding/Reactive
- 2023. 5. 2.
Spring WebFlux
Spring WebFlux는 리액티브 웹 애플리케이션 구현을 위해 Spring 5.0부터 지원하는 리액티브 웹 프레임워크다. 대량의 요청 트래픽을 Spring MVC 방식이 처리하지 못하는 상황이 잦아짐에 따라 적은 수의 스레드로 대량의 요청을 안정적으로 처리할 수 있는 비동기 Non-Blocking I/O 방식의 Spring WebFlux가 등장했다.
Spring WebFlux 기술 스택
1) 서버
Non-Blocking I/O 방식으로 동작하는 Netty 등의 서버 엔진에서 동작한다.
2) 서버 API
기본 서버 엔진이 Netty이지만 Jetty나 Undertow 같은 서버 엔진에서 지원하는 리액티브 스트림즈 어댑터를 통해 리액티브 스트림즈를 지원한다.
3) 보안
WebFliter를 이용해 Spring Security를 Spring WebFlux에서 사용한다.
4) 데이터 엑세스
데이터 엑세스 계층까지 Non-Blocking I/O를 지원할 수 있도록 Spring Data R2DBC 및 Non-Blocking I/O를 지원하는 NoSQL 모듈을 사용한다.
Spring WebFlux의 요청 처리 흐름
| 순서 | 처리 흐름에 대한 설명 |
| 1 | 최초에 클라이언트로부터 요청이 들어오면 Netty 등의 서버 엔진을 거쳐, HttpHandler가 들어오는 요청을 받는다. 각 서버의 ServerhttpRequest, ServerHttpResponse를 포함하는 ServerWebExchange를 생성한후, WebFliter 체인으로 전달한다. |
| 2 | 전달받은 ServerWebExchange는 전처리 과정을 거친 후, WebHandler 인터페이스의 구현체인 DispatcherHandler에게 전달된다. |
| 3 | DispatcherHandler에서는 HandlerMapping List를 원본 Flux의 소스로 전달받는다. |
| 4 | ServerWebExchange를 처리할 핸들러를 조회한다. |
| 5 | 조회한 핸들러의 호출을 HandlerAdapter에게 위임한다. |
| 6 | HandlerAdapter는 ServerWebExchange를 처리할 핸들러를 호출한다. |
| 7 | Controller 또는 HandlerFunction 형태의 핸들러에서 요청을 처리한 후, 응답 데이터를 리턴받는다. -> 핸들러에서 리턴한다고 표현했지만, 실제 핸들러에서 리턴되는 것은 응답 데이터를 포함하고 있는 Flux 또는 Mono Sequence 이다. 따라서 즉시 어떤 작업을 수행한다는 의미가 아니다. |
| 8 | 핸들러로부터 리턴받은 응답 데이터를 처리할 HandlerResultHandler를 조회한다. |
| 9 | 조회한 HandlerResultHandler가 응답 데이터를 적절하게 처리한 후, response로 리턴한다. |
그림과 설명을 함께 봐도, 정확하게 이해되지 않는다. 핵심 컴포넌트를 알아보자.
HttpHandler
HttpHandler는 다른 유형의 HTTP 서버 API로부터 request, response를 처리하기 위한 단 하나의 추상메서드를 갖는다.
HttpHandler.java
public interface HttpHandler {
/**
* Handle the given request and write to the response.
* @param request current request
* @param response current response
* @return indicates completion of request handling
*/
Mono<Void> handle(ServerHttpRequest request, ServerHttpResponse response);
}
위 HttpHandler.java의 구현체인 HttpWebHandlerAdapter.java
public class HttpWebHandlerAdapter extends WebHandlerDecorator implements HttpHandler {
...
@Override
public Mono<Void> handle(ServerHttpRequest request, ServerHttpResponse response) {
if (this.forwardedHeaderTransformer != null) {
try {
request = this.forwardedHeaderTransformer.apply(request);
}
catch (Throwable ex) {
if (logger.isDebugEnabled()) {
logger.debug("Failed to apply forwarded headers to " + formatRequest(request), ex);
}
response.setStatusCode(HttpStatus.BAD_REQUEST);
return response.setComplete();
}
}
ServerWebExchange exchange = createExchange(request, response);
LogFormatUtils.traceDebug(logger, traceOn ->
exchange.getLogPrefix() + formatRequest(exchange.getRequest()) +
(traceOn ? ", headers=" + formatHeaders(exchange.getRequest().getHeaders()) : ""));
return getDelegate().handle(exchange)
.doOnSuccess(aVoid -> logResponse(exchange))
.onErrorResume(ex -> handleUnresolvedError(exchange, ex))
.then(Mono.defer(response::setComplete));
}
...
}1) ServerWebExchange를 생성
handler() 메서드의 파라미터로 전달받은 ServerHttpRequest, ServerHttpResponse로 ServerWebExchange를 생성한다.
ServerWebExchange exchange = createExchange(request, response);
2) WebHandler을 호출한다.
return getDelegate().handle(exchange)
.doOnSuccess(aVoid -> logResponse(exchange))
.onErrorResume(ex -> handleUnresolvedError(exchange, ex))
.then(Mono.defer(response::setComplete));
// HttpWebHandlerAdapter.java
public WebHandler getDelegate() {
return this.delegate;
}
WebFilter
핸들러가 요청을 처리하기 전에 전처리 작업을 수행할 수 있도록 해준다. 애플리케이션 내에 정의된 모든 핸들러에 공통으로 동작한다.
WebFilter.java
public interface WebFilter {
Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain);
}파라미터로 받은 WebFilterChain 을 통해 필터 체인을 형성하여 원하는 만큼의 WebFilter를 추가할 수 있다.
WebFilter 예제
@Component
public class BookLogFilter implements WebFilter {
@Override
public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
String path = exchange.getRequest().getURI().getPath();
return chain.filter(exchange).doAfterTerminate(() -> {
if (path.contains("books")) {
System.out.println("path: " + path + ", status: " +
exchange.getResponse().getStatusCode());
}
});
}
}Book 리소스에 request 시에만 로그를 출력한다.
1) doAfterTerminate()
종료 이벤트(onComplete, onError) 발생시, 수행된다.
HandlerFilterFunction
함수형 기반의 요청 핸들러에 적용할 수 있는 Filter로, 함수형 기반의 핸들러에서만 동작한다. 따라서 함수형 기반의 핸들러에서만 제한적으로 필터링 작업을 수행하고 싶을때 사용한다.
@FunctionalInterface
public interface HandlerFilterFunction<T extends ServerResponse, R extends ServerResponse> {
Mono<R> filter(ServerRequest request, HandlerFunction<T> next);
...
}파라미터로 전달받은 HandlerFunction에 연결된다.
HandlerFilterFunction 예제
public class BookRouterFunctionFilter implements HandlerFilterFunction {
@Override
public Mono<ServerResponse> filter(ServerRequest request, HandlerFunction next) {
String path = request.requestPath().value();
return next.handle(request).doAfterTerminate(() -> {
System.out.println("path: " + path + ", status: " +
request.exchange().getResponse().getStatusCode());
});
}
}Book 리소스에 request 시에만 로그를 출력한다.
위 WebFliter의 구현체는 Spring Bean으로 등록되는 반면, HandlerFilterFunction의 구현체는 애너테이션 기반의 핸들러가 아닌 함수형 기반의 요청 핸들러에서 함수 형태로 사용되기 때문에 Spring Bean으로 등록되지 않는다.
HandlerFilterFunction 적용 예제
아래 API 요청이 오면, 정의된 Response가 내려가고, filter 안의 로그가 찍힌다.
@Configuration
public class BookRouterFunction {
@Bean
public RouterFunction routerFunction() {
return RouterFunctions
.route(GET("/v1/router/books/{book-id}"),
(ServerRequest request) -> this.getBook(request))
.filter(new BookRouterFunctionFilter());
}
public Mono<ServerResponse> getBook(ServerRequest request) {
return ServerResponse
.ok()
.body(Mono.just(BookDto.Response.builder()
.bookId(Long.parseLong(request.pathVariable("book-id")))
.bookName("Advanced Reactor")
.author("Tom")
.isbn("222-22-2222-222-2").build()), BookDto.Response.class);
}
}함수형 요청 핸들러를 라우팅해 주는 RouterFunction의 filter() 메서드에 파라미터로 BookRouterFunctionFilter를 전달해서 필터링 작업을 수행하고있다.
DispatcherHandler
WebHandler 인터페이스의 구현체다. Spring MVC에서 Front Controller 패턴이 적용된 DispatcherServlet처럼 중앙에서 먼저 요청을 전달받은 후에 다른 컴포넌트에 요청 처리를 위임한다.
DispatcherHandler 자체가 Spring Bean으로 등록되도록 설계되어있고, ApplicationContext에서 HandlerMapping, HandlerAdapter, HandlerResultHandler 등의 요청 처리를 위한 위임 컴포넌트를 검색한다.
public class DispatcherHandler implements WebHandler, PreFlightRequestHandler, ApplicationContextAware {
@Nullable
private List<HandlerMapping> handlerMappings;
@Nullable
private List<HandlerAdapter> handlerAdapters;
@Nullable
private List<HandlerResultHandler> resultHandlers;
public DispatcherHandler() {
}
...
protected void initStrategies(ApplicationContext context) {
Map<String, HandlerMapping> mappingBeans = BeanFactoryUtils.beansOfTypeIncludingAncestors(
context, HandlerMapping.class, true, false);
ArrayList<HandlerMapping> mappings = new ArrayList<>(mappingBeans.values());
AnnotationAwareOrderComparator.sort(mappings);
this.handlerMappings = Collections.unmodifiableList(mappings);
Map<String, HandlerAdapter> adapterBeans = BeanFactoryUtils.beansOfTypeIncludingAncestors(
context, HandlerAdapter.class, true, false);
this.handlerAdapters = new ArrayList<>(adapterBeans.values());
AnnotationAwareOrderComparator.sort(this.handlerAdapters);
Map<String, HandlerResultHandler> beans = BeanFactoryUtils.beansOfTypeIncludingAncestors(
context, HandlerResultHandler.class, true, false);
this.resultHandlers = new ArrayList<>(beans.values());
AnnotationAwareOrderComparator.sort(this.resultHandlers);
}
@Override
public Mono<Void> handle(ServerWebExchange exchange) {
if (this.handlerMappings == null) {
return createNotFoundError();
}
if (CorsUtils.isPreFlightRequest(exchange.getRequest())) {
return handlePreFlight(exchange);
}
return Flux.fromIterable(this.handlerMappings)
.concatMap(mapping -> mapping.getHandler(exchange))
.next()
.switchIfEmpty(createNotFoundError())
.flatMap(handler -> invokeHandler(exchange, handler))
.flatMap(result -> handleResult(exchange, result));
}
...
private Mono<HandlerResult> invokeHandler(ServerWebExchange exchange, Object handler) {
if (ObjectUtils.nullSafeEquals(exchange.getResponse().getStatusCode(), HttpStatus.FORBIDDEN)) {
return Mono.empty(); // CORS rejection
}
if (this.handlerAdapters != null) {
for (HandlerAdapter handlerAdapter : this.handlerAdapters) {
if (handlerAdapter.supports(handler)) {
return handlerAdapter.handle(exchange, handler);
}
}
}
return Mono.error(new IllegalStateException("No HandlerAdapter: " + handler));
}
private Mono<Void> handleResult(ServerWebExchange exchange, HandlerResult result) {
return getResultHandler(result).handleResult(exchange, result)
.checkpoint("Handler " + result.getHandler() + " [DispatcherHandler]")
.onErrorResume(ex ->
result.applyExceptionHandler(ex).flatMap(exResult -> {
String text = "Exception handler " + exResult.getHandler() +
", error=\"" + ex.getMessage() + "\" [DispatcherHandler]";
return getResultHandler(exResult).handleResult(exchange, exResult).checkpoint(text);
}));
}
private HandlerResultHandler getResultHandler(HandlerResult handlerResult) {
if (this.resultHandlers != null) {
for (HandlerResultHandler resultHandler : this.resultHandlers) {
if (resultHandler.supports(handlerResult)) {
return resultHandler;
}
}
}
throw new IllegalStateException("No HandlerResultHandler for " + handlerResult.getReturnValue());
}
...
}1) protected void initStrategies(ApplicationContext context)
BeanFactoryUtils를 이용해 ApplicationContext로부터 HandlerMapping Bean, HandlerAdapter Bean, HandlerResultHandler Bean을 검색한 후 각각의 타입인 List 객체를 생성한다.
2) public Mono<Void> handle(ServerWebExchange exchange)
List<HandlerMapping>을 Flux.fromIterable()의 원본 데이터 소스로 입력받은 후에 getHandler() 메서드를 통해 매치되는 Handler 중에서 첫번째 핸들러를 사용한다.
3) private Mono<HandlerResult> invokeHandler(ServerWebExchange exchange, Object handler)
실제 핸들러 호출은 invokeHandler() 내부에서 Handler 객체와 매핑되는 HandlerAdapter를 통해서 이뤄진다.
4) private HandlerResultHandler getResultHandler(HandlerResult handlerResult)
실제 응답 처리는 handleResult() 내부에서 호출한 getResultHandler()에서 HandlerResult 객체와 매핑되는 HandlerResultHandler를 통해서 이뤄진다.
HandlerMapping
request와 handler object에 대한 매핑을 정의하는 인터페이스이다. HandlerMapping 인터페이스를 구현하는 구현 클래스로는 RequestMappingHandlerMapping, RouterFunctionMapping 등이 있다.
HandlerMapping.java
public interface HandlerMapping {
...
Mono<Object> getHandler(ServerWebExchange exchange);
}getHandler() 메서드는 파라미터로 입력받은 ServerWebExchange에 매치되는 handler object를 리턴한다.
HandlerAdapter
HandlerAdapter는 HandlerMapping을 통해 얻은 핸들러를 직접적으로 호출을 하며, 응답 결과로 Mono<HandlerResult>를 리턴받는다.
HandlerAdapter.java
public interface HandlerAdapter {
boolean supports(Object handler);
Mono<HandlerResult> handle(ServerWebExchange exchange, Object handler);
}1) boolean supports(Object handler)
파라미터로 전달받은 handler object를 지원하는지 체크한다.
2) Mono<HandlerResult> handle(ServerWebExchange exchange, Object handler)
파라미터로 전달받은 handler object를 통해 핸들러 메서드를 호출한다.
다른 그림으로 다시 정리
1) 클라이언트 요청
2) HttpHandler가 요청을 받고 ServerWebExchange(ServerhttpRequest, ServerHttpResponse)를 생성3) 하여 WebFilter 체인으로 전달
4) DIspatcherHandler에게 전달
5) HandlerMapping에서 ServerWebExchange을 처리할 핸들러 조회
6) HandlerAdapater가 요청을 위임받음
7) 실제 핸들러 호출
8) HandlerResultHandler을 거쳐 response 반환
'Coding > Reactive' 카테고리의 다른 글
| ReactiveStreams 구현 라이브러리별 예제코드 정리 (Reactor, RxJava, Munity) (0) | 2024.01.16 |
|---|---|
| [리액티브 프로그래밍] 리액티브 스트림즈(Reactive Streams) 핵심 컴포넌트의 동작과정 (0) | 2023.06.11 |
| [리액티브 프로그래밍] Reactor의 debugging (0) | 2023.04.30 |
| [리액티브 프로그래밍] Reactor의 Context (0) | 2023.04.29 |
| [리액티브 프로그래밍] Reactor의 Scheduler (0) | 2023.04.28 |