R2DBC 是什么
R2DBC（Reactive Relational Database Connectivity）是响应式关系型数据库连接规范。它与 JDBC 的关系类似 Reactor 与阻塞 I/O：
| 维度 | JDBC | R2DBC |
|---|---|---|
| API 风格 | 同步阻塞 | 响应式非阻塞 |
| 返回值 | T / List<T> | Mono<T> / Flux<T> |
| 连接获取 | DataSource.getConnection() 阻塞 | ConnectionFactory.create().subscribe() |
| 结果遍历 | ResultSet.next() 循环 | Result.map(...) 流式 |
| 事务 | Connection.commit() 阻塞 | Connection.commitTransaction() 返回 Publisher<Void> |
| 线程模型 | 线程持有连接 | EventLoop 驱动 I/O |
R2DBC 不是 JDBC 的响应式包装——它是全新协议栈，驱动层直接与数据库的异步协议交互。
核心接口
R2DBC 定义四个核心接口：
// 1. 连接工厂：创建数据库连接
io.r2dbc.spi.ConnectionFactory
// 2. 连接：执行语句、管理事务
io.r2dbc.spi.Connection
// 3. 语句：绑定参数、执行查询
io.r2dbc.spi.Statement
// 4. 结果：流式消费行数据
io.r2dbc.spi.Result
创建连接
import io.r2dbc.postgresql.PostgresqlConnectionConfiguration;
import io.r2dbc.postgresql.PostgresqlConnectionFactory;
import io.r2dbc.spi.ConnectionFactory;
ConnectionFactory factory = new PostgresqlConnectionFactory(
PostgresqlConnectionConfiguration.builder()
.host("localhost")
.port(5432)
.database("eventd")
.username("eventd")
.password("eventd")
.build()
);
// 获取连接（异步）
Mono<io.r2dbc.spi.Connection> connection = Mono.from(factory.create());
factory.create() 返回 Publisher<Connection>，需要用 Mono.from() 转为 Reactor 类型。
执行查询
Mono<User> user = Mono.usingWhen(
factory.create(), // 资源：连接
conn -> Mono.from(conn.createStatement("SELECT id, name FROM users WHERE id = $1")
.bind("$1", 1)
.execute())
.flatMap(result -> Mono.from(result.map((row, metadata) ->
new User(row.get("id", Long.class), row.get("name", String.class))))),
conn -> Mono.from(conn.close()) // 正常完成时关闭
);
Connection.close() 返回的是 Publisher<Void>，不是阻塞方法。写成 doFinally(sig -> conn.close()) 只会创建一个从未被订阅的 Publisher，连接不会被归还；必须用 Mono.from(conn.close()) 或交给 usingWhen 统一管理。usingWhen 还会在取消和异常路径上关闭连接，比手写 doFinally 更可靠。
执行链：
ConnectionFactory.create()
↓ Mono<Connection>
Connection.createStatement("SQL")
↓ Statement
Statement.bind("$1", value).execute()
↓ Publisher<Result>
Result.map((row, metadata) -> ...)
↓ Flux<User>
driver 异步读取 PostgreSQL 协议数据
↓ 背压：request(n) 控制读取批次
tcp socket
Statement 参数绑定
占位符风格由驱动的 Statement 实现决定，SPI 层只有两种绑定方式：按名字绑定（bind(String, ...)）和按 0-based 索引绑定（bind(int, ...)）：
| 数据库 | SQL 占位符 | 绑定方式 |
|---|---|---|
| PostgreSQL | $1, $2 | stmt.bind("$1", value)，名字就是 "$1" |
| MySQL | ? | stmt.bind(0, value)，0-based |
| H2 | $1 或 ? | 两种都支持，按实际 SQL 选择 |
// PostgreSQL：字符串名字绑定
Statement stmt = conn.createStatement("SELECT * FROM users WHERE id = $1 AND status = $2");
stmt.bind("$1", 1);
stmt.bind("$2", "ACTIVE");
// MySQL：0-based 索引绑定
Statement stmt = conn.createStatement("SELECT * FROM users WHERE id = ? AND status = ?");
stmt.bind(0, 1);
stmt.bind(1, "ACTIVE");
Spring Data R2DBC 的 :name 命名参数不是驱动能力：它在 Spring 层把 :name 改写成驱动认识的占位符，再逐个绑定，所以底层 Statement 里看不到 :name。
插入与返回生成键
Mono<Long> insertedId = Mono.from(conn.createStatement(
"INSERT INTO orders (symbol, quantity, price) VALUES ($1, $2, $3)")
.bind("$1", "BTC")
.bind("$2", 1)
.bind("$3", 40000)
.returnGeneratedValues("id") // PostgreSQL 会据此追加 RETURNING id
.execute())
.flatMap(result -> Mono.from(result.map((row, meta) -> row.get("id", Long.class))));
returnGeneratedValues("id") 之后，结果集里是生成的主键行，要用 result.map(...) 读；换成 getRowsUpdated() 拿到的是受影响行数，主键就丢了。两者是互斥的消费方式，不能在同一结果上依次取。PostgreSQL 用 RETURNING 实现这个语义；MySQL 没有 RETURNING（MariaDB 才有），驱动通常改用 SELECT LAST_INSERT_ID() 再查一次。
只关心影响行数时则相反：
Mono<Long> affected = Mono.from(conn.createStatement(...).execute())
.flatMap(result -> Mono.from(result.getRowsUpdated()));
批量操作
R2DBC Statement 支持 add（分批绑定）：
Statement stmt = conn.createStatement("INSERT INTO logs (level, message) VALUES ($1, $2)");
stmt.bind("$1", "INFO").bind("$2", "Order created").add();
stmt.bind("$1", "WARN").bind("$2", "Low stock").add();
stmt.bind("$1", "ERROR").bind("$2", "Payment failed").add();
// 每次 add() 会产生一个 Result；用 Mono.from 只会取第一个并取消其余结果
Flux.from(stmt.execute())
.flatMap(Result::getRowsUpdated)
.reduce(0L, Long::sum)
.subscribe(total -> System.out.println("Total inserted: " + total));
三次 add() 表示三条 INSERT 在同一条 prepared statement 中批量发送，减少网络往返。
背压在 Result 流中
R2DBC Result 的 map() 返回 Publisher<T>，订阅端用 request(n) 向上游要数据。但「处理慢」不等于「自动在别的线程上处理」——如果 heavyProcess 是阻塞调用，必须显式切到弹性线程池：
result.map((row, meta) -> row.get("data", String.class))
.flatMap(row -> Mono.fromCallable(() -> heavyProcess(row))
.subscribeOn(Schedulers.boundedElastic()), 4) // 最多 4 个并发
.subscribe();
subscribeOn 写在链尾不会逐个隔离 heavyProcess：它只影响整条链的订阅线程，而 flatMap 的映射函数仍在驱动发信号的线程上执行。逐个切线程必须把 subscribeOn 放进 flatMap 内部的 Mono 里，并用并发参数限制同时占用弹性线程的数量。
背压能否传到数据库，还取决于驱动的读取方式：PostgreSQL 只有在游标/fetchSize 大于 0 时才分批取行；fetchSize 为 0 时服务端一次把结果发完，客户端再慢也挡不住服务端发送。把 request(n) 一路传到 TCP 窗口，需要驱动配置和语句级 fetchSize 配合，不是 SPI 自动保证的。
与 Spring Data R2DBC 的关系
原始 R2DBC SPI 很底层。Spring Data R2DBC 提供更高级的封装（第 7.3 节）：
Spring Data R2DBC
↓ DatabaseClient / ReactiveRepository
R2DBC SPI (io.r2dbc.spi)
↓ ConnectionFactory / Connection / Statement / Result
PostgreSQL Wire Protocol
↓ TCP socket
PostgreSQL Server
本章先掌握底层 SPI，再理解 Spring 的封装才有意义。
eventd v2 的数据层设计
eventd-data 模块使用 R2DBC SPI 直接构建数据访问层：
eventd-data/
├── DataSourceConfig.java # ConnectionFactory Bean
├── UserRepository.java # 原始 R2DBC 查询（本章）
├── OrderRepository.java # 原始 R2DBC 查询
├── TransactionConfig.java # 响应式事务（第 7.2 节）
└── SpringDataConfig.java # Spring Data R2DBC（第 7.3 节）
生产版本迁移到 Spring Data R2DBC，但底层 SPI 的理解是排查性能问题的必备知识。
常见错误
| 错误 | 原因 | 修复 |
|---|---|---|
| 连接未归还、池被耗尽 | doFinally(sig -> conn.close()) 没有订阅返回的 Publisher | 用 Mono.usingWhen 或 Mono.from(conn.close()) |
Connection is closed | 关闭之后还有操作在订阅 | 所有数据库操作完成后再关闭连接 |
Parameter index out of bounds | MySQL 驱动用 0-based，PostgreSQL 用 $1 | 确认驱动占位符风格 |
Timeout waiting for connection | 连接池耗尽 | 增加 maxSize 或减少并发 |
DataBuffer Leak | ByteBuf 未释放 | 使用 DataBufferUtils.release() |
小结
| 概念 | SPI | Spring Data 封装 |
|---|---|---|
| 连接创建 | ConnectionFactory.create() | ConnectionFactory Bean |
| 查询执行 | Statement.execute() | DatabaseClient.sql() |
| 结果消费 | Result.map(...) | DatabaseClient.map(...) |
| 参数绑定 | Statement.bind("$1", value) | .bind("name", value) |
| 批量 | Statement.add() | 逐条执行或 bindValues(List) 配合多值 SQL |
| 生成键 | returnGeneratedValues("id") + result.map(...) | .map((row, meta) -> row.get("id")) |
| 命名参数 | 驱动不支持 | SQL 里写 :name，.bind("name", value) |
| 背压 | 原生 Publisher 支持 | 继承原生支持 |
下一节 7.2 响应式事务 ：TransactionalOperator、事务传播、回滚策略、savepoint 与编程式事务。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。