java

7.3 响应式 Repository

Spring Data R2DBC 响应式 Repository 模式：ReactiveCrudRepository 与 ReactiveSortingRepository、派生查询方法 naming convention、@Query 注解自定义 SQL、DatabaseClient 底层操作，以及与 JPA 的对比。

Leeting Yan 2026-11-01 3 分钟阅读 1281 字

ReactiveCrudRepository

Spring Data R2DBC 的核心是 ReactiveCrudRepository：

import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Flux;

public interface UserRepository extends ReactiveCrudRepository<User, Long> {
    Mono<User> findByName(String name);
    Flux<User> findByStatus(String status);
    Mono<Boolean> existsByName(String name);
}

Like 的 % 和 _ 不会被自动补上，findBySymbolLike("BTC") 等价于 symbol = 'BTC'，想模糊匹配要传 "%BTC%"。这是 Like 与 Containing、StartingWith 的关键区别：后者由 Spring Data 负责加通配符。

Spring Data 会在运行时自动生成实现。方法命名遵循约定：

前缀	作用	示例
`find...By`	查询	`findByName` → `WHERE name = ?`
`find...By...And...`	多条件	`findByStatusAndCreatedAtAfter`
`count...By`	计数	`countByStatus`
`exists...By`	存在判断	`existsByName`
`delete...By`	删除	`deleteByStatus`

派生查询的关键字

public interface OrderRepository extends ReactiveCrudRepository<Order, Long> {
    // 等于
    Flux<Order> findBySymbol(String symbol);

    // 大于 / 小于
    Flux<Order> findByPriceGreaterThan(double price);
    Flux<Order> findByQuantityLessThanEqual(int qty);

    // Like：通配符要调用方自己带，这里传 "%BTC%"
    Flux<Order> findBySymbolLike(String pattern);

    // 区间
    Flux<Order> findByCreatedAtBetween(Instant start, Instant end);

    // In
    Flux<Order> findByStatusIn(List<String> statuses);

    // Ignore case
    Flux<Order> findBySymbolIgnoreCase(String symbol);

    // 排序
    Flux<Order> findByStatusOrderByCreatedAtDesc(String status);

    // 分页
    Flux<Order> findByStatus(String status, Pageable pageable);
    Mono<Long> countByStatus(String status);
}

@Query 自定义 SQL

当派生查询无法满足需求时，用 @Query：

public interface OrderRepository extends ReactiveCrudRepository<Order, Long> {

    @Query("""
        SELECT o.*, u.name as user_name
        FROM orders o
        JOIN users u ON o.user_id = u.id
        WHERE o.symbol = :symbol
        ORDER BY o.created_at DESC
        LIMIT :limit
        """)
    Flux<OrderWithUser> findOrdersWithUserBySymbol(
        @Param("symbol") String symbol,
        @Param("limit") int limit
    );

    @Query("SELECT COUNT(*) FROM orders WHERE created_at > :since")
    Mono<Long> countRecentOrders(@Param("since") Instant since);
}

:symbol 是命名参数（需要 @Param 注解或编译时 -parameters 标志）。@Query 的返回值可以是 Mono<T>、Flux<T> 或 Mono<Void>。

DatabaseClient：底层操作

DatabaseClient 是 Spring Data R2DBC 的低级 API，介于原始 R2DBC SPI 和 Repository 之间：

@Service
public class ReportService {
    private final DatabaseClient db;

    public ReportService(DatabaseClient db) {
        this.db = db;
    }

    public Mono<Map<String, Object>> getStats() {
        return db.sql("""
            SELECT
                COUNT(*) as total_orders,
                SUM(price * quantity) as total_value,
                AVG(price) as avg_price
            FROM orders
            WHERE status = :status
            """)
            .bind("status", "COMPLETED")
            .fetch()
            .one();  // 返回单行 Map
    }

    public Flux<Map<String, Object>> getDailyReport() {
        return db.sql("""
            SELECT DATE(created_at) as day, COUNT(*) as cnt
            FROM orders
            GROUP BY DATE(created_at)
            ORDER BY day DESC
            """)
            .fetch()
            .all();  // 返回多行 Flux
    }
}

DatabaseClient 适合报表、聚合、复杂 JOIN 等不适合 Repository 派生查询的场景。

实体映射

Spring Data R2DBC 需要显式表映射（不像 JPA 自动推断）：

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

@Table("users")
public record User(
    @Id Long id,
    String name,
    String email,
    String status
) {}

注解	作用
`@Table`	指定表名
`@Id`	主键
`@Column`	自定义列名（字段名不同时）
`@Transient`	不持久化
`@CreatedDate` / `@LastModifiedDate`	自动填充时间戳

@Table("orders")
public record Order(
    @Id Long id,
    @Column("user_id") Long userId,
    String symbol,
    int quantity,
    double price,
    @CreatedDate Instant createdAt
) {}

启用审计：

@Configuration
@EnableR2dbcAuditing
public class R2dbcConfig {
}

分页与排序

import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Sort;

// 分页：页码 0-based，index=1 是人类习惯的第 2 页；每页 20 条，按 createdAt 降序
PageRequest page = PageRequest.of(1, 20, Sort.by("createdAt").descending());

Flux<Order> pageContent = orderRepo.findByStatus("PENDING", page);
Mono<Long> total = orderRepo.countByStatus("PENDING");

注意：R2DBC Repository 的 Page<T> 返回不支持（没有 findAll(Pageable)），需要手动组合 Flux<T> + Mono<Long>。

与 JPA 的对比

能力	JPA	R2DBC
延迟加载（Lazy）	✅ `@ManyToOne(fetch = LAZY)`	❌ 无延迟加载
级联操作	✅ `CascadeType.ALL`	❌ 手动处理
一级缓存/脏检查	✅ 持久化上下文内自动	❌ 无状态，每次查询都走数据库
二级缓存	✅ JPA 二级缓存	❌ 无内置缓存，可用 Spring Cache（第 9 章）
自动 DDL	✅ `hibernate.ddl-auto`	⚠️ 需 Flyway/Liquibase
响应式	❌ 阻塞	✅ Mono/Flux
复杂查询	JPQL / Criteria	原生 SQL / `@Query`

R2DBC 是「更简单但更底层」的选择。如果你需要响应式，放弃 JPA 的自动魔法是必要代价。

eventd v2 的数据层

eventd-data 的 Repository 结构：

// UserRepository.java
public interface UserRepository extends ReactiveCrudRepository<User, Long> {
    Mono<User> findByName(String name);
}

// OrderRepository.java
public interface OrderRepository extends ReactiveCrudRepository<Order, Long> {
    @Query("SELECT * FROM orders WHERE user_id = :userId ORDER BY created_at DESC LIMIT :limit")
    Flux<Order> findRecentByUserId(@Param("userId") Long userId, @Param("limit") int limit);
}

// InventoryRepository.java（纯 DatabaseClient）
@Service
public class InventoryRepository {
    private final DatabaseClient db;
    public Mono<Integer> deduct(String symbol, int qty) {
        return db.sql("UPDATE inventory SET quantity = quantity - :qty WHERE symbol = :symbol AND quantity >= :qty")
            .bind("qty", qty).bind("symbol", symbol)
            .fetch().rowsUpdated();
    }
}

小结

层级	API	适用场景
高级	`ReactiveCrudRepository`	CRUD + 简单查询
中级	`@Query` 注解	自定义 SQL
低级	`DatabaseClient`	报表、聚合、复杂 JOIN
原始	R2DBC SPI	性能调优、特殊驱动功能

第 7 章结束。下一节 8.1 Schema 变更：Flyway 配置与响应式兼容。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页