使用 R2DBC 进行数据访问
R2DBC(“Reactive Relational Database Connectivity”)是一个社区驱动的规范工作,旨在使用响应式模式标准化对 SQL 数据库的访问。
包层次结构
Spring 框架的 R2DBC 抽象框架包含两个不同的包
-
core
:org.springframework.r2dbc.core
包包含DatabaseClient
类以及各种相关类。请参阅 使用 R2DBC 核心类来控制基本的 R2DBC 处理和错误处理。 -
connection
:org.springframework.r2dbc.connection
包包含一个用于轻松访问ConnectionFactory
的实用程序类,以及各种简单的ConnectionFactory
实现,您可以使用它们进行测试和运行未修改的 R2DBC。请参阅 控制数据库连接。
使用 R2DBC 核心类来控制基本的 R2DBC 处理和错误处理
本节介绍如何使用 R2DBC 核心类来控制基本的 R2DBC 处理,包括错误处理。它包括以下主题
使用 DatabaseClient
DatabaseClient
是 R2DBC 核心包中的核心类。它处理资源的创建和释放,这有助于避免常见的错误,例如忘记关闭连接。它执行核心 R2DBC 工作流的基本任务(例如语句创建和执行),让应用程序代码提供 SQL 并提取结果。DatabaseClient
类
-
运行 SQL 查询
-
更新语句和存储过程调用
-
对
Result
实例进行迭代 -
捕获 R2DBC 异常并将它们转换为
org.springframework.dao
包中定义的通用、更具信息量的异常层次结构。(请参阅 一致的异常层次结构。)
该客户端具有使用响应式类型进行声明式组合的功能性、流畅的 API。
当您在代码中使用 DatabaseClient
时,您只需要实现 java.util.function
接口,为它们提供明确定义的契约。给定 DatabaseClient
类提供的 Connection
,Function
回调会创建一个 Publisher
。对于提取 Row
结果的映射函数也是如此。
您可以在 DAO 实现中通过使用 ConnectionFactory
引用直接实例化来使用 DatabaseClient
,或者您可以在 Spring IoC 容器中配置它,并将其作为 bean 引用提供给 DAO。
创建 DatabaseClient
对象的最简单方法是通过静态工厂方法,如下所示
-
Java
-
Kotlin
DatabaseClient client = DatabaseClient.create(connectionFactory);
val client = DatabaseClient.create(connectionFactory)
ConnectionFactory 应该始终在 Spring IoC 容器中配置为 bean。
|
上述方法创建了一个具有默认设置的 DatabaseClient
。
您还可以从 DatabaseClient.builder()
获取一个 Builder
实例。您可以通过调用以下方法来自定义客户端
-
….bindMarkers(…)
:提供一个特定的BindMarkersFactory
来配置命名参数到数据库绑定标记的转换。 -
….executeFunction(…)
:设置ExecuteFunction
如何运行Statement
对象。 -
….namedParameters(false)
:禁用命名参数扩展。默认情况下启用。
方言由 BindMarkersFactoryResolver 从 ConnectionFactory 解析,通常通过检查 ConnectionFactoryMetadata 来解析。
您可以通过注册一个实现 org.springframework.r2dbc.core.binding.BindMarkersFactoryResolver$BindMarkerFactoryProvider 的类,通过 META-INF/spring.factories 来让 Spring 自动发现您的 BindMarkersFactory 。BindMarkersFactoryResolver 使用 Spring 的 SpringFactoriesLoader 从类路径中发现绑定标记提供程序实现。
|
当前支持的数据库是
-
H2
-
MariaDB
-
Microsoft SQL Server
-
MySQL
-
Postgres
此类发出的所有 SQL 都在 DEBUG
级别下记录,类别对应于客户端实例的完全限定类名(通常为 DefaultDatabaseClient
)。此外,每次执行都会在反应式序列中注册一个检查点,以帮助调试。
以下部分提供了一些 DatabaseClient
使用的示例。这些示例并非 DatabaseClient
公开的全部功能的详尽列表。有关详细信息,请参阅相应的 javadoc。
执行语句
DatabaseClient
提供了运行语句的基本功能。以下示例展示了创建新表所需的最小但功能齐全的代码。
-
Java
-
Kotlin
Mono<Void> completion = client.sql("CREATE TABLE person (id VARCHAR(255) PRIMARY KEY, name VARCHAR(255), age INTEGER);")
.then();
client.sql("CREATE TABLE person (id VARCHAR(255) PRIMARY KEY, name VARCHAR(255), age INTEGER);")
.await()
DatabaseClient
旨在提供便捷、流畅的使用体验。它在执行规范的每个阶段都公开了中间、延续和终端方法。上面的示例使用 then()
返回一个完成 Publisher
,该 Publisher
在查询(或如果 SQL 查询包含多个语句,则为查询)完成后立即完成。
execute(…) 接受 SQL 查询字符串或查询 Supplier<String> ,以便在执行之前推迟实际查询创建。
|
查询 (SELECT
)
SQL 查询可以通过 Row
对象或受影响的行数返回结果。DatabaseClient
可以根据发出的查询返回更新的行数或行本身。
以下查询从表中获取 id
和 name
列
-
Java
-
Kotlin
Mono<Map<String, Object>> first = client.sql("SELECT id, name FROM person")
.fetch().first();
val first = client.sql("SELECT id, name FROM person")
.fetch().awaitSingle()
以下查询使用绑定变量
-
Java
-
Kotlin
Mono<Map<String, Object>> first = client.sql("SELECT id, name FROM person WHERE first_name = :fn")
.bind("fn", "Joe")
.fetch().first();
val first = client.sql("SELECT id, name FROM person WHERE first_name = :fn")
.bind("fn", "Joe")
.fetch().awaitSingle()
您可能已经注意到上面的示例中使用了 fetch()
。fetch()
是一个延续运算符,允许您指定要消费的数据量。
调用 first()
返回结果中的第一行并丢弃剩余的行。您可以使用以下运算符消费数据
-
first()
返回整个结果的第一行。它的 Kotlin 协程变体对于非空返回值命名为awaitSingle()
,对于可选值命名为awaitSingleOrNull()
。 -
one()
返回正好一个结果,如果结果包含多行则失败。使用 Kotlin 协程,对于正好一个值使用awaitOne()
,对于可能为null
的值使用awaitOneOrNull()
。 -
all()
返回结果的所有行。使用 Kotlin 协程时,使用flow()
。 -
rowsUpdated()
返回受影响的行数 (INSERT
/UPDATE
/DELETE
计数)。它的 Kotlin 协程变体命名为awaitRowsUpdated()
。
在没有指定进一步的映射细节的情况下,查询将表格结果作为 Map
返回,其键是不区分大小写的列名,映射到它们的列值。
您可以通过提供一个 Function<Row, T>
来控制结果映射,该函数将为每个 Row
调用,以便它可以返回任意值(单个值、集合和映射以及对象)。
以下示例提取name
列并输出其值
-
Java
-
Kotlin
Flux<String> names = client.sql("SELECT name FROM person")
.map(row -> row.get("name", String.class))
.all();
val names = client.sql("SELECT name FROM person")
.map{ row: Row -> row.get("name", String.class) }
.flow()
或者,有一个映射到单个值的快捷方式
Flux<String> names = client.sql("SELECT name FROM person")
.mapValue(String.class)
.all();
或者您可以映射到具有 bean 属性或记录组件的结果对象
// assuming a name property on Person
Flux<Person> persons = client.sql("SELECT name FROM person")
.mapProperties(Person.class)
.all();
使用DatabaseClient
更新(INSERT
、UPDATE
和DELETE
)
修改语句的唯一区别是这些语句通常不返回表格数据,因此您使用rowsUpdated()
来使用结果。
以下示例显示一个返回更新行数的UPDATE
语句
-
Java
-
Kotlin
Mono<Integer> affectedRows = client.sql("UPDATE person SET first_name = :fn")
.bind("fn", "Joe")
.fetch().rowsUpdated();
val affectedRows = client.sql("UPDATE person SET first_name = :fn")
.bind("fn", "Joe")
.fetch().awaitRowsUpdated()
将值绑定到查询
典型的应用程序需要参数化的 SQL 语句来根据某些输入选择或更新行。这些通常是受WHERE
子句约束的SELECT
语句,或者接受输入参数的INSERT
和UPDATE
语句。如果参数没有正确转义,参数化语句存在 SQL 注入的风险。DatabaseClient
利用 R2DBC 的bind
API 来消除查询参数的 SQL 注入风险。您可以使用execute(…)
运算符提供参数化的 SQL 语句,并将参数绑定到实际的Statement
。然后,您的 R2DBC 驱动程序使用准备好的语句和参数替换来运行该语句。
参数绑定支持两种绑定策略
-
按索引,使用基于零的参数索引。
-
按名称,使用占位符名称。
以下示例显示了查询的参数绑定
db.sql("INSERT INTO person (id, name, age) VALUES(:id, :name, :age)")
.bind("id", "joe")
.bind("name", "Joe")
.bind("age", 34);
或者,您可以传入一个名称和值的映射
Map<String, Object> params = new LinkedHashMap<>();
params.put("id", "joe");
params.put("name", "Joe");
params.put("age", 34);
db.sql("INSERT INTO person (id, name, age) VALUES(:id, :name, :age)")
.bindValues(params);
或者您可以传入一个具有 bean 属性或记录组件的参数对象
// assuming id, name, age properties on Person
db.sql("INSERT INTO person (id, name, age) VALUES(:id, :name, :age)")
.bindProperties(new Person("joe", "Joe", 34);
查询预处理器将命名的 Collection
参数展开为一系列绑定标记,以消除根据参数数量进行动态查询创建的需要。嵌套的对象数组将被展开,以允许使用(例如)选择列表。
考虑以下查询
SELECT id, name, state FROM table WHERE (name, age) IN (('John', 35), ('Ann', 50))
前面的查询可以参数化并按如下方式运行
-
Java
-
Kotlin
List<Object[]> tuples = new ArrayList<>();
tuples.add(new Object[] {"John", 35});
tuples.add(new Object[] {"Ann", 50});
client.sql("SELECT id, name, state FROM table WHERE (name, age) IN (:tuples)")
.bind("tuples", tuples);
val tuples: MutableList<Array<Any>> = ArrayList()
tuples.add(arrayOf("John", 35))
tuples.add(arrayOf("Ann", 50))
client.sql("SELECT id, name, state FROM table WHERE (name, age) IN (:tuples)")
.bind("tuples", tuples)
选择列表的使用取决于供应商。 |
以下示例展示了使用 IN
谓词的更简单变体
-
Java
-
Kotlin
client.sql("SELECT id, name, state FROM table WHERE age IN (:ages)")
.bind("ages", Arrays.asList(35, 50));
client.sql("SELECT id, name, state FROM table WHERE age IN (:ages)")
.bind("ages", arrayOf(35, 50))
R2DBC 本身不支持类似 Collection 的值。尽管如此,在上面的示例中扩展给定的 List 适用于 Spring 的 R2DBC 支持中的命名参数,例如用于 IN 子句,如上所示。但是,插入或更新数组类型的列(例如在 Postgres 中)需要底层 R2DBC 驱动程序支持的数组类型:通常是 Java 数组,例如 String[] 来更新 text[] 列。不要将 Collection<String> 或类似的作为数组参数传递。
|
语句过滤器
有时您需要在实际的 Statement
运行之前微调其选项。为此,请使用 DatabaseClient
注册一个 Statement
过滤器 (StatementFilterFunction
) 来拦截和修改其执行中的语句,如下面的示例所示
-
Java
-
Kotlin
client.sql("INSERT INTO table (name, state) VALUES(:name, :state)")
.filter((s, next) -> next.execute(s.returnGeneratedValues("id")))
.bind("name", …)
.bind("state", …);
client.sql("INSERT INTO table (name, state) VALUES(:name, :state)")
.filter { s: Statement, next: ExecuteFunction -> next.execute(s.returnGeneratedValues("id")) }
.bind("name", …)
.bind("state", …)
DatabaseClient
还公开了一个简化的 filter(…)
重载,它接受一个 Function<Statement, Statement>
-
Java
-
Kotlin
client.sql("INSERT INTO table (name, state) VALUES(:name, :state)")
.filter(statement -> s.returnGeneratedValues("id"));
client.sql("SELECT id, name, state FROM table")
.filter(statement -> s.fetchSize(25));
client.sql("INSERT INTO table (name, state) VALUES(:name, :state)")
.filter { statement -> s.returnGeneratedValues("id") }
client.sql("SELECT id, name, state FROM table")
.filter { statement -> s.fetchSize(25) }
StatementFilterFunction
实现允许过滤 Statement
和过滤 Result
对象。
DatabaseClient
最佳实践
DatabaseClient
类实例在配置后是线程安全的。这很重要,因为它意味着您可以配置 DatabaseClient
的单个实例,然后安全地将此共享引用注入到多个 DAO(或存储库)中。DatabaseClient
是有状态的,因为它维护对 ConnectionFactory
的引用,但此状态不是会话状态。
使用 DatabaseClient
类时的一种常见做法是在您的 Spring 配置文件中配置一个 ConnectionFactory
,然后将该共享 ConnectionFactory
bean 依赖注入到您的 DAO 类中。DatabaseClient
在 ConnectionFactory
的 setter 中创建。这会导致类似于以下内容的 DAO
-
Java
-
Kotlin
public class R2dbcCorporateEventDao implements CorporateEventDao {
private DatabaseClient databaseClient;
public void setConnectionFactory(ConnectionFactory connectionFactory) {
this.databaseClient = DatabaseClient.create(connectionFactory);
}
// R2DBC-backed implementations of the methods on the CorporateEventDao follow...
}
class R2dbcCorporateEventDao(connectionFactory: ConnectionFactory) : CorporateEventDao {
private val databaseClient = DatabaseClient.create(connectionFactory)
// R2DBC-backed implementations of the methods on the CorporateEventDao follow...
}
显式配置的另一种选择是使用组件扫描和注释支持进行依赖注入。在这种情况下,您可以使用 @Component
注释该类(使其成为组件扫描的候选者),并使用 @Autowired
注释 ConnectionFactory
setter 方法。以下示例展示了如何操作
-
Java
-
Kotlin
@Component (1)
public class R2dbcCorporateEventDao implements CorporateEventDao {
private DatabaseClient databaseClient;
@Autowired (2)
public void setConnectionFactory(ConnectionFactory connectionFactory) {
this.databaseClient = DatabaseClient.create(connectionFactory); (3)
}
// R2DBC-backed implementations of the methods on the CorporateEventDao follow...
}
1 | 使用 @Component 注释该类。 |
2 | 使用 @Autowired 注解 ConnectionFactory 的 setter 方法。 |
3 | 使用 ConnectionFactory 创建一个新的 DatabaseClient 。 |
@Component (1)
class R2dbcCorporateEventDao(connectionFactory: ConnectionFactory) : CorporateEventDao { (2)
private val databaseClient = DatabaseClient(connectionFactory) (3)
// R2DBC-backed implementations of the methods on the CorporateEventDao follow...
}
1 | 使用 @Component 注释该类。 |
2 | 通过构造函数注入 ConnectionFactory 。 |
3 | 使用 ConnectionFactory 创建一个新的 DatabaseClient 。 |
无论您选择使用哪种模板初始化方式(或不使用),很少需要每次运行 SQL 时都创建一个新的 DatabaseClient
实例。配置完成后,DatabaseClient
实例是线程安全的。如果您的应用程序访问多个数据库,您可能需要多个 DatabaseClient
实例,这需要多个 ConnectionFactory
,进而需要多个配置不同的 DatabaseClient
实例。
检索自动生成的键
INSERT
语句在将行插入到定义了自动递增或标识列的表中时可能会生成键。要完全控制要生成的列名,只需注册一个 StatementFilterFunction
,它会请求所需列的生成键。
-
Java
-
Kotlin
Mono<Integer> generatedId = client.sql("INSERT INTO table (name, state) VALUES(:name, :state)")
.filter(statement -> s.returnGeneratedValues("id"))
.map(row -> row.get("id", Integer.class))
.first();
// generatedId emits the generated key once the INSERT statement has finished
val generatedId = client.sql("INSERT INTO table (name, state) VALUES(:name, :state)")
.filter { statement -> s.returnGeneratedValues("id") }
.map { row -> row.get("id", Integer.class) }
.awaitOne()
// generatedId emits the generated key once the INSERT statement has finished
控制数据库连接
本节介绍
使用 ConnectionFactory
Spring 通过 ConnectionFactory
获取与数据库的 R2DBC 连接。ConnectionFactory
是 R2DBC 规范的一部分,是驱动程序的常用入口点。它允许容器或框架将连接池和事务管理问题隐藏在应用程序代码中。作为开发人员,您无需了解如何连接到数据库的详细信息。这是设置 ConnectionFactory
的管理员的责任。您很可能在开发和测试代码时同时担任这两个角色,但您不一定需要知道生产数据源是如何配置的。
当您使用 Spring 的 R2DBC 层时,您可以使用第三方提供的连接池实现来配置自己的连接池。一个流行的实现是 R2DBC Pool (r2dbc-pool
)。Spring 发行版中的实现仅用于测试目的,不提供池化功能。
要配置 ConnectionFactory
-
使用
ConnectionFactory
获取连接,就像您通常获取 R2DBCConnectionFactory
一样。 -
提供一个 R2DBC URL(有关正确的值,请参阅驱动程序的文档)。
以下示例展示了如何配置 ConnectionFactory
-
Java
-
Kotlin
ConnectionFactory factory = ConnectionFactories.get("r2dbc:h2:mem:///test?options=DB_CLOSE_DELAY=-1;DB_CLOSE_ON_EXIT=FALSE");
val factory = ConnectionFactories.get("r2dbc:h2:mem:///test?options=DB_CLOSE_DELAY=-1;DB_CLOSE_ON_EXIT=FALSE");
使用 ConnectionFactoryUtils
ConnectionFactoryUtils
类是一个方便且功能强大的辅助类,它提供 static
方法来从 ConnectionFactory
获取连接并关闭连接(如果需要)。
它支持订阅者 Context
绑定的连接,例如 R2dbcTransactionManager
。
使用 SingleConnectionFactory
SingleConnectionFactory
类是 DelegatingConnectionFactory
接口的实现,它包装了一个单个 Connection
,该连接在每次使用后不会关闭。
如果任何客户端代码在假设池化连接的情况下调用 close
(如使用持久性工具时),您应该将 suppressClose
属性设置为 true
。此设置返回一个关闭抑制代理,该代理包装物理连接。请注意,您不能再将其强制转换为本机 Connection
或类似对象。
SingleConnectionFactory
主要是一个测试类,可用于特定需求,例如如果您的 R2DBC 驱动程序允许,则可用于流水线。与池化的 ConnectionFactory
相比,它始终重用同一个连接,避免过度创建物理连接。
使用 TransactionAwareConnectionFactoryProxy
TransactionAwareConnectionFactoryProxy
是目标 ConnectionFactory
的代理。代理包装目标 ConnectionFactory
以增加对 Spring 管理的事务的感知。
如果您使用的是未与 Spring 的 R2DBC 支持集成在一起的 R2DBC 客户端,则需要使用此类。在这种情况下,您仍然可以使用此客户端,同时让此客户端参与 Spring 管理的事务。通常,最好将 R2DBC 客户端与对 ConnectionFactoryUtils 的适当访问权限集成,以进行资源管理。
|
有关更多详细信息,请参阅 TransactionAwareConnectionFactoryProxy
javadoc。
使用 R2dbcTransactionManager
R2dbcTransactionManager
类是针对单个 R2DBC ConnectionFactory
的 ReactiveTransactionManager
实现。它将来自指定 ConnectionFactory
的 R2DBC Connection
绑定到订阅者 Context
,可能允许每个 ConnectionFactory
拥有一个订阅者 Connection
。
应用程序代码需要通过 ConnectionFactoryUtils.getConnection(ConnectionFactory)
检索 R2DBC Connection
,而不是 R2DBC 的标准 ConnectionFactory.create()
。所有框架类(如 DatabaseClient
)都隐式使用此策略。如果未使用事务管理器,则查找策略的行为与 ConnectionFactory.create()
完全相同,因此可以在任何情况下使用。