Spring Boot 如何整合 Spring Data Reactive 实现响应式数据访问？

在传统的 Spring Boot 应用里，数据访问层几乎都建立在 JDBC 之上。JDBC 是典型的阻塞式 API，一次查询会一直占用当前线程，直到数据库返回结果。当并发量上来之后，Tomcat 的工作线程池很快就会被占满，请求开始排队，响应时间被拉长。响应式编程给出了一条不同的路：把数据访问也改成非阻塞的，让线程在等待数据库返回时去处理别的请求。Spring Data Reactive 就是为了这个目标而生的，它为 MongoDB、Redis、R2DBC（关系型数据库的响应式驱动）等提供了统一的响应式数据访问抽象。这篇文章就来说说如何在 Spring Boot 中整合 Spring Data Reactive，把数据层真正变成响应式的。

一、先弄清楚 JDBC 与 R2DBC 的区别

要理解 Spring Data Reactive 的价值，得先从底层驱动说起。JDBC 的设计诞生于上世纪九十年代，它的所有方法都是同步阻塞的，调用 executeQuery() 之后线程就停在那里等结果。这种模型在低并发下没什么问题，但每个等待中的线程都会占用约 1MB 的栈内存，并发一高，资源消耗非常可观。

R2DBC（Reactive Relational Database Connectivity）是专门为响应式场景设计的 relational database 连接规范，它的核心 API 基于 Reactive Streams 标准，方法返回的是 Publisher 类型。发起查询后线程不会被阻塞，数据库返回数据时通过回调的方式推送给订阅者。目前 PostgreSQL、MySQL（通过 r2dbc-mysql）、MariaDB、H2、MSSQL 都有对应的 R2DBC 驱动实现。

需要注意的是，R2DBC 不是 JDBC 的替代品，而是一个平行的规范。它牺牲了一些 JDBC 的成熟生态（比如部分监控工具的兼容性），换取了非阻塞的能力。如果你的应用并发不高、以传统 MVC 为主，强行上 R2DBC 反而增加复杂度；只有在配合 WebFlux 构建全链路非阻塞的应用时，它的价值才能充分发挥。

二、整合步骤：依赖、配置与 Reactive Repository

接下来动手搭建一个整合了 R2DBC 的响应式应用。第一步是引入依赖，注意要引入 spring-boot-starter-data-r2dbc 而不是普通的 data-jpa，同时排除默认驱动，换成 PostgreSQL 的 R2DBC 实现：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<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>

第二步是配置数据源连接。R2DBC 的连接串协议前缀是 r2dbc:postgresql，这一点和 JDBC 明显不同，很多人第一次配置时照抄 JDBC 的 URL 结果启动报错：

spring:
  r2dbc:
    url: r2dbc:postgresql://localhost:5432/orderdb
    username: postgres
    password: postgres
    pool:
      enabled: true
      initial-size: 10
      max-size: 30
      max-idle-time: 30m

第三步是定义实体和 Repository。注意 @Table 注解来自 org.springframework.data.relational.core.mapping 包，而不是 JPA 的 javax.persistence，两者很容易搞混：

// 实体类，使用 Spring Data Relational 的注解
@Table("t_order")
public class Order {
    @Id
    private Long id;
    private String orderNo;
    private BigDecimal amount;
    private String status;
    // 省略 getter 和 setter
}

// 响应式 Repository，返回类型是 Mono / Flux
public interface OrderRepository extends ReactiveCrudRepository<Order, Long> {
    Flux<Order> findByStatus(String status);
    Mono<Order> findByOrderNo(String orderNo);
}

ReactiveCrudRepository 提供的方法签名和普通的 CrudRepository 非常相似，区别在于返回值换成了 Mono（0 或 1 个元素）和 Flux（0 到 N 个元素）。方法名派生查询的规则也完全一致，findByStatus 会自动翻译成对应的 SQL。此外还可以通过 DatabaseClient 执行原生 SQL，它是 R2DBC 世界里类似 JdbcTemplate 的存在：

@Service
public class OrderService {
    private final DatabaseClient databaseClient;

    public OrderService(DatabaseClient databaseClient) {
        this.databaseClient = databaseClient;
    }

