Spring R2DBC 리액티브 DB

글쓴이 /

R2DBC란?

R2DBC(Reactive Relational Database Connectivity)는 JDBC의 블로킹 한계를 넘어, 관계형 데이터베이스에 논블로킹 리액티브 접근을 제공하는 SPI(Service Provider Interface)입니다. Spring WebFlux와 결합하면 HTTP 요청부터 DB 쿼리까지 전 구간 논블로킹 파이프라인을 구축할 수 있습니다.

JDBC vs R2DBC 비교

항목 JDBC R2DBC
I/O 모델 블로킹 (스레드 점유) 논블로킹 (이벤트 루프)
반환 타입 ResultSet, List Mono, Flux
트랜잭션 ThreadLocal 기반 Reactor Context 기반
커넥션 풀 HikariCP r2dbc-pool
스레드 효율 요청당 1스레드 소수 스레드로 다수 요청
JPA 호환 완전 지원 미지원 (별도 추상화)

Spring Data R2DBC 설정

의존성 및 설정

// build.gradle.kts
dependencies {
    implementation("org.springframework.boot:spring-boot-starter-data-r2dbc")
    implementation("org.springframework.boot:spring-boot-starter-webflux")
    runtimeOnly("org.postgresql:r2dbc-postgresql")
    // 마이그레이션은 여전히 JDBC 필요
    runtimeOnly("org.postgresql:postgresql")
    implementation("org.flywaydb:flyway-core")
}
# application.yml
spring:
  r2dbc:
    url: r2dbc:postgresql://localhost:5432/mydb
    username: app
    password: ${DB_PASSWORD}
    pool:
      initial-size: 5
      max-size: 20
      max-idle-time: 30m
      max-life-time: 60m
      validation-query: SELECT 1

엔티티와 Repository

Spring Data R2DBC는 JPA와 달리 영속성 컨텍스트가 없습니다. 엔티티는 단순 매핑 객체이며, lazy loading도 지원하지 않습니다.

@Table("orders")
public class Order {
    @Id
    private Long id;
    private String customerId;
    private BigDecimal totalAmount;
    private OrderStatus status;
    
    @CreatedDate
    private LocalDateTime createdAt;
    
    @LastModifiedDate
    private LocalDateTime updatedAt;
    
    // R2DBC는 연관관계 매핑 미지원
    // 별도 쿼리로 조인 처리
}

public interface OrderRepository extends ReactiveCrudRepository<Order, Long> {
    
    Flux<Order> findByCustomerIdOrderByCreatedAtDesc(String customerId);
    
    @Query("SELECT * FROM orders WHERE status = :status AND created_at > :since")
    Flux<Order> findActiveOrdersSince(
        @Param("status") String status, 
        @Param("since") LocalDateTime since
    );
    
    @Modifying
    @Query("UPDATE orders SET status = :status WHERE id = :id")
    Mono<Integer> updateStatus(@Param("id") Long id, @Param("status") String status);
}

R2dbcEntityTemplate — 동적 쿼리

복잡한 동적 쿼리는 R2dbcEntityTemplate의 Criteria API를 활용합니다.

@Service
@RequiredArgsConstructor
public class OrderQueryService {
    private final R2dbcEntityTemplate template;
    
    public Flux<Order> search(OrderSearchCriteria criteria) {
        var query = Query.empty();
        var where = Criteria.empty();
        
        if (criteria.customerId() != null) {
            where = where.and("customer_id").is(criteria.customerId());
        }
        if (criteria.minAmount() != null) {
            where = where.and("total_amount").greaterThanOrEquals(criteria.minAmount());
        }
        if (criteria.status() != null) {
            where = where.and("status").is(criteria.status().name());
        }
        if (criteria.fromDate() != null) {
            where = where.and("created_at").greaterThanOrEquals(criteria.fromDate());
        }
        
        query = Query.query(where)
            .sort(Sort.by(Sort.Direction.DESC, "created_at"))
            .limit(criteria.limit())
            .offset(criteria.offset());
        
        return template.select(Order.class)
            .matching(query)
            .all();
    }
}

트랜잭션 관리

R2DBC 트랜잭션은 Reactor Context 기반으로 동작합니다. JDBC의 ThreadLocal 방식과 근본적으로 다릅니다.

@Service
@RequiredArgsConstructor
public class OrderService {
    private final OrderRepository orderRepo;
    private final OrderItemRepository itemRepo;
    private final TransactionalOperator txOperator;
    
    // 방법 1: @Transactional (선언적)
    @Transactional
    public Mono<Order> createOrder(CreateOrderCommand cmd) {
        return orderRepo.save(Order.from(cmd))
            .flatMap(order -> 
                Flux.fromIterable(cmd.items())
                    .map(item -> OrderItem.of(order.getId(), item))
                    .collectList()
                    .flatMapMany(itemRepo::saveAll)
                    .then(Mono.just(order))
            );
    }
    
    // 방법 2: TransactionalOperator (프로그래매틱)
    public Mono<Order> cancelOrder(Long orderId) {
        return orderRepo.findById(orderId)
            .switchIfEmpty(Mono.error(new OrderNotFoundException(orderId)))
            .flatMap(order -> {
                order.cancel();
                return orderRepo.save(order)
                    .then(itemRepo.deleteByOrderId(orderId))
                    .thenReturn(order);
            })
            .as(txOperator::transactional);
    }
}

연관관계 수동 조합 패턴

