Spring Webflux 실습 - 2

간단한 spring boot 서버를 만들어 줍니다.

`mvc` 서버라고 지칭하겠습니다.

package com.example.mvc;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import java.util.Map;

@SpringBootApplication
@RestController
public class MvcApplication {

    public static void main(String[] args) {
        SpringApplication.run(MvcApplication.class, args);
    }

    @GetMapping("/posts/{id}")
    public Map<String, String> getPosts(@PathVariable(name = "id") Long id) throws Exception{
        Thread.sleep(300);
        if(id > 10L){
            throw new Exception("Too big");
        }
        return Map.of("id", id.toString(), "content", "Posts content is %d".formatted(id));
    }
}

이전 시간에 만든 webflux 서버를 활용하겠습니다.

아래 클래스를 만듭니다.

1. PostContrller

2. PostService

3. PostClient

4. WebClientConfig

PostController

PostController는 Spring WebFlux 기반의 REST API 엔드포인트를 제공하는 컨트롤러입니다. 클라이언트로부터 들어오는 요청을 처리하고, 서비스 계층인 PostService를 호출하여 데이터를 처리한 후 응답을 반환합니다.

package com.example.webflux1.controller;

import com.example.webflux1.dto.PostResponse;
import com.example.webflux1.service.PostService;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.List;

@RestController
@RequiredArgsConstructor
@RequestMapping("/posts")
public class PostController {

    private final PostService postService;

    @GetMapping("/{id}")
    public Mono<PostResponse> getPostContent(@PathVariable(name="id") Long id){
        return postService.getPostContent(id);
    }

    @GetMapping("/search1")
    public Flux<PostResponse> getMultiplePostContent(@RequestParam(name="ids") List<Long> ids){
        return postService.getMultiplePostContent(ids);
    }

    @GetMapping("/search2")
    public Flux<PostResponse> getMultiplePostContent(@RequestParam(name="ids") List<Long> ids){
        return postService.getParallelMultiplePostContent(ids);
    }
}

1. getPostContent 메서드 (/posts/{id})

@GetMapping("/{id}")
public Mono<PostResponse> getPostContent(@PathVariable(name="id") Long id){
    return postService.getPostContent(id);
}

• 설명:
  • 이 엔드포인트는 특정 게시글 ID에 대한 내용을 가져오는 역할을 합니다.
  • URL 경로: /posts/{id}. 예를 들어 /posts/1과 같은 방식으로 요청을 보냅니다.
  • 경로 변수: {id}는 게시글의 ID입니다. 클라이언트가 ID 값을 전달하면 그 값에 해당하는 게시글을 비동기적으로 조회합니다.
  • 반환 값: Mono<PostResponse>는 단일 게시글의 내용을 포함하는 비동기 응답을 의미합니다.
  • 서비스 계층에서 getPostContent 메서드를 호출해 해당 ID의 게시글을 조회합니다.

2. getMultiplePostContent (순차적 비동기) 메서드 (/posts/search1)

@GetMapping("/search1")
public Flux<PostResponse> getMultiplePostContent(@RequestParam(name="ids") List<Long> ids){
    return postService.getMultiplePostContent(ids);
}

• 설명:
  • 이 엔드포인트는 여러 게시글의 ID 리스트를 받아서, 각각의 게시글을 순차적으로 비동기 처리하는 역할을 합니다.
  • URL 경로: /posts/search1. 예를 들어 /posts/search1?ids=1,2,3 과 같은 방식으로 여러 ID에 대해 요청을 보냅니다.
  • 쿼리 매개변수: ids는 요청 URL의 파라미터로 전달되는 ID 리스트입니다. 이 리스트에 있는 각각의 ID에 대해 게시글을 순차적으로 조회합니다.
  • 반환 값: Flux<PostResponse>는 다중 게시글의 내용을 포함하는 비동기 스트림을 의미합니다.
  • 순차적 비동기 처리: PostService의 getMultiplePostContent 메서드는 순차적으로 ID 리스트에 있는 게시글을 하나씩 비동기적으로 처리합니다. 비록 비동기이지만, 요청된 순서대로 게시글이 반환됩니다.

3. getMultiplePostContent (병렬 비동기) 메서드 (/posts/search2)

@GetMapping("/search2")
public Flux<PostResponse> getMultiplePostContent(@RequestParam(name="ids") List<Long> ids){
    return postService.getParallelMultiplePostContent(ids);
}

