WebFlux 전환하기

어플리케이션의 성능을 높이기 위해 기존에 MVC를 사용하던 대기열 서비스를 WebFlux로 전환하게 되었다. 먼저 WebFlux는 MVC와 어떤 차이가 있는지 알아보자.

1. MVC VS WebFlux

1.1. Spring MVC

MVC는 기본적으로 Thread Per Request 방식으로 동작하며 Servlet Container에서 요청을 Thread에 할당하여 Blocking 방식으로 호출하게 된다.

 

 

만약 MVC로 구성된 서버에 동시에 많은 요청이 들어오게 된다면 사용 가능한 Thread 수만큼 요청이 처리되며 Blocking IO 작업을 할 때 해당 Thread는 아무것도 하지 않고 대기하게 되어 병목 현상이 발생할 수 있다.

 

스레드 풀의 크기를 늘려 사용할 수 있는 스레드 수를 증가시킬 수 있지만, CPU 코어 수에 비해 너무 많은 스레드를 구성하면 Context Switching에 따른 오버헤드가 발생하여 성능 저하를 초래할 수 있다.

 

 

이 문제를 해결하기 위해서 비동기 처리가 필요한데..

 

MVC에서도 Callable, DeferredResult, @Async 등의 방식을 통해 비동기 처리를 지원한다. 그러나 이러한 방식은 기본적으로 서블릿 스레드 풀에 의존하며, 여전히 스레드 블로킹이 발생할 수 있다.

 

1.2. Spring WebFlux

 

그렇다면 MVC와 비교하여 WebFlux는 무엇이 다를까?

 

Spring WebFlux는 비동기 및 논블로킹 프로그래밍 모델을 제공하는 웹 프레임워크이다. 반응형 프로그래밍을 지원하며, 대규모 트래픽을 처리할 수 있도록 설계되었다.

 

WebFlux는 내부적으로 이벤트 루프 모델을 통해 blocking 상황을 최소화하는데, 이를 통해 하나의 스레드가 다수의 요청을 비동기적으로 처리하게 된다. 즉, Webflux는 기본적으로 적은 개수의 스레드들을 통해 비동기 처리를 하게 된다. 이에 따라 Context Switching 비용이 줄어들어 좋은 성능을 가질 수 있다.

 

 

다음으로 이러한 비동기 처리가 가능하게 하는 핵심인 Reactive 프로그래밍에 대해 알아보자.

 

---

 

2. Reactive 프로그래밍

Reactive 프로그래밍은 비동기 데이터 흐름과 변화에 대한 프로그래밍 패러다임으로, 주로 이벤트 기반 시스템에서 데이터의 흐름을 처리하는 데 유용하다.

 

2.1. 메서드 체이닝

Reactive 프로그래밍의 큰 특징 중 하나는 메서드 체이닝을 통해 데이터를 필터링하거나 변환하는 작업을 연속적으로 수행하는 것이다. 해당 과정은 비동기, 논블로킹 방식으로 진행되며, 데이터가 준비되는 즉시 다음 단계로 넘어가게 된다.

 

userFlux
    .filter(user -> user.isActive())
    .map(user -> user.getName())
    .subscribe(name -> System.out.println("Active User: " + name));

 

2.2. Reactive Streams

WebFlux의 비동기 스트림 처리에 대한 표준 API Reactive Streams는 데이터의 흐름을 비동기적으로 처리하는 데 필요한 구성 요소를 정의한다. 해당 API는 다음과 같은 주요 인터페이스로 구성된다.

 

Publisher: 데이터를 발행하는 주체로, 데이터를 스트림으로 전송한다.
Subscriber: 데이터를 소비하는 주체로, Publisher로부터 데이터를 수신한다.
Subscription: Publisher와 Subscriber 간의 연결을 나타내며, Subscriber가 데이터를 요청할 수 있는 메커니즘을 제공한다.
Processor: Publisher와 Subscriber를 결합하여 데이터를 변환하거나 필터링하는 역할을 한다.

 

 

---

 

3. WebFlux 전환하기

종속성 추가

먼저 WebFlux를 사용하기 위해서 기존에 사용하던 spring-boot-starter-web 설정을 지우고 spring-boot-starter-webflux를 추가해줘야 한다.

 

implementation 'org.springframework.boot:spring-boot-starter-webflux'

 

또한, Reactive 프로그래밍을 구성하기 위한 R2DBC(Reactive Relational Database Connecivity)를 지원하는 라이브러리를 추가해야한다.

 

나의 경우.. 기존에 사용하던 Redisson 클라이언트에서 다음과 같이 설정하면 Redis를 Reacticve하게 사용할 수 있다.

 

RedissonConfig

@Configuration
public class RedissonConfig {

