WebFlux 사용시 WebClient를 써야하는 이유

Asynchronous - Non Blocking framework가 화두가 된지 꽤 오랜 시간이 흘렀다. WebFlux, Akka, Node.js, Vert.X 같은 다양한 Asynchronous - Non Blocking 모델을 제공하는 Web Framework가 등장하였고, Spring 을 사용하는 다수의 팀들은 WebFlux를 선택하여 Blocking 모델인 WebMvc로 부터 벗어나 Asynchronous - Non Blocking을 지향하고 있다고 이야기하고 있다. 하지만 우리는 정말 Thread를 block 시키지 않도록 코드를 작성하고 있을까?? Asynchronous와 Non Blocking에 대한 이야기 부터 간단하게 소개하고 현재 우리는 어떤 방식으로 Asynchronous - Non Blocking 방식의 고성능 Server Application을 개발하고 있는지 고찰해보려고 한다.

Synchronous vs Asynchronous & Blocking vs Non Blocking

Synchronous vs Asynchronous & Blocking vs Non Blocking에대한 내용은 관점에 따라 조금씩 차이가 있는 설명들이 있지만 필자는 Thread 관점에서 살펴보려고한다. (메서드를 호출하는 호출자 관점으로 이해하여도 무방하다.)

Thread-A가 어떤 task를 Thread-B에게 전달 후 task의 결과를 기다리는 상황이다. task를 전달 후 바로 return 받지 못하고 있다면 blocking 모델이다.

return을 받은 후에는 Thread-B가 task를 수행한 결과를 이용하여 Thread-A가 후 처리를 하고 있다. task의 결과를 Thread-A(전달자)가 기다린 후 처리한다면 synchronous 모델이다.

어떠한 task를 다른 thread에게 전달 후 task가 끝날 때 까지 return을 받지 못하고 기다린다면 blocking 모델, task의 결과를 돌려받아 자기 자신이 처리한다면 synchronous 모델이라고 할 수 있다. 즉 위의 모델은 synchronous & blocking 모델이다.

Thread-A가 어떤 task를 Thread-B에게 전달 후 바로 응답을 받는 상황이다. 응답을 받은 후 Thread-A는 Thread-B에게 전달한 task의 결과를 더 이상 신경쓰지 않고 다른 일을 이어나간다. task를 전달 후 바로 return을 받아 다른 일을 처리하고 있다면 non blocking 모델이다.

task의 결과에 따른 후처리는 Thread-B가 처리한다. 이때, task의 결과에 따라 어떤 후처리를 진행할 건지 Thread-A가 Thread-B에게 callback method 형태로 task와 함께 전달하기도 한다. Thread-A 입장에서는 task를 전달 후 해당 결과에 대한 후처리를 전혀 신경쓰지 않는다. task 결과에 대한 후처리는 Thread-B가 책임을 지고 있는 상황이다. 이러한 상황은 Thread-A 시점에서는 asynchrounous 모델이다. 즉, 위의 모델은 asynchronous & non blocking 모델이다.

위의 내용들을 간단하게 다음과 같이 정리할 수 있다.

• blocking vs non blocking: task 전달 후 바로 return을 받을 수 있는지의 여부
• synchronous vs asynchronous: task의 결과에 대한 후 처리를 task를 전달한 thread가 처리하는지 task를 받은 thread가 처리하는지의 여부 (task를 전달한 thread의 시점)

위의 개념을 잘 이해하였다면 synchronous & non blocking, asynchronous & blocking 모델도 생각해볼수 있다.

위의 그림에서는 Thread-A가 task를 Thread-B에게 전달 후 바로 return 받아서 다른 일을 처리하지만, 다른 일을 처리 후 task의 결과를 기다리고 있는 모습을 볼 수 있다. Thread-A (task 전달자)가 결과에 대한 후처리를 진행하고 있기 때문이다. 즉, 위의 모델은 synchronous & non blocking 모델이다.

그렇다면 위와 같이 asynchronous & blocking 모델도 그려 볼 수 있다. 위의 내용들을 정리한 그림은 다음과 같다.

WebFlux & WebClient