• 설명:
  • 이 엔드포인트는 여러 게시글의 ID 리스트를 받아서, 각각의 게시글을 병렬로 비동기 처리하는 역할을 합니다.
  • URL 경로: /posts/search2. 예를 들어 /posts/search2?ids=1,2,3 과 같은 방식으로 여러 ID에 대해 요청을 보냅니다.
  • 쿼리 매개변수: ids는 요청 URL의 파라미터로 전달되는 ID 리스트입니다. 이 리스트에 있는 각각의 ID에 대해 게시글을 병렬로 조회합니다.
  • 반환 값: Flux<PostResponse>는 다중 게시글의 내용을 포함하는 비동기 스트림을 의미합니다.
  • 병렬 비동기 처리: PostService의 getParallelMultiplePostContent 메서드는 병렬 스레드 풀을 사용하여, 여러 게시글을 동시에 조회하여 처리 속도를 높입니다. 결과는 다시 순차적으로 반환되지만, 처리 속도는 병렬 처리를 통해 더 빠르게 처리됩니다.

차이점 요약

• /search1 (순차적 비동기): ID 리스트에 있는 게시글을 순차적으로 비동기 처리하여 반환합니다. 각 ID에 대해 처리된 결과는 순서대로 반환됩니다.
• /search2 (병렬 비동기): ID 리스트에 있는 게시글을 병렬로 비동기 처리하여 더 빠르게 처리합니다. 결과는 처리 완료 후 순차적으로 조합됩니다.

두 메서드 모두 비동기적으로 데이터를 처리하지만, search1은 순차적 비동기 처리를, search2는 병렬로 비동기 처리를 수행하는 차이가 있습니다.

PostService

PostService는 외부 API 호출을 담당하는 PostClient를 호출하여 비즈니스 로직을 처리합니다. 비동기 호출과 에러 핸들링을 포함한 두 가지 중요한 메서드를 가지고 있습니다.

package com.example.webflux1.service;

import com.example.webflux1.client.PostClient;
import com.example.webflux1.dto.PostResponse;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.util.List;

@Service
@RequiredArgsConstructor
public class PostService {
    private final PostClient postClient;

    public Mono<PostResponse> getPostContent(Long id){
        return postClient.getPost(id)
                .onErrorResume(error -> Mono.just(new PostResponse(id.toString(), "Fallback data %d".formatted(id))));
    }

    public Flux<PostResponse> getMultiplePostContent(List<Long> ids){
        return Flux.fromIterable(ids)
                .flatMap(this::getPostContent)
                .log();
    }

    public Flux<PostResponse> getParallelMultiplePostContent(List<Long> ids){
        return Flux.fromIterable(ids)
                .parallel()
                .runOn(Schedulers.parallel())
                .flatMap(this::getPostContent)
                .log()
                .sequential();
    }
}

PostService 클래스에 대한 설명에서 메커니즘이 사용된 이유와 그 효과를 더 깊이 설명해 보겠습니다.

---

주요 기능 및 메커니즘 설명:

1. getPostContent(Long id) 메서드

비동기 호출 및 Fallback 기능

• 비동기 호출: 이 메서드는 PostClient를 통해 외부 API에 비동기적으로 요청을 보냅니다. Spring WebFlux의 WebClient는 기본적으로 논블로킹 방식으로 작동하므로, 응답을 기다리는 동안 쓰레드를 점유하지 않습니다. 이는 동시 요청이 많은 상황에서 시스템의 효율성을 높이는 데 유리합니다.
• onErrorResume: onErrorResume은 비동기 처리 중 오류가 발생했을 때 오류를 처리하는 리액터 연산자입니다. 외부 API 호출에 실패할 경우, 기본값을 제공하는 Fallback 처리 로직이 있습니다.
  • 사용된 이유: API 호출 중 네트워크 오류나 서버 다운과 같은 예외 상황이 발생할 수 있습니다. 이런 경우 시스템이 다운되거나 장애를 일으키지 않도록 onErrorResume을 사용하여 오류를 처리하고, 대신에 Fallback 데이터를 반환합니다.
  • 효과: Fallback 기능을 통해 시스템이 높은 가용성을 유지하며, 클라이언트는 예상하지 못한 장애에 영향을 받지 않고 기본값을 받을 수 있습니다. 특히 마케팅 서비스나 인플루언서 분석과 같은 API에서 부분 데이터만으로도 충분히 기능을 제공할 수 있을 때, 이러한 Fallback 전략은 유용합니다.

---

2. getMultiplePostContent(List<Long> ids) 메서드

순차적 비동기 처리

