[Spring Boot] Spring MVC + WebClient 환경에서 JPA 사용하기

굳이 JPA를 사용하려고 하는 이유

사실 가장 편한 방법은 JPA가 아닌 R2DBC와 같은 반응형 드라이버를 제공하는 DB를 선택하는 것이 합리적이다.

 

하지만, 여기서 굳이 JPA를 사용하려는 이유가 있다.

• R2DBC와 Spring MVC는 동작 방식이 다르므로 함께 사용하는 것을 피해야 한다.
  • R2DBC : 비동기/논블로킹 기반 기술
  • Spring MVC : 서블릿 기반의 블로킹 웹 프레임워크

즉, MVC에서 R2DBC의 결과를 바로 받아서 사용하기 어렵고, block을 걸거나 리액티브 체인을 만들어야 한다
block은 전체적인 애플리케이션의 성능을 떨어뜨리기 때문에 지양해야 한다.

• 현재 프로젝트에서 WebClient는 현재 마이크로 서비스 간 통신과 외부 API 호출에만 사용된다.
  • 따라서 통신을 위해 프로젝트 Spring WebFlux 방식으로 뜯어 고치는 것은 피하려고 한다.
• 최선의 방법은 아니지만, WebFlux는 JPA와 같은 블로킹 작업 처리를 위한 방법을 제공한다.

 

그렇기에 이번에는 JPA를 활용하여 개발을 진행하고, 추후 필요하다면 WebFlux로 전환할 예정이다.

---

WebFlux의 동작 방식과 Worker Thread

우선 전체적인 구현에 앞서, WebFlux의 동작 방식에 대해 정말 간단하게 알아본다.

기초적인 Spring Webflux 동작 방식

우선 Spring Webflux는 일반적으로 아래와 같은 방식으로 동작한다.

https://singhkaushal.medium.com/spring-webflux-eventloop-vs-thread-per-request-model-a42d07ee8502

1. 클라이언트마다 SocketChannel이라는 객체를 생성하고, 서버는 이를 통해 클라이언트의 요청을 수신한다.
2. SocketChannel을 통해 받은 요청은 EventLoop에 전달된다.
   • 클라이언트의 SocketChannel들은 하나의 EventLoop를 공유한다.
3. EventLoop가 요청을 감지하면, 인바운드 핸들러 체인을 통해 요청을 처리한다.
4. EventLoop는 그 후, 애플리케이션의 특정 코드를 실행한다.
5. 작업이 완료되면, 아웃바운드 핸들러 체인을 통해 응답을 준비한다.
6. EventLoop는 완료된 작업을 기존에 요청 받은 동일한 SocketChannel을 통해 반환한다.

즉, 여기서 이벤트 루프는 SocketChannel들의 작업을 위임하여 처리하는 역할을 한다. 

여기서 하나의 이벤트 루프 스레드가 전체 요청을 처리하는 것으로 보일 수 있다.

하지만 실제로 Netty는 이벤트 루프 그룹을 통해서 이벤트 루프들을 관리한다.
여기서 이벤트 루프 그룹에 속한 이벤트 루프의 수는 기본적으로 사용 가능한 CPU의 개수만큼 할당된다.

이러한 구조 덕분에 최소한의 컨텍스트 스위칭으로 많은 작업을 수행할 수 있고, 이 때문에 적은 쓰레드와 메모리로 많은 요청을 처리할 수 있다.

 

하지만, 만약 EventLoop가 DB 작업과 같은 블로킹 작업을 수행한다면 다른 작업을 처리할 수 없기에 전체적인 성능 저하로 이어질 수 있다.

 

이를 위해서 WebFlux는 워커 스레드를 통해서 이러한 블로킹 작업을 수행할 수 있도록 지원한다.

Worker Thread를 활용한 Spring WebFlux 동작 방식

아래 그림은 블로킹 작업이 존재할 때, WebFlux의 동작 방식을 나타낸다.

https://singhkaushal.medium.com/spring-webflux-eventloop-vs-thread-per-request-model-a42d07ee8502

