R2DBC 是 "Reactive Relational Database Connectivity"的简称。
R2DBC 是一个 API 规范的倡议,声明对于访问关系型数据库驱动实现了一些响应式的API。
R2DBC的诞生为了非阻塞的应用栈, 使用很少的线程可以处理大量的并发同时减少硬件资源。大量的应用还是使用的关系型数据库,然而很多 NoSQL 数据提供了响应式客户端,并不是所有的项目都适合迁移到 NoSQL。因此,R2DBC 应运而生。
R2DBC 特点
- 对于 R2DBC 的驱动实例,Spring 支持基于Java 的
@Connfiguration
的配置 -
R2dbcEntityTemplate
作为实体绑定的核心操作类 - 整合了Spring Conversion Service 丰富的对象映射
- 基于注解元数据映射关系
- 自动实现 Reposity 接口,包含支持自定义的查询方法
R2DBC 使用
接下来通过一个官方的实例来演示
依赖 pom.xml
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-r2dbc</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<dependency>
<groupId>com.h2database</groupId>
<artifactId>h2</artifactId>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>io.r2dbc</groupId>
<artifactId>r2dbc-h2</artifactId>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<version>3.4.8</version>
<scope>compile</scope>
</dependency>
</dependencies>
定义 Person 实体
public class Person {
private final String id;
private final String name;
private final int age;
public Person(String id, String name, int age) {
this.id = id;
this.name = name;
this.age = age;
}
public String getId() {
return id;
}
public String getName() {
return name;
}
public int getAge() {
return age;
}
@Override
public String toString() {
return "Person [id=" + id + ", name=" + name + ", age=" + age + "]";
}
}
测试
import com.example.springreactivedemo.model.Person;
import io.r2dbc.spi.ConnectionFactories;
import io.r2dbc.spi.ConnectionFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.data.r2dbc.core.R2dbcEntityTemplate;
import reactor.test.StepVerifier;
public class Test1 {
private static final Log log = LogFactory.getLog(Test1.class);
public static void main(String[] args) {
ConnectionFactory connectionFactory = ConnectionFactories.get("r2dbc:h2:mem:///test?options=DB_CLOSE_DELAY=-1;DB_CLOSE_ON_EXIT=FALSE");
R2dbcEntityTemplate template = new R2dbcEntityTemplate(connectionFactory);
template.getDatabaseClient().sql("CREATE TABLE person" +
"(id VARCHAR(255) PRIMARY KEY," +
"name VARCHAR(255)," +
"age INT)")
.fetch()
.rowsUpdated()
.as(StepVerifier::create)
.expectNextCount(1)
.verifyComplete();
template.insert(Person.class)
.using(new Person("joe", "Joe", 34))
.as(StepVerifier::create)
.expectNextCount(1)
.verifyComplete();
template.select(Person.class)
.first()
.doOnNext(it -> log.info(it))
.as(StepVerifier::create)
.expectNextCount(1)
.verifyComplete();
}
}
运行结果:
15:34:40.101 [main] DEBUG org.springframework.r2dbc.core.DefaultDatabaseClient - Executing SQL statement [INSERT INTO person (id, name, age) VALUES ($1, $2, $3)]
15:34:40.103 [main] DEBUG io.r2dbc.h2.client.SessionClient - Request: INSERT INTO person (id, name, age) VALUES ($1, $2, $3) {1: 'joe', 2: 'Joe', 3: 34}
Exception in thread "main" java.lang.AssertionError: expectation "expectNextCount(1)" failed (expected: count = 1; actual: counted = 0; signal: one rror(java.lang.IllegalStateException: Required identifier property not found for class com.example.springreactivedemo.model.Person!))
at reactor.test.MessageFormatter.assertionError(MessageFormatter.java:115)
at reactor.test.MessageFormatter.failPrefix(MessageFormatter.java:104)
at reactor.test.MessageFormatter.fail(MessageFormatter.java:73)
at reactor.test.MessageFormatter.failOptional(MessageFormatter.java:88)
通过运行官方文档的实例出现了以上的报错信息,我们来看一下错误信息,Required identifier property not found for class com.example.springreactivedemo.model.Person!
, Person 要求一个唯一属性,将 Person 类 id 增加了注解@Id
, 再次执行成功,日志如下。
15:38:06.806 [main] DEBUG org.springframework.r2dbc.core.DefaultDatabaseClient - Executing SQL statement [INSERT INTO person (id, name, age) VALUES ($1, $2, $3)]
15:38:06.808 [main] DEBUG io.r2dbc.h2.client.SessionClient - Request: INSERT INTO person (id, name, age) VALUES ($1, $2, $3) {1: 'joe', 2: 'Joe', 3: 34}
15:38:06.847 [main] DEBUG io.r2dbc.h2.client.SessionClient - Request: CALL H2VERSION()
15:38:06.847 [main] DEBUG io.r2dbc.h2.client.SessionClient - Response: org.h2.result.LocalResultImpl@72c927f1 columns: 1 rows: 1 pos: -1
15:38:06.847 [main] DEBUG org.springframework.r2dbc.core.DefaultDatabaseClient - Executing SQL statement [SELECT person.* FROM person LIMIT 1]
15:38:06.853 [main] DEBUG io.r2dbc.h2.client.SessionClient - Request: SELECT person.* FROM person LIMIT 1
15:38:06.854 [main] DEBUG io.r2dbc.h2.client.SessionClient - Response: org.h2.result.LocalResultImpl@1c32886a columns: 3 rows: 1 pos: -1
15:38:06.876 [main] INFO com.example.springreactivedemo.client.Test1 - Person [id=joe, name=Joe, age=34]
Spring Boot 整合 R2DBC 流程
1. 创建 ConnectionFactory 和 R2dbcEntityTemplate
@Configuration
public class ApplicationConfiguration extends AbstractR2dbcConfiguration {
@Override
@Bean
public ConnectionFactory connectionFactory() {
return new H2ConnectionFactory(H2ConnectionConfiguration.builder()
.url("r2dbc:h2:mem:///test?options=DB_CLOSE_DELAY=-1;DB_CLOSE_ON_EXIT=FALSE")
.build());
}
@Bean
public R2dbcEntityTemplate r2dbcEntityTemplate() {
return new R2dbcEntityTemplate(connectionFactory());
}
}
2. CRUD 使用
@RestController
@Slf4j
public class PersonController {
private R2dbcEntityTemplate r2dbcEntityTemplate;
public PersonController(R2dbcEntityTemplate r2dbcEntityTemplate) {
this.r2dbcEntityTemplate = r2dbcEntityTemplate;
// 创建表结构
r2dbcEntityTemplate.getDatabaseClient().sql("drop table person if exists; CREATE TABLE person" +
"(id VARCHAR(255) PRIMARY KEY," +
"name VARCHAR(255)," +
"age INT)").fetch().rowsUpdated().subscribe();
}
@PostMapping("/save")
public Mono<Person> insert(@RequestBody Person person) {
return r2dbcEntityTemplate.insert(Person.class).using(person);
}
@GetMapping("/list")
public Flux<Person> list() {
return r2dbcEntityTemplate.select(Query.empty(), Person.class);
}
@PutMapping("/update")
public void update(@RequestBody Person person) {
r2dbcEntityTemplate.update(Person.class)
.inTable("person") //可以指定 table
.matching(Query.query(Criteria.where("id").is(person.getId())))
.apply(Update.update("name", person.getName())).subscribe();
}
@DeleteMapping("/delete")
public Mono<Integer> delete(@RequestBody Person person) {
return r2dbcEntityTemplate.delete(Person.class).matching(Query.query(Criteria.where("id").is(person.getId()))).all();
}
}
总结
Spring Boot 整合 R2DBC 简单实例使用完成,需要注意的是响应式编程和之前命令式编程有很大的区别,在写update方法中,在 R2dbcEntityTemplate 执行 update 最后没有调用 subscribe,导致 update 方法没有执行,在响应式编程中定义的方法流程需要通过触发才会执行。代码里面使用 subscribe 的方式来触发调用,还可以将执行 update返回的值 Mono<Integer>
,这样也会触发执行。
参考:
https://docs.spring.io/spring-data/r2dbc/docs/current/reference/html/#introduction