• 비동기 방식으로 여러 게시글을 처리: 이 메서드는 주어진 게시글 ID 목록에 대해 각 ID마다 비동기적으로 getPostContent 메서드를 호출합니다. 여기서 사용되는 리액터 연산자인 flatMap은 비동기 작업을 처리하는 데 핵심적인 역할을 합니다.
  • 사용된 이유: flatMap은 각 게시글 ID에 대해 비동기 호출을 수행하는데, 이를 통해 요청이 병렬적으로 실행될 수 있습니다. 순차적으로 처리하지만 비동기이기 때문에 응답 시간이 빨라지고, 리소스 사용을 효율적으로 할 수 있습니다.
  • 효과: 비동기 처리를 통해 동시에 여러 요청을 처리할 수 있습니다. WebFlux의 논블로킹 특성 덕분에 대기 시간이 적고, 각 게시글 API 요청이 비동기적으로 처리되기 때문에 성능을 최적화할 수 있습니다. 이는 요청량이 많은 시스템에서 매우 효과적입니다.

---

3. getParallelMultiplePostContent(List<Long> ids) 메서드

병렬 처리

• parallel 스트림 사용 및 Schedulers.parallel(): 이 메서드는 각 게시글 요청을 병렬로 처리합니다. parallel() 연산자를 통해 요청을 여러 스레드에서 병렬로 실행하며, Schedulers.parallel()을 사용해 병렬 작업을 위한 전용 스레드 풀에서 작업을 실행합니다.
  • 사용된 이유: 다수의 게시글을 처리하는 상황에서 순차적 비동기 처리만으로는 성능의 한계가 있을 수 있습니다. 특히 처리해야 할 데이터가 많고 각 요청이 시간이 오래 걸릴 경우, 병렬 처리를 통해 작업을 동시에 수행할 수 있습니다.
  • 효과: 병렬 처리를 통해 여러 요청을 동시에 빠르게 처리할 수 있습니다. 이를 통해 API 호출 성능을 극대화하며, 처리 속도가 크게 향상됩니다. 예를 들어, 10개의 API 요청을 순차적으로 처리하는 경우 10초가 걸릴 수 있지만, 병렬 처리를 사용하면 그 시간이 2~3초로 줄어들 수 있습니다. Schedulers.parallel()을 사용함으로써 각 요청이 병렬로 수행되고 CPU 코어를 더 잘 활용할 수 있게 됩니다.
  • sequential(): 병렬 처리된 결과들을 sequential()로 다시 순차적으로 정렬하여 응답합니다. 병렬 처리를 통해 성능을 높이되, 최종적으로 클라이언트에게는 일관된 순서로 데이터를 전달할 수 있게 합니다.

---

결론 및 효과 요약

• 비동기 처리: 각 메서드는 비동기 방식으로 동작하며, 특히 WebFlux와 WebClient를 활용한 비동기 호출은 서버의 리소스를 효율적으로 사용하게 만듭니다. 결과적으로 높은 성능과 확장성을 제공하며, 특히 대규모 트래픽을 처리할 때 유리합니다.
• onErrorResume(Fallback): 비동기 호출 중 발생할 수 있는 오류를 처리하고, 기본값을 제공하는 Fallback 전략을 통해 서비스 가용성을 높입니다. 장애 발생 시에도 시스템이 다운되지 않고 최소한의 응답을 제공할 수 있습니다.
• 순차적 비동기 처리(flatMap): 여러 API 호출을 비동기적으로 처리하여, 각 요청이 완료될 때까지 기다리지 않고 동시에 여러 요청을 처리할 수 있게 만듭니다. 이를 통해 성능을 개선하고 대기 시간을 줄일 수 있습니다.
• 병렬 처리(parallel() + Schedulers.parallel()): 다수의 요청을 병렬로 처리하여 성능을 극대화하며, 다중 스레드를 활용해 짧은 시간 안에 많은 데이터를 처리할 수 있습니다. 특히 많은 API 요청을 처리하는 시스템에서 병렬 처리는 성능을 획기적으로 향상시킵니다.

이러한 메커니즘들을 통해 WebFlux 기반 시스템은 높은 성능과 확장성을 갖춘 비동기 논블로킹 서버를 구현할 수 있게 됩니다.

---

PostClient - WebClient를 통한 외부 API 호출

PostClient는 WebClient를 사용하여 MVC 서버의 API를 비동기적으로 호출하는 클라이언트 역할을 합니다. WebClient는 Spring WebFlux의 비동기 HTTP 클라이언트로, 효율적인 네트워크 요청을 처리할 수 있습니다.

package com.example.webflux1.client;

