使用Spring Data R2DBC进行异步RDBMS访问

不久之前，发布了JDBC驱动程序的反应型。称为R2DBC。它允许将数据异步流传输到已预订的任何端点。通过将R2DBC之类的反应性驱动程序与Spring WebFlux结合使用，可以编写一个完整的应用程序，以异步方式处理数据的接收和发送。在本文中，我们将重点介绍数据库。从连接到数据库，然后最终保存和检索数据。为此，我们将使用Spring Data。与所有Spring Data模块一样，它为我们提供了现成的配置。减少为实现应用程序设置而需要编写的样板代码数量。最重要的是，它在数据库驱动程序上提供了一层，使执行简单任务变得更加容易，而较困难的任务则减轻了一些痛苦。

对于本文的内容，我正在使用Postgres数据库。在撰写本文时，仅Postgres，H2和Microsoft SQL Server具有其自己的R2DBC驱动程序实现。

我以前写过两篇有关反应式Spring Data库的文章，一篇关于Mongo ，另一篇关于Cassandra 。您可能已经注意到，这些数据库都不是RDBMS数据库。现在有很长一段时间都可以使用其他反应式驱动程序（我将近两年前写了Mongo文章），但是在为RDBMS数据库编写反应式驱动程序时，这仍然是一件很新的事情。这篇文章将遵循类似的格式。

此外，我还写了一篇关于使用Spring WebFlux的文章，我在引言中提到过。如果您有兴趣生产完全反应式的Web应用程序，请随时查看。

依存关系

<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><dependency><groupId>org.springframework.data</groupId><artifactId>spring-data-r2dbc</artifactId><version>1.0.0.M1</version></dependency><dependency><groupId>io.r2dbc</groupId><artifactId>r2dbc-postgresql</artifactId><version>1.0.0.M6</version></dependency><dependency><groupId>io.projectreactor</groupId><artifactId>reactor-core</artifactId></dependency>
</dependencies><repositories><repository><id>repository.spring.milestone</id><name>Spring Milestone Repository</name><url>http://repo.spring.io/milestone</url></repository>
</repositories>

这里有几点要指出。

使用Spring Boot的次数越多，就越会习惯于为您想做的很酷的事情导入单个spring-boot-starter依赖项。例如，我希望会有spring-boot-starter-r2dbc依赖关系，但不幸的是，没有依赖关系。然而。简而言之，该库位于较新的一侧，在编写本文时，它没有自己的Spring Boot模块，该模块包含所需的任何依赖项以及通过自动配置的更快设置。我确信这些事情会在某个时候出现，并使设置R2DBC驱动程序变得更加容易。

现在，我们将需要手动填写一些额外的依赖项。

此外，R2DBC库仅具有Milestone版本（更多证明是新版本），因此我们需要确保引入Spring Milestone存储库。当我获得发布版本时，将来可能会需要更新此帖子。

连接到数据库

由于Spring Data为我们做了很多工作，因此唯一需要手动创建的Bean是ConnectionFactory ，其中包含数据库的连接详细信息：

@Configuration
@EnableR2dbcRepositories
class DatabaseConfiguration(@Value("\${spring.data.postgres.host}") private val host: String,@Value("\${spring.data.postgres.port}") private val port: Int,@Value("\${spring.data.postgres.database}") private val database: String,@Value("\${spring.data.postgres.username}") private val username: String,@Value("\${spring.data.postgres.password}") private val password: String
) : AbstractR2dbcConfiguration() {override fun connectionFactory(): ConnectionFactory {return PostgresqlConnectionFactory(PostgresqlConnectionConfiguration.builder().host(host).port(port).database(database).username(username).password(password).build())}
}

这里首先要注意的是AbstractR2dbcConfiguration的扩展。此类包含大量不再需要手动创建的Bean。实现connectionFactory是该类的唯一要求，因为创建DatabaseClient Bean是必需的。这种结构是Spring Data模块的典型结构，因此在尝试其他结构时会感到非常熟悉。此外，我希望一旦有自动配置，就可以删除此手动配置，并且可以通过application.properties单独驱动。

我在此处包括了port属性，但是如果您还没有使用Postgres配置，那么可以依靠默认值5432 。

PostgresqlConnectionFactory定义的四个属性： host ， database ， username和password是使它正常工作的最低要求。少了一点，您将在启动过程中遇到异常。

使用此配置，Spring可以连接到正在运行的Postgres实例。

该示例中最后一个值得注意的信息是@EnableR2dbcRepositories的使用。该注释指示Spring查找扩展Spring的Repository接口的任何存储Repository接口。这用作检测Spring Data存储库的基本接口。我们将在下一部分中对此进行更仔细的研究。从这里获得的主要信息是，您需要使用@EnableR2dbcRepositories批注以充分利用Spring Data的功能。

创建一个Spring数据仓库

如上所述，在本节中，我们将介绍添加Spring Data Repository。这些存储库是Spring Data的一个不错的功能，这意味着您无需写很多额外的代码即可简单地编写查询。不幸的是，至少就目前而言，Spring R2DBC无法以与其他Spring Data模块当前相同的方式来推断查询（我肯定会在某个时候添加它）。这意味着您将需要使用@Query批注并手动编写SQL。让我们来看看：

@Repository
interface PersonRepository : R2dbcRepository<Person, Int> {@Query("SELECT * FROM people WHERE name = $1")fun findAllByName(name: String): Flux<Person>@Query("SELECT * FROM people WHERE age = $1")fun findAllByAge(age: Int): Flux<Person>
}