spring 5.x 부터 webflux와 webclient가 등장하였고 asynchronous & non blocking 모델로 고성능의 web server가 필요할 때 선택할 수 있는 도구이다. webflux를 사용한 사례를 작성한 글이 많이 나오기 시작하였고, 여러 컨퍼런스에서도 webflux를 사용하면 asynchronous & non blocking 메커니즘으로 인해 servlet 기반의 request per thread 모델보다 더 나은 성능지표가 나온다고 소개하기 시작했다. webflux의 내장 http 서버로는 여러 선택지가 있지만 default는 reactor netty를 사용한다. reactor netty는 netty를 reactor로 wrapping한 framework이며, netty는 java nio를 더욱 손쉽게 사용할 수 있도록 wrapping한 network programming framework이다. 그렇다면 정말 webflux만 사용하면 asynchronous & non blocking 모델을 통해 모든 thread들이 block되지 않고 cpu를 잘 활용할 수 있게 되는 걸까? 그렇지 않다. webflux만으로는 요청을 처리하는 모든 flow를 asynchronous & non blocking하게 할 수 없다. 아직까지 jdbc는 blocking 모델이고, asynchronous & non blocking을 지원하는 network I/O client를 사용하지 않는다면 thread는 blocking되게 된다. 이를 해결하기 위해 대부분 webflux에서 사용하는 event loop worker thread외에 별도의 thread pool을 만들어 해당 thread pool에 blocking task를 전달함으로써 event loop worker thread만 blocking을 피하도록 한다. 결국 어떤지점에서는 thread가 block되는 것이다. kotlin coroutine은 복잡한 asynchronous 프로그래밍을 syntax sugar를 통해 imperative하게 짤 수 있도록 도와주는 도구이다. 이 또한 coroutine context thread pool의 thread들은 blocking이된다. 이런 blocking이 잘못된 것이 아니다. non blocking을 지원하지 않는 라이브러리를 써야만하는 상황은 굉장히 일반적이고, non blocking 흐름을 유지하려면 앞서 설명한 별도의 thread pool에 던져놓는 방식이 아니라면 마땅한 해결책이 없다. 다만, 정말 모든 thread를 blocking 하지 않는 흐름은 없는 걸까? (이 글에서는 설명하지 않지만 proactor pattern도 참고해보자.)

Java NIO

java io는 Stream으로 data를 read & write 하며 InputStream과 OutStream을 별도로 구성하여 read & write를 처리한다. 즉, 단반향이다. Stream을 read & write 할 때는 작업이 끝날 때까지 thread가 block된다. 이에 따라 여러 요청을 처리하려면 여러 thread가 필요하다. 하지만 java nio는 하나의 Channel에서 read & write 하며 (양방향) Selector를 통해 read & write가 끝났을 때 이벤트를 받아 작업을 이어나간다. 즉, thread를 block 시키지 않기 때문에 하나의 thread로 여러 요청을 처리할 수 있다. 단, java nio를 사용한다고 모든 io가 non blocking으로 처리되지는 않는다. non blocking을 지원하는 것이지 non blocking 모드로 설정 하지 않으면 blocking으로 동작한다. socket channel은 non blocking모드가 가능하지만 file channel은 non blocking 모드를 지원하지 않는다. java7에서 AsynchronousFileChannel을 제공하지만 select() 메커니즘과는 다른 방법으로 non blocking을 지원한다고 한다. (⇒ 내부 thread pool로 작업을 위임한 후 바로 return 하는 방식)

이에 더해 기존의 java io는 커널 버퍼를 직접 읽어올 방법이 없어 jvm의 내부로 복사 한후 사용한다. 커널 버퍼의 값을 jvm 내부로 복사할 때 thread들은 block 상태가 된다. java nio는 커널 버퍼를 직접 read & write 할 수 있는 ByteBuffer를 제공하기 때문에 kernel buffer에서 jvm heap으로 한 번더 copy를 하는 overhead가 없다. 이를 zero copy라고 하는데 해당 내용은 이번 글에서 다루지 않겠다.

Netty

Netty는 저수준의 nio를 wrapping한 network I/O framework이다. Netty관련 자세한 내용은 다른 글에서 다루고 있으니 해당 글을 참고하면 좋을 것 같다.

2020.06.14 - [Framework/Netty] - Netty의 스레드 모델

Netty의 스레드 모델

2020.05.17 - [Framework/Netty] - Netty의 기본 Component 및 Architecture

Netty의 기본 Component 및 Architecture

WebClient vs RestTemplate

WebClient는 reactor-netty의 HttpClient를 이용하는 spring 5에서 RestTemplate을 대체하기 위한 component이다. asynchronous & non blocking을 지원한다. WebFlux를 사용하면서 모든 thread를 block시키지 않고 network I/O 처리를 하려면 WebClient를 쓰는 것이 좋다. RestTemplate을 통해서도 Server의 thread를 block시키지 않고 network I/O를 처리 할 수 있다. 위에서 언급한 것처럼 별도의 thread pool에 task를 던져놓고 Server의 thread로 바로 return 해주면 된다. 하지만 thread pool의 thread는 block이 되기 때문에 모든 thread가 non blocking이 되는 모델은 아니다. Server의 thread를 block 시키는 대신 다른 thread를 block 시키는 것 뿐이다. thread pool의 thread가 모두 busy상태이고 queue까지 다 차버렸다면 Server thread는 block될 것이다. node.js의 event loop 모델의 문제점이 webflux에도 똑같이 있는 것이다.