    @Value("${spring.redis.host}")
    private String redisHost;

    @Value("${spring.redis.port}")
    private String redisPort;

    @Bean
    public RedissonClient redissonClient() {
        Config config = new Config();
        config.useSingleServer()
                .setAddress(String.format("redis://%s:%s", redisHost, redisPort));
        return Redisson.create(config);
    }

    @Bean
    public RedissonReactiveClient redissonReactiveClient(RedissonClient redissonClient) {
        return redissonClient.reactive();
    }

}

 

RedisRepository

@Repository
@RequiredArgsConstructor
public class RedisRepository {

    private final RedissonReactiveClient redissonClient;

    public RBucketReactive<String> getBucket(String key) {
        return redissonClient.getBucket(key, StringCodec.INSTANCE);
    }

    public RAtomicLongReactive getCounter(String key) {
        return redissonClient.getAtomicLong(key);
    }

}

 

코드 수정하기

다음과 같이 사용하는 메서드들을 하나의 체인 형태로 연결지으면 된다. 꼭 중간에 blocking되는 부분이 없게 해야 한다...
메서드들을 한 흐름으로 처리하기 때문에 디버깅이 어려운데, 각 단계에서의 상태를 파악하기 위해 로깅을 하는 것이 중요하다.

 

WaitingQueueService

@Slf4j
@Service
@RequiredArgsConstructor
public class WaitingQueueService {

    private final WaitingQueueRepository waitingQueueRepository;
    private final WorkingQueueRepository workingQueueRepository;

    public Mono<GeneralQueueTokenResponse> enterWaitingQueue(String userId, String performanceId) {
        val command = InsertWaitingQueueTokenCommand.create(userId, performanceId);

        return waitingQueueRepository.insertWaitingQueueToken(command)
                .doOnSuccess(token -> log.info("대기열 진입 완료 {}", token))
                .map(GeneralQueueTokenResponse::from);
    }

    public Mono<GeneralQueueTokenResponse> getQueueInfo(String userId, String performanceId) {
        val command = FindWaitingQueueTokenCommand.create(userId, performanceId);

        return waitingQueueRepository.findWaitingQueueToken(command)
                .doOnSuccess(token -> log.info("대기열 토큰 조회 완료 {}", token))
                .map(GeneralQueueTokenResponse::from)
                .switchIfEmpty(findWorkingQueueToken(userId, performanceId));
    }

    private Mono<GeneralQueueTokenResponse> findWorkingQueueToken(String userId, String performanceId) {
        val command = FindWorkingQueueTokenCommand.create(userId, performanceId);

        return workingQueueRepository.findWorkingQueueToken(command)
                .doOnSuccess(token -> log.info("작업열 토큰 조회 완료 {}", token))
                .map(GeneralQueueTokenResponse::from)
                .switchIfEmpty(Mono.error(new ApplicationException(USER_TOKEN_NOT_FOUND)));
    }

}

 

WaitingQueueRepository

@Repository
@RequiredArgsConstructor
public class WaitingQueueRepositoryImpl implements WaitingQueueRepository {

    private final SaveTokenScript saveTokenScript;
    private final GetRankAndSizeScript getRankAndSizeScript;
    private final DeleteFirstTokenScript deleteFirstTokenScript;

    @Override
    public Mono<QueueToken> insertWaitingQueueToken(InsertWaitingQueueTokenCommand command) {
        return saveTokenScript.saveToken(command)
                .flatMap(tokenStatus -> createToken(
                        tokenStatus,
                        command.getUserId(),
                        command.getPerformanceId()
                ));
    }

    private Mono<QueueToken> createToken(TokenStatus tokenStatus, String userId, String performanceId) {
        return tokenStatus == TokenStatus.WORKING
                ? Mono.just(WorkingQueueToken.create(userId, performanceId))
                : Mono.just(WaitingQueueToken.create(userId, performanceId));
    }

    @Override
    public Mono<WaitingQueueToken> findWaitingQueueToken(FindWaitingQueueTokenCommand command) {
        return getRankAndSizeScript.getRankAndSize(command.getQueueName(), command.getTokenValue())
                .map(tuple -> WaitingQueueToken.withPosition(
                        command.getUserId(),
                        command.getPerformanceId(),
                        command.getTokenValue(),
                        tuple.getT1() + 1,
                        tuple.getT2()
                ));
    }

    @Override
    public Mono<WaitingQueueToken> deleteFirstWaitingQueueToken(DeleteFirstWaitingQueueTokenCommand command) {
        return deleteFirstTokenScript.deleteFirstToken(command)
                .map(tokenValue -> WaitingQueueToken.valueOf(command.getPerformanceId(), tokenValue));
    }

}