该接口扩展了R2dbcRepository 。这依次扩展了ReactiveCrudRepository ，然后扩展到Repository 。 ReactiveCrudRepository提供了标准的CRUD功能，据我了解， R2dbcRepository不提供任何额外的功能，而是为更好的情境命名而创建的接口。

R2dbcRepository接受两个通用参数，一个是它作为输入并作为输出产生的实体类。第二个是主键的类型。因此，在这种情况下， Person类由PersonRepository管理（有意义），并且Person内部的主键字段是Int 。

此类中的函数以及ReactiveCrudRepository提供的函数的返回类型为Flux和Mono （此处未显示）。这些是Spring用作默认反应流类型的Project Reactor类型。 Flux代表多个元素的流，而Mono则是单个结果。

最后，正如我在示例之前提到的那样，每个函数都使用@Query注释。语法非常简单，SQL是注释中的字符串。 $1 （用于更多输入的$2 ， $3等）表示输入到函数中的值。完成此操作后，Spring将处理其余部分，并将输入传递到各自的输入参数中，收集结果并将其映射到存储库的指定实体类。

快速浏览实体

在这里不多说，仅显示PersonRepository使用的Person类。

@Table("people")
data class Person(@Id val id: Int? = null,val name: String,val age: Int
)

实际上，这里有一点需要说明。 id已设置为可为空，并提供默认值null以允许Postgres自身生成下一个合适的值。如果这不能为空并且提供了一个id值，那么Spring在保存时实际上将尝试运行更新而不是插入操作。还有其他解决方法，但是我认为这已经足够了。

该实体将映射到下面定义的people表：

CREATE TABLE people (id SERIAL PRIMARY KEY, name VARCHAR NOT NULL, age INTEGER NOT NULL
);

看到一切都在行动

现在让我们看看它实际上在做什么。下面是一些插入一些记录并以几种不同方式检索它们的代码：

@SpringBootApplication
class Application : CommandLineRunner {@Autowiredprivate lateinit var personRepository: PersonRepositoryoverride fun run(vararg args: String?) {personRepository.saveAll(listOf(Person(name = "Dan Newton", age = 25),Person(name = "Laura So", age = 23))).log().subscribe()personRepository.findAll().subscribe { log.info("findAll - $it") }personRepository.findAllById(Mono.just(1)).subscribe { log.info("findAllById - $it") }personRepository.findAllByName("Laura So").subscribe { log.info("findAllByName - $it") }personRepository.findAllByAge(25).subscribe { log.info("findAllByAge - $it") }}
}

关于此代码，我将提到一件事。它很可能在没有实际插入或读取某些记录的情况下执行。但是，当您考虑时。这说得通。反应性应用程序旨在异步执行操作，因此该应用程序已开始处理不同线程中的函数调用。如果不阻塞主线程，这些异步进程可能永远不会完全执行。出于这个原因，此代码中有一些Thread.sleep调用，但我从示例中删除了它们，以使所有内容保持整洁。

运行上面的代码的输出如下所示：

2019-02-11 09:04:52.294  INFO 13226 --- [           main] reactor.Flux.ConcatMap.1                 : onSubscribe(FluxConcatMap.ConcatMapImmediate)
2019-02-11 09:04:52.295  INFO 13226 --- [           main] reactor.Flux.ConcatMap.1                 : request(unbounded)
2019-02-11 09:04:52.572  INFO 13226 --- [actor-tcp-nio-1] reactor.Flux.ConcatMap.1                 : onNext(Person(id=35, name=Dan Newton, age=25))
2019-02-11 09:04:52.591  INFO 13226 --- [actor-tcp-nio-1] reactor.Flux.ConcatMap.1                 : onNext(Person(id=36, name=Laura So, age=23))
2019-02-11 09:04:52.591  INFO 13226 --- [actor-tcp-nio-1] reactor.Flux.ConcatMap.1                 : onComplete()
2019-02-11 09:04:54.472  INFO 13226 --- [actor-tcp-nio-2] com.lankydanblog.tutorial.Application    : findAll - Person(id=35, name=Dan Newton, age=25)
2019-02-11 09:04:54.473  INFO 13226 --- [actor-tcp-nio-2] com.lankydanblog.tutorial.Application    : findAll - Person(id=36, name=Laura So, age=23)
2019-02-11 09:04:54.512  INFO 13226 --- [actor-tcp-nio-4] com.lankydanblog.tutorial.Application    : findAllByName - Person(id=36, name=Laura So, age=23)
2019-02-11 09:04:54.524  INFO 13226 --- [actor-tcp-nio-5] com.lankydanblog.tutorial.Application    : findAllByAge - Person(id=35, name=Dan Newton, age=25)

这里有一些要注意的地方：

onSubscribe和request发生在调用Flux的主线程上。仅saveAll输出此内容，因为它已包含log功能。将其添加到其他调用中将导致记录到主线程的结果相同。
subscription函数中包含的执行和Flux的内部步骤在单独的线程上运行。

这与在实际应用程序中如何使用反应流的真实表示不尽相同，但希望能演示如何使用它们，并对它们的执行方式提供一些见解。

结论

总之，由于R2DBC驱动程序和Spring Data在顶部建立了一层，使所有内容变得更加整洁，因此Reactive Streams进入了某些RDBMS数据库。通过使用Spring Data R2DBC，我们能够创建与数据库的连接并开始查询它，而无需太多代码。尽管Spring已经为我们做了大量工作，但它可能会做更多的事情。当前，它不具有Spring Boot自动配置支持。这有点烦人。但是，我相信有人很快就会做起来并使所有事情变得比现在更好。

这篇文章中使用的代码可以在我的GitHub上找到。