Test

실제로 실험을 해보자. 실험을 위해 다음과 같은 준비를 해야한다.

1. api를 호출하면 query param으로 들어오는 delay 값 만큼 sleep 후 응답을 주는 간단한 api server를 하나 만든다.
2. WebFlux를 실행시키는 event loop의 worker의 thread 수를 1로 설정한다.
3. WebClient를 준비하는데, WebClient를 실행시키는 event loop의 worker thread 수를 1로 설정한다.
4. RestTemplate을 준비한다. Mono로 wrapping하며 subscribeOn에 사용할 scheduler의 thread 수를 1로 설정한다.
5. WebFlux를 호출하면 WebClient와 RestTemplate을 각각 이용하여 delay api server를 호출하도록한다.
6. visualvm을 통해 thread를 분석한다.

1 ~ 5번에 해당하는 코드는 다음과 같다.

Delay Api Server

@RestController
class Api {

    @GetMapping("/**")
    fun allListen(request: HttpServletRequest, @RequestParam("delay") delay: Long): String {
        println("------------------------------------")
        println("listen at: ${LocalDateTime.now()}")
        println("path: ${request.servletPath}")
        println("query param: ${request.queryString}")
        println("thread name: ${Thread.currentThread().name} id: ${Thread.currentThread().id}")
        println("------------------------------------")
        Thread.sleep(delay)
        return "OK"
    }
}

WebFlux Config

 

@Configuration
class ServerConfig {

    @Bean
    fun reactiveWebServerFactory(): ReactiveWebServerFactory {
        val factory = NettyReactiveWebServerFactory()
        factory.addServerCustomizers(NettyServerCustomizer { it.runOn(LoopResources.create("webflux", 1, true)) })

        return factory
    }
}

WebClient & RestTemplate Config

@Configuration
class HttpClientConfig {

    @Bean
    fun webClient(): WebClient {
        val httpClient = HttpClient.create()
            .doOnRequest { _, _ -> println("WebClient doOnRequest ${Thread.currentThread().name} : ${Thread.currentThread().id}") }
            .doAfterRequest { _, _ -> println("WebClient doAfterRequest ${Thread.currentThread().name} : ${Thread.currentThread().id}") }
            .doOnResponse { _, _ -> println("WebClient doOnResponse ${Thread.currentThread().name} : ${Thread.currentThread().id}") }
            .runOn(LoopResources.create("webclient", 1, true))
            .followRedirect(true)
        val reactorClientHttpConnector = ReactorClientHttpConnector(httpClient)

        return WebClient.builder()
            .clientConnector(reactorClientHttpConnector)
            .build()
    }

    @Bean
    fun restTemplate(): RestTemplate {
        val httpClient = HttpClientBuilder.create()
            .setRedirectStrategy(LaxRedirectStrategy())
            .setMaxConnTotal(10)
            .setMaxConnPerRoute(10)
            .build()

        val factory = HttpComponentsClientHttpRequestFactory()
        factory.httpClient = httpClient

        return RestTemplateBuilder()
            .requestFactory { factory }
            .additionalInterceptors(RestTemplateLoggingRequestInterceptor())
            .build()
    }

    class RestTemplateLoggingRequestInterceptor : ClientHttpRequestInterceptor {

        override fun intercept(
            request: HttpRequest,
            body: ByteArray,
            execution: ClientHttpRequestExecution,
        ): ClientHttpResponse {
            println("RestTemplate doOnRequest ${Thread.currentThread().name} : ${Thread.currentThread().id}")
            val response = execution.execute(request, body);
            println("RestTemplate doOnResponse ${Thread.currentThread().name} : ${Thread.currentThread().id}")

            return response
        }
    }
}

어떤 thread에서 실행하는지 확인하기위해 http 호출 전후의 thread들을 print하도록 설정하였다.

Delay Controller & DelayClient

@RestController
class DelayApi(private val delayClient: DelayClient) {

    @GetMapping("/api/delay-async")
    fun delayAsync(@RequestParam("delay") delay: Long): Mono<String> {
        return delayClient.invokeAsync(delay)
    }

    @GetMapping("/api/delay")
    fun delay(@RequestParam("delay") delay: Long): Mono<String> {
        return delayClient.invoke(delay)
    }
}

