spring-r2dbc

Spring Data R2DBC

快速入门

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-r2dbc</artifactId>
    </dependency>
    <dependency>
        <groupId>org.postgresql</groupId>
        <artifactId>r2dbc-postgresql</artifactId>
        <scope>runtime</scope>
    </dependency>
</dependencies>

# application.yml
spring:
  r2dbc:
    url: r2dbc:postgresql://localhost:5432/mydb
    username: ${DB_USERNAME}
    password: ${DB_PASSWORD}
    pool:
      enabled: true
      initial-size: 5
      max-size: 20

---

实体定义

import org.springframework.data.annotation.Id;
import org.springframework.data.annotation.CreatedDate;
import org.springframework.data.annotation.LastModifiedDate;
import org.springframework.data.relational.core.mapping.Table;
import org.springframework.data.relational.core.mapping.Column;

@Table("products")
public class Product {

    @Id
    private Long id;

    @Column("product_name")
    private String name;

    private String description;
    private BigDecimal price;

    @Column("category_id")
    private Long categoryId;

    @CreatedDate
    private Instant createdAt;

    @LastModifiedDate
    private Instant updatedAt;

    private boolean active;

    // Constructors, getters, setters...
}

// For immutable entities with records (Java 17+)
@Table("orders")
public record Order(
    @Id Long id,
    @Column("customer_id") Long customerId,
    BigDecimal total,
    OrderStatus status,
    @CreatedDate Instant createdAt
) {
    public Order withStatus(OrderStatus newStatus) {
        return new Order(id, customerId, total, newStatus, createdAt);
    }
}

---

仓库接口

import org.springframework.data.r2dbc.repository.R2dbcRepository;
import org.springframework.data.r2dbc.repository.Query;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public interface ProductRepository extends R2dbcRepository<Product, Long> {

    // Automatic derived queries
    Flux<Product> findByActiveTrue();
    Flux<Product> findByNameContainingIgnoreCase(String name);
    Flux<Product> findByCategoryId(Long categoryId);
    Flux<Product> findByPriceBetween(BigDecimal min, BigDecimal max);
    Mono<Product> findByNameIgnoreCase(String name);

    // Ordering and limiting
    Flux<Product> findTop10ByActiveTrueOrderByCreatedAtDesc();

    // Count and Exists
    Mono<Long> countByActiveTrue();
    Mono<Boolean> existsByName(String name);

    // Custom query
    @Query("SELECT * FROM products WHERE category_id = :categoryId AND price < :maxPrice")
    Flux<Product> findByCategoryWithMaxPrice(Long categoryId, BigDecimal maxPrice);

    @Query("UPDATE products SET active = false WHERE id = :id")
    @Modifying
    Mono<Integer> deactivateProduct(Long id);

    // Projection with DTO
    @Query("SELECT id, product_name as name, price FROM products WHERE active = true")
    Flux<ProductSummary> findAllSummaries();
}

public record ProductSummary(Long id, String name, BigDecimal price) {}

完整参考：复杂查询请参阅 database-client.md，了解 DatabaseClient 的用法。

---

服务层

@Service
@RequiredArgsConstructor
@Slf4j
public class ProductService {

    private final ProductRepository productRepository;

    // Create
    public Mono<Product> createProduct(CreateProductRequest request) {
        Product product = Product.create(request.name(), request.description(), request.price());
        product.setCategoryId(request.categoryId());
        return productRepository.save(product)
            .doOnSuccess(p -> log.info("Created product: {}", p.getId()));
    }

    // Read
    public Mono<Product> getProduct(Long id) {
        return productRepository.findById(id)
            .switchIfEmpty(Mono.error(new ProductNotFoundException(id)));
    }

    public Flux<Product> getAllActiveProducts() {
        return productRepository.findByActiveTrue();
    }

