Spring Webflux R2dbc - Transaction 적용

Spring Webflux는 리액티브 프로그래밍이므로 MVC에서 썼던 Mybatis나 JPA를 쓰면 블로킹 처리가 되기 때문에 MVC를 쓰는 것과 차이가 없어지므로 R2dbc라는 라이브러리를 사용해야 합니다. Spring에서 제공하고 있기 때문에 Spring-data-r2dbc starter를 적용하면 됩니다. JPA와 유사하게 쓸 수 있고 @Query Annotation도 지원하기 때문에 기존에 JPA를 썼던 분이라면 어렵지 않게 적응 가능하실 겁니다. 그럼 이제 Webflux R2dbc에서 어떻게 Transaction을 적용해야 할 지 알아 봅시다. 

 

 

1. @Transactinal

가장 기초적인 방식으로 Annotation을 Method에 지정하는 방식입니다.

    @Transactional
    public Mono<ResponseEntity<CommonModel.CommonResponse>> insertStudent(StudentModel.StudentRequestModel param) throws Exception {
        return Mono.just(TblStudent.builder()
                    .age(param.getAge())
                    .schoolType(param.getSchoolType())
                    .name(param.getName())
                    .phoneNumber(param.getPhoneNumber())
                    .build()
                )
                .map(studentRepository::save)
                .flatMap(tblStudent -> 
                        studentSubjectRepository.saveAllStudentSubject()
                        .then(Mono.error(new CommonException("INTENTIONAL_ERROR", "롤백을 위한 에러", 400)))
                )
                .then(commonService.created("/students"))
                .onErrorResume(throwable -> Mono.error(new CommonException("ALREADY_EXIST_STUDENT", "이미 존재하는 학생입니다. [" + param.getPhoneNumber() + "]", 400)));
    }

 

그런데, 저렇게 해놓고 테스트를 해보니 Rollback이 실제로 발생하지 않습니다? 뭐지? R2dbc는 분명 Transactional Annotation을 지원한다고 했는데 왜 Rollback이 되지 않지? 그 이유는 @Transactional Annotation은 기본적으로 Runtime Exception에만 작동하기 때문입니다. 지금 저기 선언해 놓은 CommonException은 그냥 Exception을 상속 받은 것이기에 Runtime Exception이 아니므로 Rollback이 발생하지 않습니다. 따라서 아래와 같이 수정이 필요합니다. 

 

    @Transactional(rollbackFor = Exception.class)
    public Mono<ResponseEntity<CommonModel.CommonResponse>> insertStudent(StudentModel.StudentRequestModel param) throws Exception {
        return Mono.just(TblStudent.builder()
                    .age(param.getAge())
                    .schoolType(param.getSchoolType())
                    .name(param.getName())
                    .phoneNumber(param.getPhoneNumber())
                    .build()
                )
                .map(studentRepository::save)
                .flatMap(tblStudent ->
                        studentSubjectRepository.saveAllStudentSubject()
                        .then(Mono.error(new CommonException("INTENTIONAL_ERROR", "롤백을 위한 에러", 400)))
                )
                .then(commonService.created("/students"))
                .onErrorResume(throwable -> Mono.error(new CommonException("ALREADY_EXIST_STUDENT", "이미 존재하는 학생입니다. [" + param.getPhoneNumber() + "]", 400)));
    }

 

이렇게 수정해주면 정상적으로 Rollback 처리 되는 걸 볼 수 있습니다.

 

 

2. TransactionalOperator

두번 째는 TransactionalOperator 사용하는 방식입니다. 

    private final TransactionalOperator operator;
 
    public Mono<ResponseEntity<CommonModel.CommonResponse>> insertStudent(StudentModel.StudentRequestModel param) throws Exception {
        return Mono.just(TblStudent.builder()
                    .age(param.getAge())
                    .schoolType(param.getSchoolType())
                    .name(param.getName())
                    .phoneNumber(param.getPhoneNumber())
                    .build()
                )
                .map(studentRepository::save)
                .flatMap(tblStudent ->
                        studentSubjectRepository.saveAllStudentSubject()
                        .then(Mono.error(new CommonException("INTENTIONAL_ERROR", "롤백을 위한 에러", 400)))
                )
                .then(commonService.created("/students"))
                .onErrorResume(throwable -> Mono.error(new CommonException("ALREADY_EXIST_STUDENT", "이미 존재하는 학생입니다. [" + param.getPhoneNumber() + "]", 400)))
                .as(operator::transactional);
    }

 

 위와 같이 TransactionalOperator를 불러온 뒤에 끝에 as로 transactional을 선언해두면 정상적으로 Rollback이 되는 걸 볼 수 있습니다. 근데 첫 번째 방법도 두 번째 방법도 하나 하나 전부 Method에 선언해 줘야 한다는 게 영 불편합니다. 그래서 원래 MVC할 때 주로 썼던 TransactionInterceptor를 써 봤습니다.

 

 

3. TransactionInterceptor

@Configuration
@EnableR2dbcAuditing
@EnableAspectJAutoProxy
@RequiredArgsConstructor
@Slf4j
public class R2dbcConfig {
 
    private final String AOP_POINTCUT = "within(@org.springframework.stereotype.Service *)";
 
    private final R2dbcTransactionManager transactionManager;
 
    @Bean
    public TransactionInterceptor txAdvice() {
        TransactionInterceptor interceptor = new TransactionInterceptor();
        Properties txProperties = new Properties();
        List<RollbackRuleAttribute> rollbackRules = new ArrayList<>();
        rollbackRules.add(new RollbackRuleAttribute(Exception.class));
 
        RuleBasedTransactionAttribute writeAttribute = new RuleBasedTransactionAttribute(TransactionDefinition.PROPAGATION_REQUIRED, rollbackRules);
        writeAttribute.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);
        writeAttribute.setTimeout(60);
 
        String writeTransactionAttributesDefinition = writeAttribute.toString();
        log.info("Write Attributes :: {}", writeTransactionAttributesDefinition);
 
        txProperties.setProperty("*", writeTransactionAttributesDefinition);
 
        interceptor.setTransactionAttributes(txProperties);
        interceptor.setTransactionManager(transactionManager);
 
        return interceptor;
    }
 
    @Bean
    public Advisor txAdviceAdvisor() {
        AspectJExpressionPointcut pointcut = new AspectJExpressionPointcut();
        pointcut.setExpression(AOP_POINTCUT);
 
        return new DefaultPointcutAdvisor(pointcut, txAdvice());
    }
}

 

 AOP를 이용해 Service로 등록 된 모든 Service를 불러와서 전 Method에 적용하는 방식입니다. 위와 같이 적용 후 테스트 해보니 잘 Rollback되는 걸 확인할 수 있었습니다. 다만 만약에 Service 중에 Mono나 Flux가 아닌 타입으로 Return하는 Method가 있으면 Error가 나는 경우가 있으니 AOP 범위에 대해서는 일부 수정이 필요할 수 있습니다.