    public Flux<Order> queryBigOrders(BigDecimal minAmount) {
        return databaseClient.sql("SELECT * FROM t_order WHERE amount > :minAmount")
                .bind("minAmount", minAmount)
                .map((row, metadata) -> {
                    Order o = new Order();
                    o.setId(row.get("id", Long.class));
                    o.setOrderNo(row.get("order_no", String.class));
                    o.setAmount(row.get("amount", BigDecimal.class));
                    o.setStatus(row.get("status", String.class));
                    return o;
                })
                .all();
    }
}

三、与 WebFlux 配合：Controller 层的正确姿势

数据层响应式了，Web 层也要跟上。Controller 应该直接返回 Mono 或 Flux，配合 spring-boot-starter-webflux 提供的 Netty 服务器，整条链路都是非阻塞的：

@RestController
@RequestMapping("/api/orders")
public class OrderController {
    private final OrderRepository orderRepository;

    public OrderController(OrderRepository orderRepository) {
        this.orderRepository = orderRepository;
    }

    @GetMapping("/{id}")
    public Mono<Order> getById(@PathVariable Long id) {
        return orderRepository.findById(id);
    }

    @GetMapping
    public Flux<Order> list(@RequestParam(defaultValue = "NEW") String status) {
        return orderRepository.findByStatus(status);
    }

    @PostMapping
    public Mono<Order> create(@RequestBody Order order) {
        return orderRepository.save(order);
    }
}

这里有几个容易踩的坑要提醒一下。第一个是千万不要在响应式链路里调用 block()，它会把当前线程阻塞住，等于把非阻塞又改回了同步，在 Netty 的 EventLoop 线程上调用甚至可能直接抛出异常导致死锁。如果确实需要和阻塞代码交互，应该用 subscribeOn(Schedulers.boundedElastic()) 把阻塞操作调度到专门的弹性线程池。

第二个坑是事务。传统的 @Transactional 在响应式环境下不能直接用，必须引入响应式事务管理器 R2dbcTransactionManager，Spring Boot 在检测到 R2DBC 依赖后会自动装配它。事务方法的返回值也必须是 Mono 或 Flux，事务的提交和回滚发生在流终止时，而不是方法返回时，这个时序差异需要特别注意：

@Service
public class OrderTxService {
    private final OrderRepository orderRepository;

    public OrderTxService(OrderRepository orderRepository) {
        this.orderRepository = orderRepository;
    }

    @Transactional
    public Mono<Order> createOrder(Order order) {
        return orderRepository.save(order)
                .flatMap(saved -> {
                    // 后续操作失败时，前面的 save 会一起回滚
                    return doSomething(saved).thenReturn(saved);
                });
    }
}

四、切换到 MongoDB 与常见优化建议

Spring Data Reactive 最早其实是在 MongoDB 上落地的，如果你的数据天然适合文档模型，切换成本几乎为零。把依赖换成 spring-boot-starter-data-mongodb-reactive，配置改成连接串形式，Repository 继承 ReactiveMongoRepository，其余代码结构完全不用动：

public interface UserRepository extends ReactiveMongoRepository<User, String> {
    Flux<User> findByAgeGreaterThan(int age);
}

不管用哪种数据库，有几点实践经验值得参考。其一，连接池参数要按实际并发压测调整，R2DBC 连接池的 max-size 过小会在高并发下出现获取连接排队，过大则可能压垮数据库。其二，善用背压，当 Flux 拉取大量数据时，可以通过 limitRate() 控制向数据库请求的速率，避免一次性把结果集全部加载进内存。其三，日志排查时可以打开 io.r2dbc.postgresql.QUERY` 级别为 DEBUG 的日志，能看到实际执行的 SQL，方便定位性能问题。

最后总结一下：Spring Boot 整合 Spring Data Reactive 的关键在于全链路一致性，WebFlux 负责接入层的非阻塞，R2DBC 或 Reactive MongoDB 负责数据层的非阻塞，中间不能混入任何阻塞调用。做到这一点之后，少量线程就能支撑起远高于传统 MVC 模型的并发吞吐，这正是响应式架构最大的回报。

Spring Boot响应式编程Spring Data Reactive修改时间：2026-09-10 20:26:53