[Spring] WebFlux란?

Spring5에서 새롭게 추가된 Reactive 스타일의 어플리케이션 개발을 도와주는 Reactive-Stack Web Framework

특징

• 클라이언트와 서버에서 Reactive 애플리케이션 개발을 위한 non-blocking Reactive Stream(Reactor)을 지원한다.
• 적은 양의 스레드로 동시성을 처리한다.
• Functional Programming
• AsyncRestTemplate -> WebClient 
• DeferredResult -> Mono, Flux(Publisher)

---

SpringMVC와 WebFlux 비교

SpringMVC

• 하나의 요청에 대해 하나의 스레드가 사용된다.(thread-per-request) -> 다수의 요청을 대비하여 미리 스레드 풀을 생성해 놓으며, 각 요청마다 스레드를 할당하여 처리한다.
• 문제점
  • Blocking Call -> 동작 중인 스레드가 블로킹 상태가 되면 다른 스레드에서 CPU를 사용하기 위해 Context Switch가 발생한다. Blocking 로직으로 인한 잦은 Context Switch가 발생하면 오버헤드가 생긴다.
  • 많은 요청량 -> 요청량이 증가하면 스레드 수도 이에 비례하여 증가하고 서버가 감당하지 못할 만큼의 메모리를 먹을 수 있다.

WebFlux

• non-blocking과 고정된 스레드 수 만으로 모든 요청을 처리한다.
• 서버는 매우 적은 수의 스레드로 운영하며, CPU 코어 수 개수의 스레드를 가진 워커 풀을 생성하여 해당 워커 풀 내의 스레드로 모든 요청을 처리한다.
• non-blocking으로 코드를 작성해야 하며, 만일 blocking 라이브러리를 사용한다면, 워커 풀의 스레드가 아닌 외부 스레드로 요청을 처리해야 한다.

---

WebFlux 사용

리턴 타입을 Mono, Flux로 설정하고 반환하면, Spring에서 자동으로 subscribe를 호출한다.

 

| 코드

@Slf4j
@RestController
public class BasicController {
    public static final String Service1Uri = "http://localhost:9090/remote-service?req={req}";
    public static final String Service2Uri = "http://localhost:9090/remote-service2?req={req}";

    private final MyService myService;
    private final WebClient webClient = WebClient.create();

    public BasicController(MyService myService) {
        this.myService = myService;
    }

    /**
     * 의존성이 존재하는 경우
     */
    @GetMapping("/rest")
    public Mono<String> rest4(@RequestParam(required = false) int idx) {
        Mono<String> result = webClient
                .get()
                .uri(Service1Uri, idx)
                .retrieve()
                .bodyToMono(String.class);

        return Mono.just("Hello");
    }

}

위와 같이 코드를 실행하면, WebClient는 요청을 날리지 않고 Hello 응답 후 종료한다. -> Publisher는 subscribe하지 않으면 데이터를 생성하지 않기 때문이다.

 

@Slf4j
@RestController
public class BasicController {
    public static final String Service1Uri = "http://localhost:9090/remote-service?req={req}";
    public static final String Service2Uri = "http://localhost:9090/remote-service2?req={req}";

    private final MyService myService;
    private final WebClient webClient = WebClient.create();

    public BasicController(MyService myService) {
        this.myService = myService;
    }

    /**
     * 의존성이 존재하는 경우
     */
    @GetMapping("/rest")
    public Mono<String> rest4(@RequestParam(required = false) int idx) {
        Mono<String> result = webClient
                .get()
                .uri(Service1Uri, idx)
                .retrieve()
                .bodyToMono(String.class);

		result.subscribe((res) -> log.info("res = {}", res));

        return Mono.just("Hello");
    }

}

위와 같이 코드를 작성해야 WebClient 요청이 실행되고 res log가 출력된다.

 

@Slf4j
@RestController
public class BasicController {
    public static final String Service1Uri = "http://localhost:9090/remote-service?req={req}";
    public static final String Service2Uri = "http://localhost:9090/remote-service2?req={req}";

    private final MyService myService;
    private final WebClient webClient = WebClient.create();

    public BasicController(MyService myService) {
        this.myService = myService;
    }

    /**
     * 의존성이 존재하는 경우
     */
    @GetMapping("/rest")
    public Mono<String> rest4(@RequestParam(required = false) int idx) {
    
        return webClient
                .get()
                .uri(Service1Uri, idx)
                .retrieve()
                .bodyToMono(String.class);
    }

}

위와 같이 코드를 작성하면 Spring에서 subscribe()를 호출하기 때문에 정상적으로 처리된다.

 

@Slf4j
@RestController
public class BasicController {
    public static final String Service1Uri = "http://localhost:9090/remote-service?req={req}";
    public static final String Service2Uri = "http://localhost:9090/remote-service2?req={req}";

    private final MyService myService;
    private final WebClient webClient = WebClient.create();

    public BasicController(MyService myService) {
        this.myService = myService;
    }

    @GetMapping("/rest")
    public Mono<String> rest4(@RequestParam(required = false) int idx) {
        return webClient
                .get()
                .uri(Service1Uri, idx)
                .retrieve()
                .bodyToMono(String.class)
                .flatMap(res -> webClient.get().uri(Service2Uri, res).retrieve().bodyToMono(String.class))
                .flatMap(res -> Mono.fromCompletionStage(myService.work(res)));
    }

    @Service
    public static class MyService {
        @Async
        public CompletableFuture<String> work(String req) {
            return CompletableFuture
                    .completedFuture(req + "/asyncWork");
        }
    }

    @Bean
    ThreadPoolTaskExecutor threadPoolTaskExecutor() {
        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
        threadPoolTaskExecutor.setCorePoolSize(10);
        threadPoolTaskExecutor.setThreadNamePrefix("myThread");

        threadPoolTaskExecutor.initialize();
        return threadPoolTaskExecutor;
    }
}

이전 포스팅의 최종 코드는 위와 같이 수정될 수 있다.

* Service 로직을 비동기적으로 실행시키지 않으면, reactor-http-nio thread를 사용하여 로직이 실행된다. -> reactor-http-nio thread를 사용할 것인지, 새로운 thread pool을 만들어 비동기적으로 실행시킬지 선택해서 코드를 작성해야 한다.

 

* REF

https://www.youtube.com/watch?v=ScH7NZU_zvk&list=PLOLeoJ50I1kkqC4FuEztT__3xKSfR2fpw&index=8