1. 클라이언트마다 SocketChannel이라는 객체를 생성하고, 서버는 이를 통해 클라이언트의 요청을 수신한다.
2. SocketChannel을 통해 받은 요청은 EventLoop에 전달된다.
3. EventLoop가 요청을 감지하면, 인바운드 핸들러 체인을 통해 요청을 처리한다.
4. 만약 해당 요청이 블로킹 작업이라면, 이 작업을 Worker Thread에 위임한다.
   1. Worker Thread에서 블로킹 작업을 수행한다.
   2. 작업이 완료되면, 작업 결과에 대한 응답을 큐(ScheduledTaskQueue)에 저장한다.
5. EventLoop는 큐에 응답이 있다면, 이를 폴링한다.
6. EventLoop는 추가로 처리할 작업이 있으면, 이를 완료한 후 아웃바운드 핸들러 체인을 통해 응답을 준비한다.
7. EventLoop는 완료된 작업을 기존에 요청을 받았던 SocketChannel에 다시 반환한다.

따라서 우리는 JPA의 DB 작업을 워커 스레드에서 수행되도록 하여 전체적인 리엑터 체인을 유지한 채 작업이 수행되도록 할 것이다.

---

WebClient Reactor Chain에서 JPA + @Transactional 사용하기

WebClient Reactor Chain에서 JPA : 그냥 잘못된 코드

우선 아래 코드를 확인해보자.

// 잘못된 코드 1 : blocking task on EventLoop + no Transaction
@Transactional
public Mono<UserAuthResponse> processKakaoAuth(String authCode) {
    return kakaoAuthHttpClient.processLogin(authCode)
            .map(userInfo -> {
                User user = userRepository.findById(userInfo.getEmail()).orElse(null);
                if (user == null) user = userRepository.save(userInfo.toEntity());
                String accessToken = jwtUtils.generateAccessToken(user.getUserEmail());
                String refreshToken = jwtUtils.generateRefreshToken(user.getUserEmail());
                if (!tokenService.saveAccessAndRefreshToken(user.getUserEmail(), accessToken, refreshToken)) {
                    throw new RuntimeException("failed to save access and refresh token");
                }
                return new UserAuthResponse(accessToken, refreshToken);
            });
}

 

이 코드는 Reactor Chain 내부에서 JPA Repository를 통해 데이터에 접근하고 있다.

위 코드에서는 이러한 블로킹 작업이 이벤트 루프 스레드에서 수행되기에 전체적인 성능 저하가 있을 수 있다.

 

따라서 우리는 위 코드의 블로킹 작업을 워커 노드로 위임할 것이다.

WebClient Reactor Chain에서 JPA : 개선되었지만, 아직 잘못된 코드

우리는 위 코드를 아래와 같이 수정할 수 있다.

// 잘못된 코드 2 : blocking task on worker thread, but no Transaction
@Transactional
public Mono<UserAuthResponse> processKakaoAuth(String authCode) {
    return kakaoAuthHttpClient.processLogin(authCode)
            .flatMap(userInfo -> 
                Mono.fromCallable(() -> {
                    User user = userRepository.findById(userInfo.getEmail()).orElse(null);
                    if (user == null) user = userRepository.save(userInfo.toEntity());
                    String accessToken = jwtUtils.generateAccessToken(user.getUserEmail());
                    String refreshToken = jwtUtils.generateRefreshToken(user.getUserEmail());
                    if (!tokenService.saveAccessAndRefreshToken(user.getUserEmail(), accessToken, refreshToken)) {
                        throw new RuntimeException("failed to save access and refresh token");
                    }
                    return new UserAuthResponse(accessToken, refreshToken);
                }).subscribeOn(Schedulers.boundedElastic())
            );
}

 

여기서 주의 깊게 확인할 변경 사항은 크게 두가지이다.

• Mono.fromCallable : Callable<T>를 인자로 받아서 실행 결과를 Mono<T>로 반환함
  • try-catch가 내장되어 있음
  • 추가적인 설정을 통해 작업이 워커 쓰레드에서 수행되도록 할 수 있음
