1. Overview
1.概述
In this tutorial, we’ll show how we can use R2DBC to perform database operations in a reactive way.
在本教程中，我们将展示如何使用R2DBC来以反应式方式执行数据库操作。
In order to explore R2DBC, we’ll create a simple Spring WebFlux REST application that implements CRUD operations for a single entity, using only asynchronous operations to achieve that goal.
为了探索R2DBC，我们将创建一个简单的Spring WebFlux REST应用，为一个实体实现CRUD操作，只使用异步操作来实现这一目标。
2. What Is R2DBC?
2.什么是R2DBC？
Reactive development is on the rise, with new frameworks coming every day and existing ones seeing increasing adoption. However, a major issue with reactive development is the fact that database access in the Java/JVM world remains basically synchronous. This is a direct consequence of the way JDBC was designed and led to some ugly hacks to adapt those two fundamentally different approaches.
反应式开发正在兴起，每天都有新的框架出现，而现有的框架也在不断被采用。然而，反应式开发的一个主要问题是，Java/JVM世界中的数据库访问基本上仍然是同步的。这是JDBC设计方式的直接后果，导致了一些丑陋的黑客来适应这两种根本不同的方法。
To address the need for asynchronous database access in the Java land, two standards have emerged. The first one, ADBC (Asynchronous Database Access API), is backed by Oracle but, as of this writing, seems to be somewhat stalled, with no clear timeline.
为了解决Java领域对异步数据库访问的需求，已经出现了两个标准。第一个，ADBC（异步数据库访问API），由甲骨文公司支持，但截至目前，似乎有些停滞，没有明确的时间表。
The second one, which we’ll cover here, is R2DBC (Reactive Relational Database Connectivity), a community effort led by a team from Pivotal and other companies. This project, which is still in beta, has shown more vitality and already provides drivers for Postgres, H2, and MSSQL databases.
第二个是R2DBC（Reactive Relational Database Connectivity），这是一个由Pivotal公司和其他公司的团队领导的社区工作，我们将在这里介绍。这个项目仍处于测试阶段，显示出更大的活力，已经为Postgres、H2和MSSQL数据库提供驱动。
3. Project Setup
3.项目设置
Using R2DBC in a project requires that we add dependencies to the core API and a suitable driver. In our example, we’ll be using H2, so this means just two dependencies:
在一个项目中使用R2DBC，需要我们在核心API和合适的驱动中添加依赖项。在我们的例子中，我们将使用H2，所以这意味着只有两个依赖项。
<dependency>
<groupId>io.r2dbc</groupId>
<artifactId>r2dbc-spi</artifactId>
<version>0.8.0.M7</version>
</dependency>
<dependency>
<groupId>io.r2dbc</groupId>
<artifactId>r2dbc-h2</artifactId>
<version>0.8.0.M7</version>
</dependency>Maven Central still has no R2DBC artifacts for now, so we also need to add a couple of Spring’s repositories to our project:
Maven中心暂时还没有R2DBC工件，所以我们还需要在项目中添加几个Spring的仓库。
<repositories>
<repository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/milestone</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</repository>
<repository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/snapshot</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
</repository>
</repositories>4. Connection Factory Setup
4.连接工厂设置
The first thing we need to do to access a database using R2DBC is to create a ConnectionFactory object, which plays a similar role to JDBC’s DataSource. The most straightforward way to create a ConnectionFactory is through the ConnectionFactories class.
我们使用R2DBC访问数据库需要做的第一件事是创建一个ConnectionFactory对象，它的作用类似于JDBC的DataSource。创建ConnectionFactory最直接的方法是通过ConnectionFactories类。
This class has static methods that take a ConnectionFactoryOptions object and return a ConnectionFactory. Since we’ll only need a single instance of our ConnectionFactory, let’s create a @Bean that we can later use via injection wherever we need:
这个类有一些静态方法，可以接收一个ConnectionFactoryOptions对象并返回一个ConnectionFactory。由于我们只需要一个ConnectionFactory的实例，让我们创建一个@Bean，以后我们可以在需要的地方通过注入使用。
@Bean
public ConnectionFactory connectionFactory(R2DBCConfigurationProperties properties) {
ConnectionFactoryOptions baseOptions = ConnectionFactoryOptions.parse(properties.getUrl());
Builder ob = ConnectionFactoryOptions.builder().from(baseOptions);
if (!StringUtil.isNullOrEmpty(properties.getUser())) {
ob = ob.option(USER, properties.getUser());
}
if (!StringUtil.isNullOrEmpty(properties.getPassword())) {
ob = ob.option(PASSWORD, properties.getPassword());
}
return ConnectionFactories.get(ob.build());
}
Here, we take options received from a helper class decorated with the @ConfigurationProperties annotation and populate our ConnectionFactoryOptions instance. To populate it, R2DBC implements a builder pattern with a single option method that takes an Option and a value.
在这里，我们从一个用@ConfigurationProperties注解装饰的辅助类中接收选项，并填充我们的ConnectionFactoryOptions实例。为了填充它，R2DBC实现了一个带有单个option方法的构建器模式，该方法接收一个Option和一个值。
R2DBC defines a number of well-known options, such as USERNAME and PASSWORD that we’ve used above. Another way to set those options is to pass a connection string to the parse() method of the ConnectionFactoryOptions class.
R2DBC定义了一些众所周知的选项，比如我们上面使用的USERNAME和PASSWORD。设置这些选项的另一种方法是将连接字符串传递给ConnectionFactoryOptions类的parse()方法。
Here’s an example of a typical R2DBC connection URL:
下面是一个典型的R2DBC连接URL的例子。
r2dbc:h2:mem://./testdbLet’s break this string into its components:
让我们把这个字符串分解成其组成部分。
- r2dbc: Fixed-scheme identifier for R2DBC URLs — another valid scheme is rd2bcs, used for SSL-secured connections
- h2: Driver identifier used to locate the appropriate connection factory
- mem: Driver-specific protocol — in our case, this corresponds to an in-memory database
- //./testdb: Driver-specific string, usually containing host, database, and any additional options.
Once we have our option set ready, we pass it to the get() static factory method to create our ConnectionFactory bean.
一旦我们准备好了我们的选项集，我们就把它传递给get() 静态工厂方法来创建我们的ConnectionFactory bean。
5. Executing Statements
5.执行声明
Similarly to JDBC, using R2DBC is mostly about sending SQL statements to the database and processing result sets. However, since R2DBC is a reactive API, it depends heavily on reactive streams types, such as Publisher and Subscriber.
与JDBC类似，使用R2DBC主要是向数据库发送SQL语句并处理结果集。然而，由于R2DBC是一个反应式API，它在很大程度上依赖于反应式流的类型，如Publisher 和Subscriber。
Using those types directly is a bit cumbersome, so we’ll use project reactor’s types like Mono and Flux that help us to write cleaner and more concise code.
直接使用这些类型有点麻烦，所以我们将使用项目反应器的类型，如Mono和Flux，它们可以帮助我们写出更干净、更简洁的代码。
In the next sections, we’ll see how to implement database-related tasks by creating a reactive DAO class for a simple Account class. This class contains just three properties and has a corresponding table in our database:
在接下来的章节中，我们将看到如何通过为一个简单的Account类创建一个反应式DAO类来实现数据库相关的任务。这个类只包含三个属性，并且在我们的数据库中有一个相应的表。
public class Account {
private Long id;
private String iban;
private BigDecimal balance;
// ... getters and setters omitted
}5.1. Getting a Connection
5.1.获得连接
Before we can send any statements to the database, we need a Connection instance. We’ve already seen how to create a ConnectionFactory, so it’s no surprise that we’ll use it to get a Connection. What we must remember is that now, instead of getting a regular Connection, what we get is a Publisher of a single Connection.
在我们向数据库发送任何语句之前，我们需要一个Connection实例。我们已经看到了如何创建一个ConnectionFactory，所以我们用它来获取一个Connection也就不奇怪了。我们必须记住的是，现在我们得到的不是一个普通的Connection，而是一个单一Connection的Publisher。
Our ReactiveAccountDao, which is a regular Spring @Component, gets its ConnectionFactory via constructor injection, so it’s readily available in handler methods.
我们的ReactiveAccountDao，是一个常规的Spring @Component，它通过构造函数注入获得其ConnectionFactory，因此它可以在处理方法中随时使用。
Let’s take a look at the first couple of lines of the findById() method to see how to retrieve and start using a Connection:
让我们看一下findById()方法的前几行，看看如何检索并开始使用Connection。
public Mono<Account>> findById(Long id) {
return Mono.from(connectionFactory.create())
.flatMap(c ->
// use the connection
)
// ... downstream processing omitted
}Here, we’re adapting the Publisher returned from our ConnectionFactory into a Mono that is the initial source for our event stream.
在这里，我们将从ConnectionFactory返回的Publisher调整为Mono，它是我们事件流的初始源。
5.1. Preparing and Submitting Statements
5.1.准备和提交声明
Now that we have a Connection, let’s use it to create a Statement and bind a parameter to it:
现在我们有了一个Connection，让我们用它来创建一个Statement并为它绑定一个参数。
.flatMap( c ->
Mono.from(c.createStatement("select id,iban,balance from Account where id = $1")
.bind("$1", id)
.execute())
.doFinally((st) -> close(c))
)
The Connection‘s method createStatement takes a SQL query string, which can optionally have bind placeholders — referred to as “markers” in the spec.
Connection的方法createStatement接收一个SQL查询字符串，该字符串可以选择绑定占位符 – 在规范中被称为 “标记” 。
A couple of noteworthy points here: first, createStatement is a synchronous operation, which allows us to use a fluent style to bind values to the returned Statement; second, and very important, placeholder/marker syntax is vendor-specific!
这里有几个值得注意的地方：首先，createStatement是一个同步操作，这使得我们可以使用流畅的风格来为返回的Statement绑定值；第二，也是非常重要的一点，placeholder/marker语法是供应商特定的！。
In this example, we’re using H2’s specific syntax, which uses $n to mark parameters. Other vendors may use different syntax, such as :param, @Pn, or some other convention. This is an important aspect that we must pay attention to when migrating legacy code to this new API.
在这个例子中，我们使用了H2的特定语法，它使用$n来标记参数。其他供应商可能使用不同的语法，如:param、@Pn，或其他一些约定。这是我们在将遗留代码迁移到这个新API时必须注意的一个重要方面。
The binding process itself is quite straightforward, due to the fluent API pattern and simplified typing: there’s just a single overloaded bind() method that takes care of all typing conversions — subject to database rules, of course.
由于流畅的API模式和简化的类型，绑定过程本身是非常简单的。只有一个重载的bind()方法来处理所有的类型转换–当然，要遵守数据库规则。
The first parameter passed to bind() can be a zero-based ordinal that corresponds to the marker’s placement in the statement, or it can be a string with the actual marker.
传递给bind()的第一个参数可以是一个基于零的序号，对应于标记在语句中的位置，也可以是一个包含实际标记的字符串。
Once we’ve set values to all parameters, we call execute(), which returns a Publisher of Result objects, which we again wrap into a Mono for further processing. We attach a doFinally() handler to this Mono so that we make sure that we’ll close our connection whether the stream processing completes normally or not.
一旦我们为所有参数设置了值，我们就会调用execute()，它将返回一个Publisher的Result对象，我们再次将其包装成Mono，以便进一步处理。我们将一个doFinally()处理程序附加到这个Mono中，这样我们就能确保无论流处理是否正常完成，我们都会关闭我们的连接。
5.2. Processing Results
5.2.处理结果
The next step in our pipeline is responsible for processing Result objects and generating a stream of ResponseEntity<Account> instances.
我们管道的下一步是负责处理Result对象并生成ResponseEntity<Account> instances。
Since we know that there can be only one instance with the given id, we’ll actually return a Mono stream. The actual conversion happens inside the function passed to the map() method of the received Result:
因为我们知道只有一个具有给定id的实例，我们实际上将返回一个Mono流。实际的转换发生在传递给接收到的Result的map()方法的函数中。
.map(result -> result.map((row, meta) ->
new Account(row.get("id", Long.class),
row.get("iban", String.class),
row.get("balance", BigDecimal.class))))
.flatMap(p -> Mono.from(p));
The result’s map() method expects a function that takes two parameters. The first one is a Row object that we use to gather values for each column and populate an Account instance. The second, meta, is a RowMetadata object that contains information about the current row, such as column names and types.
结果的map()方法期望一个函数，它需要两个参数。第一个是一个Row对象，我们用它来收集每一列的值并填充一个Account实例。第二个，meta，是一个RowMetadata对象，包含关于当前行的信息，如列名和类型。
The previous map() call in our pipeline resolves to a Mono<Producer<Account>>, but we need to return a Mono<Account> from this method. To fix this, we add a final flatMap() step, which adapts the Producer into a Mono.
我们管道中的前一个map()调用解析为一个Mono<Producer<Account>>，但是我们需要从这个方法返回一个Mono<Account>。为了解决这个问题，我们添加了一个最后的flatMap()步骤，它将Producer调整为Mono.。
5.3. Batch Statements
5.3.批量语句
R2DBC also supports the creation and execution of statement batches, which allow for the execution of multiple SQL statements in a single execute() call. In contrast with regular statements, batch statements do not support binding and are mainly used for performance reasons in scenarios such as ETL jobs.
R2DBC还支持创建和执行语句批处理，它允许在一次execute()调用中执行多个SQL语句。与常规语句相比，批处理语句不支持绑定，主要用于ETL作业等场景中的性能原因。
Our sample project uses a batch of statements to create the Account table and insert some test data into it:
我们的示例项目使用一批语句来创建Account表并向其中插入一些测试数据。
@Bean
public CommandLineRunner initDatabase(ConnectionFactory cf) {
return (args) ->
Flux.from(cf.create())
.flatMap(c ->
Flux.from(c.createBatch()
.add("drop table if exists Account")
.add("create table Account(" +
"id IDENTITY(1,1)," +
"iban varchar(80) not null," +
"balance DECIMAL(18,2) not null)")
.add("insert into Account(iban,balance)" +
"values('BR430120980198201982',100.00)")
.add("insert into Account(iban,balance)" +
"values('BR430120998729871000',250.00)")
.execute())
.doFinally((st) -> c.close())
)
.log()
.blockLast();
}Here, we use the Batch returned from createBatch() and add a few SQL statements. We then send those statements for execution using the same execute() method available in the Statement interface.
在这里，我们使用从createBatch()返回的Batch并添加一些SQL语句。然后我们使用Execute()方法将这些语句发送给Statement界面中的执行。
In this particular case, we are not interested in any results — just that the statements all execute fine. Had we needed any produced results, all we had to do is to add a downstream step in this stream to process the emitted Result objects.
在这个特殊的案例中，我们对任何结果都不感兴趣–只是对语句的执行情况感兴趣。如果我们需要任何产生的结果，我们所要做的就是在这个流中添加一个下游步骤来处理发出的Result对象。
6. Transactions
6.事务
The last topic we’ll cover in this tutorial is transactions. As we should expect by now, we manage transactions as in JDBC, that is, by using methods available in the Connection object.
在本教程中，我们将涉及的最后一个主题是事务。正如我们现在所期望的那样，我们像在JDBC中一样管理事务，也就是说，通过使用Connection 对象中的方法。
As before, the main difference is that now all transaction-related methods are asynchronous, returning a Publisher that we must add to our stream at appropriate points.
和以前一样，主要的区别是，现在所有与事务相关的方法都是异步的，返回一个Publisher，我们必须在适当的时候将其添加到我们的流中。
Our sample project uses a transaction in its implementation of the createAccount() method:
我们的示例项目在实现createAccount() 方法时使用了一个事务。
public Mono<Account> createAccount(Account account) {
return Mono.from(connectionFactory.create())
.flatMap(c -> Mono.from(c.beginTransaction())
.then(Mono.from(c.createStatement("insert into Account(iban,balance) values($1,$2)")
.bind("$1", account.getIban())
.bind("$2", account.getBalance())
.returnGeneratedValues("id")
.execute()))
.map(result -> result.map((row, meta) ->
new Account(row.get("id", Long.class),
account.getIban(),
account.getBalance())))
.flatMap(pub -> Mono.from(pub))
.delayUntil(r -> c.commitTransaction())
.doFinally((st) -> c.close()));
}Here, we’ve added transaction-related calls in two points. First, right after getting a new connection from the database, we call the beginTransactionMethod(). Once we know that the transaction was successfully started, we prepare and execute the insert statement.
在这里，我们在两个点上添加了与交易相关的调用。首先，在从数据库获得一个新的连接后，我们调用beginTransactionMethod()。一旦我们知道交易被成功启动，我们就准备并执行insert语句。
This time we’ve also used the returnGeneratedValues() method to instruct the database to return the identity value generated for this new Account. R2DBC returns those values in a Result containing a single row with all generated values, which we use to create the Account instance.
这次我们也使用了returnGeneratedValues()方法来指示数据库返回为这个新Account生成的身份值。R2DBC在一个Result中返回这些值，其中包含所有生成的值的单一行，我们用它来创建Account实例。
Once again, we need to adapt the incoming Mono<Publisher<Account>> into a Mono<Account>, so we add a flatMap() to solve this. Next, we commit the transaction in a delayUntil() step. We need this because we want to make sure the returned Account has already been committed to the database.
再次，我们需要将传入的Mono<Publisher<Account>>调整为Mono<Account>，所以我们添加一个flatMap()来解决这个问题。接下来，我们在一个delayUntil()步骤中提交事务。我们需要这样做，因为我们要确保返回的账户已经被提交到数据库中。
Finally, we attach a doFinally step to this pipeline that closes the Connection when all events from the returned Mono are consumed.
最后，我们给这个管道附加一个doFinally步骤，当返回的Mono中的所有事件都被消耗后，关闭Connection。
7. Sample DAO Usage
7.DAO使用示例
Now that we have a reactive DAO, let’s use it to create a simple Spring WebFlux application to showcase how to use it in a typical application. Since this framework already supports reactive constructs, this becomes a trivial task. For instance, let’s take a look at the implementation of the GET method:
现在我们有了一个反应式DAO，让我们用它来创建一个简单的Spring WebFlux应用程序，以展示如何在典型应用程序中使用它。由于该框架已经支持反应式结构，这就成为一项微不足道的任务。例如，让我们看一下GET方法的实现。
@RestController
public class AccountResource {
private final ReactiveAccountDao accountDao;
public AccountResource(ReactiveAccountDao accountDao) {
this.accountDao = accountDao;
}
@GetMapping("/accounts/{id}")
public Mono<ResponseEntity<Account>> getAccount(@PathVariable("id") Long id) {
return accountDao.findById(id)
.map(acc -> new ResponseEntity<>(acc, HttpStatus.OK))
.switchIfEmpty(Mono.just(new ResponseEntity<>(null, HttpStatus.NOT_FOUND)));
}
// ... other methods omitted
}Here, we’re using our DAO’s returned Mono to construct a ResponseEntity with the appropriate status code. We’re doing this just because we want a NOT_FOUND (404) status code when there is no Account with the given id.
在这里，我们使用 DAO 返回的 Mono 来构造一个具有适当状态代码的 ResponseEntity。我们这样做只是因为我们想要一个NOT_FOUND(404)状态代码，当没有Account具有给定的id。
8. Conclusion
8.结语
In this article, we’ve covered the basics of reactive database access using R2DBC. Although in its infancy, this project is quickly evolving, targeting a release date sometime in early 2020.
在这篇文章中，我们已经介绍了使用R2DBC的反应式数据库访问的基础知识。尽管处于起步阶段，这个项目正在迅速发展，目标是在2020年初的某个时候发布。
Compared to ADBA, which will definitely not be part of Java 12, R2DBC seems to be more promising and already provides drivers for a few popular databases — Oracle being a notable absence here.
与肯定不会成为Java 12一部分的ADBA相比，R2DBC似乎更有前途，并且已经为一些流行的数据库提供了驱动–Oracle是这里明显的缺席。
As usual, the complete source code used in this tutorial is available over on Github.
像往常一样，本教程中使用的完整源代码可在Github上获取。