以前咱们提到过，对于底层的数据源来讲，MongoDB, Redis, 和 Cassandra 能够直接以reactive的方式支持Spring Data。而其余不少关系型数据库好比Postgres, Microsoft SQL Server, MySQL, H2 和 Google Spanner 则能够经过使用R2DBC 来实现对reactive的支持。java
今天咱们就来具体讲解一下R2DBC的使用。react
以前咱们介绍了Reactor还有基于其之上的Spring WebFlux框架。包括vert.x，rxjava等等reactive技术。咱们实际上在应用层已经有不少优秀的响应式处理框架。git
可是有一个问题就是全部的框架都须要获取底层的数据，而基本上关系型数据库的底层读写都仍是同步的。github
为了解决这个问题，出现了两个标准，一个是oracle提出的 ADBC (Asynchronous Database Access API)，另外一个就是Pivotal提出的R2DBC (Reactive Relational Database Connectivity)。web
R2DBC是基于Reactive Streams标准来设计的。经过使用R2DBC，你能够使用reactive API来操做数据。spring
同时R2DBC只是一个开放的标准，而各个具体的数据库链接实现，须要实现这个标准。shell
今天咱们以r2dbc-h2为例，讲解一下r2dbc在Spring webFlux中的使用。数据库
咱们须要引入r2dbc-spi和r2dbc-h2两个库，其中r2dbc-spi是接口，而r2dbc-h2是具体的实现。oracle
同时咱们使用了Spring webflux,因此还须要引入spring-boot-starter-webflux。app
具体的依赖以下：
<!-- 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咱们能够建立到数据库的链接。
先看一下数据库的配置文件，为了方便起见，这里咱们使用的是内存数据库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，还能够加上这两个配置。
这里，咱们建立一个简单的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。
为了防止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应用来对外提供服务：
@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的博客
欢迎关注个人公众号:「程序那些事」最通俗的解读，最深入的干货，最简洁的教程，众多你不知道的小技巧等你来发现！