響應式關係資料庫處理R2DBC
簡介
之前我們提到過，對於底層的數據源來說，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程式那些事
本文鏈接：//www.flydean.com/r2dbc-introduce/
本文來源：flydean的部落格
歡迎關注我的公眾號:「程式那些事」最通俗的解讀，最深刻的乾貨，最簡潔的教程，眾多你不知道的小技巧等你來發現！
