[R2DBC] "트랜잭션은 어떻게 걸죠?" Reactive Transactional과 R2dbcEntityTemplate

 

WebFlux에서 R2DBC를 쓸 때 가장 불안한 점은 트랜잭션입니다.
"A 계좌에서 돈을 빼고(update), B 계좌에 돈을 넣어야(update) 하는데, 중간에 에러 나면 롤백이 될까?"

결론부터 말씀드리면, @Transactional 어노테이션 하나면 완벽하게 동작합니다.
하지만 그 원리는 MVC와 완전히 다릅니다.

---

1. 트랜잭션 매니저 설정 ()

스프링 부트가 ConnectionFactory를 보고 알아서 빈을 등록해 주지만, 명시적으로 알고 있어야 합니다.
JDBC의 PlatformTransactionManager가 아니라, R2dbcTransactionManager가 필요합니다.

@Configuration
@EnableTransactionManagement // 트랜잭션 관리 활성화
public class R2dbcConfig {

    @Bean
    public ReactiveTransactionManager transactionManager(ConnectionFactory connectionFactory) {
        return new R2dbcTransactionManager(connectionFactory);
    }
}

---

2. 선언적 트랜잭션 ()

사용법은 MVC와 100% 똑같습니다.

@Service
@RequiredArgsConstructor
public class MoneyTransferService {

    private final AccountRepository accountRepository;

    @Transactional // ★ 예외 발생 시 알아서 Rollback 됨!
    public Mono<Void> transfer(Long fromId, Long toId, Long amount) {
        return accountRepository.findById(fromId)
                .flatMap(from -> {
                    from.withdraw(amount);
                    return accountRepository.save(from);
                })
                .flatMap(savedFrom -> accountRepository.findById(toId))
                .flatMap(to -> {
                    to.deposit(amount);
                    return accountRepository.save(to);
                })
                .then(); // Mono<Void> 리턴
    }
}

동작 원리 (Reactor Context):
트랜잭션 정보가 쓰레드가 아니라, 데이터 흐름(Pipeline)의 컨텍스트(Context)를 타고 흘러갑니다. 그래서 중간에 쓰레드가 바뀌어도 트랜잭션은 유지됩니다.

---

3. JPA가 그리울 땐:  (DatabaseClient)

ReactiveCrudRepository는 너무 단순합니다. JOIN도 어렵고 동적 쿼리도 안 되죠.
이때 JPA의 EntityManager나 MyBatis처럼 쓸 수 있는 도구가 바로 R2dbcEntityTemplate입니다.
(스프링 부트 최신 버전에서는 DatabaseClient를 더 권장합니다.)

① 복잡한 조건 검색 (Fluent API)

@Repository
@RequiredArgsConstructor
public class UserCustomRepository {

    private final R2dbcEntityTemplate template;

    public Flux<User> findUsersByAgeAndName(int age, String name) {
        // SQL 몰라도 메서드 체이닝으로 쿼리 생성!
        return template.select(User.class)
                .from("users")
                .matching(Query.query(Criteria.where("age").is(age)
                        .and("name").is(name)))
                .all();
    }
}

② 직접 SQL 작성 (Native Query)

복잡한 조인이나 통계 쿼리는 그냥 SQL로 짜는 게 편합니다.

@Repository
@RequiredArgsConstructor
public class NativeQueryRepository {

    private final DatabaseClient databaseClient; // 더 로우 레벨 API

    public Flux<UserStatDto> getUserStats() {
        return databaseClient.sql("SELECT count(*) as count, role FROM users GROUP BY role")
                .map((row, metadata) -> new UserStatDto(
                        row.get("role", String.class),
                        row.get("count", Long.class)
                ))
                .all();
    }
}

---

4. 프로그래밍 방식 트랜잭션 ()

@Transactional을 못 붙이는 상황이거나, 로직 중간에 세밀하게 트랜잭션을 제어하고 싶다면 TransactionalOperator를 씁니다.

@Service
@RequiredArgsConstructor
public class ManualTransactionService {

    private final TransactionalOperator rxtx; // 스프링이 주입해줌
    private final UserRepository userRepository;

    public Mono<User> saveWithRollback(User user) {
        return userRepository.save(user)
                .flatMap(saved -> {
                    if (saved.getName().equals("BAD_USER")) {
                        return Mono.error(new RuntimeException("롤백시켜!"));
                    }
                    return Mono.just(saved);
                })
                // ★ 이 체인 전체를 트랜잭션으로 감싸라!
                .as(rxtx::transactional); 
    }
}

.as(rxtx::transactional) 한 줄이면 끝입니다. 정말 우아하지 않나요?

---

마치며

오늘의 결론입니다.

1. WebFlux에서도 @Transactional은 완벽하게 동작한다. (쓰레드 로컬 대신 Reactor Context 사용)
2. 복잡한 쿼리는 R2dbcEntityTemplate이나 DatabaseClient를 쓰면 된다.
3. 세밀한 제어는 TransactionalOperator를 활용하자.

이로써 WebFlux로 API부터 DB 트랜잭션까지, 완전한 Non-Blocking 시스템을 구축할 수 있게 되었습니다.

그런데, 비동기 코드는 테스트하기가 정말 까다롭습니다.
"데이터가 올 때까지 기다렸다가 검증해야 하나? Thread.sleep을 써야 하나?"

다음 포스팅에서는 "비동기 코드는 어떻게 테스트해요?" block() 없이 리액티브 스트림을 완벽하게 검증하는 StepVerifier에 대해 알아보겠습니다.