@Component
class DelayClient(
    @Value("\${delay.server.url}") private val url: String,
    private val webClient: WebClient,
    private val restTemplate: RestTemplate
) {

    private val scheduler = Schedulers.newSingle("play-delay-client")

    fun invokeAsync(delay: Long): Mono<String> {
        val uri: URI = UriComponentsBuilder.fromHttpUrl(url)
            .queryParam("delay", delay)
            .build()
            .encode()
            .toUri()


        return webClient.get()
            .uri(uri)
            .exchangeToMono {
                it.bodyToMono(String::class.java)
            }
    }

    fun invoke(delay: Long): Mono<String> {
        val uri: URI = UriComponentsBuilder.fromHttpUrl(url)
            .queryParam("delay", delay)
            .build()
            .encode()
            .toUri()

        return Mono.fromCallable {
            restTemplate.getForObject(uri, String::class.java) ?: "EMPTY"
        }
            .subscribeOn(scheduler)
    }

위의 코드를 두 개의 bash shell을 이용하여 동시에 실행 시키면 어떤 결과가 나올지 예상해보자.

1. Thread Pool의 수가 1인 Scheduler로 wrapping된 RestTemplate으로 delay api server를 동시에 두 번 호출 할 경우
2. EventLoop Worker Thread수가 1인 WebClient로 delay api server를 동시에 두 번 호출 할 경우

1번 case 부터 분석해보자.

curl -X GET "http://localhost:8080/api/delay?delay=10000"

delay param으로 10000ms(10s)를 전달하면 delay api server는 10초동안 sleep하게되어 응답을 10초 뒤에 줄 것이다. 위의 command를 두 개의 shell에서 동시에 실행 시킨후 webflux server와 delay api server에서 print되는 로그를 살펴보자.

webflux server log

delay api server log

thread dump

webflux server log를 보면 같은 thread에서 호출을 하는 것을 확인할 수 있으며 delay api server log를 통해 동시에 호출한 요청이 순차적으로 10초 간격으로 inbound된 것을 확인할 수 있다. 즉, thread가 block이 된 것이고 두 개의 요청을 처리하는데 20초가 걸린 것이다.

thread dump 를 보면 기존에 제공하는 java io를 이용하여 socket read를 기다리는 것을 확인할 수 있다.

이번에는 WebClient를 사용하는 2번 case를 분석해보자. http 요청에 사용하는 thread수는 1개라는 조건은 똑같다.

curl -X GET "http://localhost:8080/api/delay-async?delay=10000"

위의 endpoint로 webflux server 에 요청하면 webclient를 통해 delay api server를 호출하게된다.

위의 command를 1번 case 실험과 똑같이 두 shell에서 동시에 실행해보자.

webflux server log

delay api server log

thread dump

1번 case와는 다른 양상을 볼 수 있다. webflux server log는 1번 case와 같이 하나의 thread에서 실행하지만 request & response 로그가 순차적으로 찍히지 않았다. 이에더해 delay api server log를 보면 요청이 동시에 inbound 된 것을 확인할 수 있다. 즉, thread가 block이 되지 않았고 두 개의 요청을 처리하는데 20초가 아닌 10초만 걸렸다.

thread dump 를 보면 java nio를 이용하여 socket read를 기다리지 않는 것을 확인할 수 있다. (sun.nio.ch package는 java nio에서 제공하는 interface를 구현한 실제 구현체가 있는 package이다.)

이와 같은 실험을 통해 WebClient를 사용하는 WebFlux는 모든 thread를 block시키지 않고 network I/O를 처리 할 수 있다. 물론 jdbc와 같이 아직까지 non blocking을 지원하지 않는 도구를 이용한다면 1번 case와 같이 별도의 thread pool로 task를 던져버리고 server thread만이라도 block이 되지 않도록 해야한다. 하지만 적어도 network I/O 만큼은 thread를 block시키지 않게 개발한다면 cpu를 좀 더 효율적으로 사용할 수 있을 것이다.

 

오랜만에 글을 작성하다보니 글이 길어진 것 같다. 테스트 관련 글은 별도의 글로 포스팅하는 것을 고민하다가 한번에 살펴보시는게 좋을 것 같아 한번에 작성하게 되었다. 해당내용에 잘못된 내용이 있다면 댓글에 악플이라도 달아주었으면 좋겠다 ㅎㅎ

 

참고 글:

• http://eincs.com/2009/08/java-nio-bytebuffer-channel-file/

[NIO] JAVA NIO의 ByteBuffer와 Channel로 File Handling에서 더 좋은 Perfermance 내기!

• https://palpit.tistory.com/643

[Java] NIO 기반 입출력 및 네트워킹 - 파일 비동기 채널

• https://www.geeksforgeeks.org/non-blocking-server-in-java-nio/

Non Blocking Server in Java NIO - GeeksforGeeks