简介
之前咱们提到过，对于底层的数据源来说，MongoDB, Redis, 和 Cassandra 能够间接以reactive的形式反对Spring Data。而其余很多关系型数据库比方Postgres, Microsoft SQL Server, MySQL, H2 和 Google Spanner 则能够通过应用R2DBC 来实现对reactive的反对。
明天咱们就来具体解说一下R2DBC的应用。
R2DBC介绍
之前咱们介绍了Reactor还有基于其之上的Spring WebFlux框架。包含vert.x，rxjava等等reactive技术。咱们实际上在应用层曾经有很多优良的响应式解决框架。
然而有一个问题就是所有的框架都须要获取底层的数据，而基本上关系型数据库的底层读写都还是同步的。
为了解决这个问题，呈现了两个规范，一个是oracle提出的 ADBC (Asynchronous Database Access API)，另一个就是Pivotal提出的R2DBC (Reactive Relational Database Connectivity)。
R2DBC是基于Reactive Streams规范来设计的。通过应用R2DBC，你能够应用reactive API来操作数据。
同时R2DBC只是一个凋谢的规范，而各个具体的数据库连贯实现，须要实现这个规范。
明天咱们以r2dbc-h2为例，解说一下r2dbc在Spring webFlux中的应用。
我的项目依赖
咱们须要引入r2dbc-spi和r2dbc-h2两个库，其中r2dbc-spi是接口，而r2dbc-h2是具体的实现。
同时咱们应用了Spring webflux,所以还须要引入spring-boot-starter-webflux。
具体的依赖如下：
<!-- R2DBC H2 Driver -->
<dependency>
<groupId>io.r2dbc</groupId>
<artifactId>r2dbc-h2</artifactId>
<version>${r2dbc-h2.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>创立ConnectionFactory
ConnectionFactory是数据库连贯的一个具体实现，通过ConnectionFactory咱们能够创立到数据库的连贯。
先看一下数据库的配置文件，为了不便起见，这里咱们应用的是内存数据库H2 ：
r2dbc.url=r2dbc:h2:mem://./r2dbc
r2dbc.user=sa
r2dbc.password=password第一个url指定的是数据库的连贯形式，上面两个是数据库的用户名和明码。
接下来咱们看一下，怎么通过这些属性来创立ConnectionFactory：
@Bean
public ConnectionFactory connectionFactory() {
ConnectionFactoryOptions baseOptions = ConnectionFactoryOptions.parse(url);
ConnectionFactoryOptions.Builder ob = ConnectionFactoryOptions.builder().from(baseOptions);
if (!StringUtil.isNullOrEmpty(user)) {
ob = ob.option(USER, user);
}
if (!StringUtil.isNullOrEmpty(password)) {
ob = ob.option(PASSWORD, password);
}
return ConnectionFactories.get(ob.build());
}通过url能够parse失去ConnectionFactoryOptions。而后通过ConnectionFactories的get办法创立ConnectionFactory。
如果咱们设置了USER或者PASSWORD，还能够加上这两个配置。
创立Entity Bean
这里，咱们创立一个简略的User对象：
@Data
@NoArgsConstructor
@AllArgsConstructor
public class Users {
private Long id;
private String firstname;
private String lastname;
}初始化数据库
尽管H5有很多更加简略的形式来初始化数据库，比方间接读取SQL文件，这里为了阐明R2DBC的应用，咱们应用手动的形式来创立：
@Bean
public CommandLineRunner initDatabase(ConnectionFactory cf) {
return (args) ->
Flux.from(cf.create())
.flatMap(c ->
Flux.from(c.createBatch()
.add("drop table if exists Users")
.add("create table Users(" +
"id IDENTITY(1,1)," +
"firstname varchar(80) not null," +
"lastname varchar(80) not null)")
.add("insert into Users(firstname,lastname)" +
"values('flydean','ma')")
.add("insert into Users(firstname,lastname)" +
"values('jacken','yu')")
.execute())
.doFinally((st) -> c.close())
)
.log()
.blockLast();
}下面的代码中，咱们应用c.createBatch()来向数据库插入一些数据。
除了createBatch,还能够应用create来创立单个的执行语句。
获取所有的用户
在Dao中，咱们提供了一个findAll的办法：
public Flux<Users> findAll() {
return Mono.from(connectionFactory.create())
.flatMap((c) -> Mono.from(c.createStatement("select id,firstname,lastname from users")
.execute())
.doFinally((st) -> close(c)))
.flatMapMany(result -> Flux.from(result.map((row, meta) -> {
Users acc = new Users();
acc.setId(row.get("id", Long.class));
acc.setFirstname(row.get("firstname", String.class));
acc.setLastname(row.get("lastname", String.class));
return acc;
})));
}简略解释一下下面的应用。
因为是一个findAll办法，咱们须要找出所有的用户信息。所以咱们返回的是一个Flux而不是一个Mono。
怎么从Mono转换成为一个Flux呢？
这里咱们应用的是flatMapMany，将select进去的后果，分成一行一行的，最初转换成为Flux。
Prepare Statement
为了避免SQL注入，咱们须要在SQL中应用Prepare statement:
public Mono<Users> findById(long id) {
return Mono.from(connectionFactory.create())
.flatMap(c -> Mono.from(c.createStatement("select id,firstname,lastname from Users where id = $1")
.bind("$1", id)
.execute())
.doFinally((st) -> close(c)))
.map(result -> result.map((row, meta) ->
new Users(row.get("id", Long.class),
row.get("firstname", String.class),
row.get("lastname", String.class))))
.flatMap( p -> Mono.from(p));
}看下咱们是怎么在R2DBC中应用prepare statement的。
事务处理
接下来咱们看一下怎么在R2DBC中应用事务：
public Mono<Users> createAccount(Users account) {
return Mono.from(connectionFactory.create())
.flatMap(c -> Mono.from(c.beginTransaction())
.then(Mono.from(c.createStatement("insert into Users(firstname,lastname) values($1,$2)")
.bind("$1", account.getFirstname())
.bind("$2", account.getLastname())
.returnGeneratedValues("id")
.execute()))
.map(result -> result.map((row, meta) ->
new Users(row.get("id", Long.class),
account.getFirstname(),
account.getLastname())))
.flatMap(pub -> Mono.from(pub))
.delayUntil(r -> c.commitTransaction())
.doFinally((st) -> c.close()));
}下面的代码中，咱们应用了事务，具体的代码有两局部：
c -> Mono.from(c.beginTransaction())
.delayUntil(r -> c.commitTransaction())开启是的时候须要应用beginTransaction，前面提交就须要调用commitTransaction。
WebFlux应用
最初，咱们须要创立WebFlux利用来对外提供服务：
@GetMapping("/users/{id}")
public Mono<ResponseEntity<Users>> getUsers(@PathVariable("id") Long id) {
return usersDao.findById(id)
.map(acc -> new ResponseEntity<>(acc, HttpStatus.OK))
.switchIfEmpty(Mono.just(new ResponseEntity<>(null, HttpStatus.NOT_FOUND)));
}
@GetMapping("/users")
public Flux<Users> getAllAccounts() {
return usersDao.findAll();
}
@PostMapping("/createUser")
public Mono<ResponseEntity<Users>> createUser(@RequestBody Users user) {
return usersDao.createAccount(user)
.map(acc -> new ResponseEntity<>(acc, HttpStatus.CREATED))
.log();
}执行成果
最初，咱们运行一下代码，执行下users:
curl "localhost:8080/users"
[{"id":1,"firstname":"flydean","lastname":"ma"},{"id":2,"firstname":"jacken","lastname":"yu"}]% 完满，试验胜利。
本文的代码：webflux-with-r2dbc
本文作者：flydean程序那些事
本文链接：http://www.flydean.com/r2dbc-introduce/
本文起源：flydean的博客
欢送关注我的公众号:「程序那些事」最艰深的解读，最粗浅的干货，最简洁的教程，泛滥你不晓得的小技巧等你来发现！
发表回复