R2DBC는 JPA의 @OneToMany, @ManyToOne을 지원하지 않습니다. 연관 데이터는 직접 조합해야 합니다.

// DTO 기반 조합
public record OrderDetailDto(
    Order order,
    List<OrderItem> items,
    Customer customer
) {}

@Service
@RequiredArgsConstructor
public class OrderDetailService {
    private final OrderRepository orderRepo;
    private final OrderItemRepository itemRepo;
    private final CustomerRepository customerRepo;
    
    public Mono<OrderDetailDto> getOrderDetail(Long orderId) {
        Mono<Order> orderMono = orderRepo.findById(orderId)
            .switchIfEmpty(Mono.error(new OrderNotFoundException(orderId)))
            .cache(); // 다중 구독 방지
        
        Mono<List<OrderItem>> itemsMono = orderMono
            .flatMapMany(o -> itemRepo.findByOrderId(o.getId()))
            .collectList();
        
        Mono<Customer> customerMono = orderMono
            .flatMap(o -> customerRepo.findById(o.getCustomerId()));
        
        return Mono.zip(orderMono, itemsMono, customerMono)
            .map(tuple -> new OrderDetailDto(
                tuple.getT1(), tuple.getT2(), tuple.getT3()
            ));
    }
    
    // DatabaseClient로 JOIN 쿼리 직접 작성
    private final DatabaseClient databaseClient;
    
    public Flux<OrderSummaryDto> getOrderSummaries(String customerId) {
        return databaseClient.sql("""
            SELECT o.id, o.total_amount, o.status, o.created_at,
                   COUNT(i.id) as item_count,
                   c.name as customer_name
            FROM orders o
            JOIN order_items i ON i.order_id = o.id
            JOIN customers c ON c.id = o.customer_id
            WHERE o.customer_id = :customerId
            GROUP BY o.id, c.name
            ORDER BY o.created_at DESC
            """)
            .bind("customerId", customerId)
            .map((row, meta) -> new OrderSummaryDto(
                row.get("id", Long.class),
                row.get("total_amount", BigDecimal.class),
                row.get("status", String.class),
                row.get("item_count", Long.class),
                row.get("customer_name", String.class)
            ))
            .all();
    }
}

커넥션 풀 튜닝과 모니터링

@Configuration
public class R2dbcConfig extends AbstractR2dbcConfiguration {
    
    @Bean
    @Override
    public ConnectionFactory connectionFactory() {
        return ConnectionFactories.get(ConnectionFactoryOptions.builder()
            .option(DRIVER, "pool")
            .option(PROTOCOL, "postgresql")
            .option(HOST, "localhost")
            .option(PORT, 5432)
            .option(DATABASE, "mydb")
            .option(USER, "app")
            .option(PASSWORD, "secret")
            // 풀 옵션
            .option(MAX_SIZE, 20)
            .option(INITIAL_SIZE, 5)
            .option(MAX_IDLE_TIME, Duration.ofMinutes(30))
            .option(MAX_ACQUIRE_TIME, Duration.ofSeconds(3)) // 핵심!
            .option(VALIDATION_QUERY, "SELECT 1")
            .build());
    }
}

// Micrometer 메트릭 연동
@Bean
ConnectionPoolMetricsCollector metricsCollector(MeterRegistry registry) {
    return new MicrometerR2dbcPoolMetricsCollector(registry, Tags.empty());
}

MAX_ACQUIRE_TIME은 커넥션 풀 고갈 시 무한 대기를 방지하는 핵심 설정입니다. 3초 내 커넥션을 못 얻으면 R2dbcTimeoutException을 던져 빠른 실패를 유도합니다.

Virtual Thread와의 비교

Spring 6.1에서 Virtual Thread가 도입되면서 “R2DBC가 필요한가?”라는 질문이 자주 나옵니다.

  • Virtual Thread + JDBC: 기존 코드 변경 없이 논블로킹 효과. 생태계 풍부 (JPA, QueryDSL)
  • R2DBC: 배압(backpressure) 네이티브 지원, 스트리밍 쿼리에 강점, WebFlux 파이프라인과 자연스러운 통합
  • 결론: 신규 프로젝트에서 WebFlux를 쓴다면 R2DBC, MVC 기반이라면 Virtual Thread + JDBC가 현실적

운영 팁

  • 마이그레이션: Flyway/Liquibase는 여전히 JDBC 필요 → spring.flyway.url에 JDBC URL 별도 설정
  • N+1 방지: JPA N+1 해결 전략과 달리 R2DBC는 자체 lazy loading이 없어 N+1이 발생하지 않지만, 수동 조합 시 반복 쿼리에 주의
  • 에러 핸들링: onErrorResume, onErrorMap으로 리액티브 체인 내에서 예외 변환
  • 테스트: @DataR2dbcTest + Testcontainers로 슬라이스 테스트 구성

정리

Spring Data R2DBC는 WebFlux 기반 애플리케이션에서 DB 구간까지 완전한 논블로킹 파이프라인을 완성하는 핵심 기술입니다. JPA의 편의성은 포기하지만, 배압 제어와 높은 동시성 처리라는 명확한 이점을 얻습니다. Virtual Thread의 등장으로 선택지가 넓어졌지만, 스트리밍 데이터 처리나 WebFlux 생태계에서는 여전히 R2DBC가 최적의 선택입니다.

위로 스크롤
WordPress Appliance - Powered by TurnKey Linux