import com.example.webflux1.dto.PostResponse;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.util.UriComponentsBuilder;
import reactor.core.publisher.Mono;

@Service
@RequiredArgsConstructor
public class PostClient {

    private final WebClient webClient;
    private final String url = "http://127.0.0.1:8090";

    public Mono<PostResponse> getPost(Long id){
        String uriString = UriComponentsBuilder.fromHttpUrl(url)
                .path("/posts/%d".formatted(id))
                .buildAndExpand()
                .toUriString();

        return webClient.get()
                .uri(uriString)
                .retrieve()
                .bodyToMono(PostResponse.class);
    }

}

주요 기능:

• WebClient를 사용하여 MVC 서버의 GET /posts/{id} API를 비동기적으로 호출합니다.
• UriComponentsBuilder를 통해 동적으로 URL을 생성하며, **retrieve()**와 **bodyToMono()**를 사용하여 응답 데이터를 비동기적으로 처리합니다.

WebClientConfig - WebClient 설정

WebClientConfig는 WebClient를 빈으로 등록하여 애플리케이션 전역에서 사용할 수 있게 구성하는 클래스입니다.

package com.example.webflux1.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.reactive.function.client.WebClient;

@Configuration
public class WebClientConfig {

    @Bean
    public WebClient webClient(){
        return WebClient.builder().build();
    }
}

주요 기능:

• WebClient.builder().build()를 사용하여 WebClient를 설정하고, 이를 애플리케이션 내에서 주입받아 사용할 수 있게 빈으로 등록합니다.

비동기와 병렬 처리의 차이

이 예제에서 중요한 개념 중 하나는 비동기 처리와 병렬 처리의 차이입니다. getMultiplePostContent 메서드는 순차적으로 비동기 처리를 하는 반면, getParallelMultiplePostContent는 parallel()을 통해 다중 요청을 병렬로 처리합니다. 이는 큰 데이터나 많은 API 호출을 처리할 때 응답 속도를 개선할 수 있는 중요한 기법입니다.

---

순차적 비동기처리

순차적 비동기 처리란, 비동기 작업을 순차적으로 처리하는 방식으로, 각 작업이 논블로킹 방식으로 처리되면서도 논리적인 순서를 유지하는 것을 의미합니다. 즉, 비동기적으로 동시에 여러 작업을 실행하지만, 처리 순서는 지정된 순서대로 진행됩니다.

이 방식은 작업 자체는 병렬로 실행되지 않지만, 각 작업이 완료될 때까지 블로킹하지 않으며, 작업이 완료된 순서대로 다음 작업을 처리하는 흐름입니다. 이를 통해 서버는 한 작업이 끝날 때까지 대기하지 않고, 다른 작업들을 효율적으로 처리할 수 있습니다.

public Flux<String> sequentialAsyncProcessing(List<Long> ids) {
    return Flux.fromIterable(ids)
            .flatMap(this::asyncCall)
            .log();
}

public Mono<String> asyncCall(Long id) {
    return Mono.just("Processed ID: " + id)
               .delayElement(Duration.ofSeconds(1));  // 비동기적으로 1초 대기
}

위 코드는 비동기적인 작업(asyncCall)을 리스트의 ID에 대해 실행하면서도, 각 작업의 결과는 순차적으로 반환됩니다.

1. 비동기 논블로킹: 각 ID에 대한 작업(asyncCall)은 1초씩 걸리지만, 이 동안 서버는 다른 작업을 처리할 수 있습니다. 블로킹되지 않고 각 요청을 처리하는 동안 대기하지 않습니다.
2. 순차적 처리: 각 ID에 대해 비동기적으로 실행되지만, flatMap을 사용하여 작업이 순차적으로 처리됩니다. 즉, ID 1, 2, 3 순서로 실행됩니다.

비동기 처리의 이점:

• 서버는 동시에 여러 작업을 처리할 수 있으므로 효율성이 극대화됩니다.
• 비동기 논블로킹 방식으로 서버 리소스를 더 잘 활용합니다.

순차적 처리의 이점:

• 처리 순서가 보장되므로, 각 작업이 실행되는 순서를 제어할 수 있습니다.
• 데이터나 작업이 순서대로 처리되어야 할 때 유용합니다.

---

비동기적 처리는 동시성에 강점을 가지고 있고, 순차적 처리는 작업의 순서를 보장해 주는 방식입니다. 순차적 비동기 처리는 이 두 가지의 특성을 결합하여, 서버의 효율성을 높이면서도 순차적인 논리 흐름을 유지하게 도와줍니다.