• subscribeOn(Schedulers.boundedElastic()) : 작업의 별도의 boundedElastic에서 수행되도록 함
  • 여기서 boundedElastic은 블로킹에 적합하게 설계된 워커 스레드 풀이다.

이러한 이유로 JPA 작업은 별도의 boundedElastic 스레드 풀에서 수행될 것이므로 EventLoop에서 블로킹이 발생하는 것을 막을 수 있다.

 

하지만, 위 코드에서는 트랜잭션이 정상적으로 동작하지는 않는다.

위 코드에서 트랜잭션이 정상적으로 동작하지 않는 이유에 대해서 알아보자.

기본적으로, JPA 트랜잭션은 EntityManager에 의해서 관리되고, EntityManager는 각 스레드마다 독립적으로 존재한다.

즉 현재 트랜잭션에 대한 정보는 현재 스레드의 ThreadLocal에 저장되어 있기에 다른 스레드에서 수행되는 작업은 현재 트랜잭션의 관리 대상이 아니다.

하지만, 위 코드에서 JPA 작업은 현재 스레드가 아닌 boundedElastic 스레드에서 발생하기에 트랜잭션이 정상적으로 동작하지 않는다. 

 

이젠 마지막으로 트랜잭션까지 작동되도록 위 코드를 수정해보자.

WebClient Reactor Chain에서 JPA :  최종 코드

위 코드의 JPA 작업이 동일한 트랜잭션 내에서 관리되도록 하기 위해서는 Mono.fromCallable 내부 코드에서 트랜잭션이 시작되도록 하면 된다.

 

이를 반영한 개선된 코드는 아래와 같다.

// 정답 코드 : blocking task on worker thread, and transaction on jpa
@Transactional
public Mono<UserAuthResponse> processKakaoAuth(String authCode) {
    return kakaoAuthHttpClient.processLogin(authCode)
            .flatMap(userInfo ->
                Mono.fromCallable(() -> transactionTemplate.execute(status -> {
                    User user = userRepository.findById(userInfo.getEmail()).orElse(null);
                    if (user == null) user = userRepository.save(userInfo.toEntity());
                    String accessToken = jwtUtils.generateAccessToken(user.getUserEmail());
                    String refreshToken = jwtUtils.generateRefreshToken(user.getUserEmail());
                    if (!tokenService.saveAccessAndRefreshToken(user.getUserEmail(), accessToken, refreshToken)) {
                        throw new RuntimeException("failed to save access and refresh token");
                    }
                    return new UserAuthResponse(accessToken, refreshToken);
                })).subscribeOn(Schedulers.boundedElastic())
            );
}

 

위 코드의 가장 큰 변화는 TransactionTemplate을 통해 트랜잭션이 Mono.fromCallable 내부 메서드 실행 시점에 시작되도록 한 것이다.

 

이를 통해 JPA 작업이 동일한 트랜잭션 내에서 관리되도록 했고 기대했던 결과가 나타나도록 했다.

또한, WebFlux의 리엑티브 체인을 통해 블로킹 작업을 따로 워커 스레드로 분리할 수 있었다.

사실 꼭 TransactionTemplate을 사용할 필요는 없다.

AOP Self Invocation 문제를 피하면서, JPA 작업에서 트랜잭션이 시작되도록 설정하면 문제 없다.

대표적으로 self injection, 클래스 분리 방식 등을 통해 프록시 객체를 통한 메서드 호출을 이루어내면 된다.

당연히 위 코드의 processKakaoAuth에 적용된 @Transactional은 위 코드의 JPA 작업에 전파되지 않는다.

---

마무리

솔직히 위 코드들은 지금까지 내가 거친 시행착오들이다. 

 

정말 어려웠지만, 덕분에 self invocation과 Webflux의 동작 방식에 대해서 알아볼 수 있었으니 값진 경험이었다고 생각한다.

 

시간이 나면 추가로 변경된 코드와 기존 코드 간 성능 비교도 해볼 예정이다.