    // Update
    public Mono<Product> updateProduct(Long id, UpdateProductRequest request) {
        return productRepository.findById(id)
            .switchIfEmpty(Mono.error(new ProductNotFoundException(id)))
            .map(product -> {
                if (request.name() != null) product.setName(request.name());
                if (request.price() != null) product.setPrice(request.price());
                return product;
            })
            .flatMap(productRepository::save);
    }

    // Delete (soft delete)
    public Mono<Void> deactivateProduct(Long id) {
        return productRepository.deactivateProduct(id)
            .filter(count -> count > 0)
            .switchIfEmpty(Mono.error(new ProductNotFoundException(id)))
            .then();
    }
}

---

响应式事务

@Service
@RequiredArgsConstructor
public class OrderService {

    private final OrderRepository orderRepository;
    private final OrderItemRepository orderItemRepository;

    @Transactional
    public Mono<Order> createOrder(CreateOrderRequest request) {
        return validateProducts(request.items())
            .then(calculateTotal(request.items()))
            .flatMap(total -> {
                Order order = new Order(null, request.customerId(), total, OrderStatus.PENDING, null);
                return orderRepository.save(order);
            })
            .flatMap(order -> saveOrderItems(order.id(), request.items())
                .then(Mono.just(order)));
    }

    private Mono<BigDecimal> calculateTotal(List<OrderItemRequest> items) {
        return Flux.fromIterable(items)
            .flatMap(item -> productRepository.findById(item.productId())
                .map(p -> p.getPrice().multiply(BigDecimal.valueOf(item.quantity()))))
            .reduce(BigDecimal.ZERO, BigDecimal::add);
    }
}

完整参考：请参阅 transactions.md，了解 TransactionalOperator 和关系处理。

---

最佳实践

应做	不应做
在 WebFlux 应用中使用 R2DBC	混用 JDBC 和 R2DBC
配置连接池	不使用连接池
使用批量查询手动处理关系	期待类似 JPA 的延迟加载
使用 @Transactional 或 TransactionalOperator	忘记事务管理
使用 StepVerifier 进行测试	在生产环境中使用 .block()

---

不应使用此技能的情况

• 阻塞式应用 - 使用 spring-data-jdbc 或 spring-data-jpa
• 复杂 ORM 功能 - R2DBC 功能简单；需要延迟加载时请使用 JPA
• 未使用 WebFlux - R2DBC 面向响应式技术栈
• MongoDB 响应式应用 - 使用 spring-data-mongodb 响应式仓库

---

常见问题

错误	原因	解决方案
NoSuchBeanDefinitionException: ConnectionFactory	缺少 R2DBC 驱动	添加 r2dbc-postgresql/mysql 依赖
Connection timeout	连接池耗尽	增大 max-size，检查连接泄漏
TransactionRequiredException	缺少 @Transactional	添加注解或使用 TransactionalOperator
实体未映射	缺少注解	检查 @Table、@Id、@Column
N+1 查询	加载关系	使用带 IN 子句的批量查询

---

反模式

反模式	问题	解决方案
在生产环境中使用 .block()	阻塞事件循环	使用响应式运算符
关系产生 N+1 查询	性能问题	使用带 IN 子句的批量查询
缺少连接池	连接耗尽	配置 r2dbc-pool
事务过大	连接占用时间过长	缩短事务
不处理错误	故障被静默忽略	使用 onErrorResume、onErrorMap

---

快速故障排除

问题	诊断	修复
连接超时	检查连接池设置	增大 max-size，检查连接泄漏
实体未映射	检查注解	添加 @Table、@Id、@Column
事务未生效	检查 @Transactional	如有需要，使用 TransactionalOperator
查询结果为空	检查列名	确认映射与数据库一致
连接池耗尽	监控活动连接	增大连接池，修复连接泄漏

---

参考文件

文件	内容
database-client.md	DatabaseClient、动态查询、聚合
transactions.md	事务、关系、分页
advanced.md	审计、转换器、连接池配置、测试

---

外部文档

• Spring Data R2DBC
• R2DBC Specification