Spring WebFlux整合R2DBC實(shí)現(xiàn)數(shù)據(jù)庫操作（反應(yīng)式編程系列）
環(huán)境：Springboot2.4.12
R2DBC簡介
Spring data R2DBC是更大的Spring data 系列的一部分，它使得實(shí)現(xiàn)基于R2DBC的存儲庫變得容易。R2DBC代表反應(yīng)式關(guān)系數(shù)據(jù)庫連接，這是一種使用反應(yīng)式驅(qū)動程序集成SQL數(shù)據(jù)庫的規(guī)范。Spring Data R2DBC使用屬性的Spring抽象和Repository支持應(yīng)用于R2DBC。它使得在反應(yīng)式應(yīng)用程序堆棧中使用關(guān)系數(shù)據(jù)訪問技術(shù)構(gòu)建Spring驅(qū)動的應(yīng)用程序變得更加容易。
Spring Data R2DBC的目標(biāo)是在概念上變得簡單。為了實(shí)現(xiàn)這一點(diǎn)，它不提供緩存、延遲加載、寫后處理或ORM框架的許多其他特性。這使得Spring Data R2DBC成為一個(gè)簡單、有限、固執(zhí)己見的對象映射器。
Spring Data R2DBC允許一種 functional 方法與數(shù)據(jù)庫交互，提供R2dbcEntityTemplate作為應(yīng)用程序的入口點(diǎn)。
首先選擇數(shù)據(jù)庫驅(qū)動程序并創(chuàng)建R2dbcEntityTemplate實(shí)例：
- H2 (io.r2dbc:r2dbc-h2)
- MariaDB (org.mariadb:r2dbc-mariadb)
- Microsoft SQL Server (io.r2dbc:r2dbc-mssql)
- MySQL (dev.miku:r2dbc-mysql)
- jasync-sql MySQL (com.github.jasync-sql:jasync-r2dbc-mysql)
- Postgres (io.r2dbc:r2dbc-postgresql)
- Oracle (com.oracle.database.r2dbc:oracle-r2dbc)
WebFlux介紹
Spring框架中包含原始web框架Spring Web MVC是專門為ServletAPI和Servlet容器構(gòu)建的。反應(yīng)式堆棧web框架Spring Web Flux是后來在5.0版中添加。它是完全非阻塞的，支持反應(yīng)流背壓（由消費(fèi)者控制生產(chǎn)者的速度），并在Netty、Undertow和Servlet 3.1+容器等服務(wù)器上運(yùn)行。
這兩個(gè)web框架都反映了它們的源模塊（Spring Web MVC和Spring Web Flux）的名稱，并在Spring框架中共存。每個(gè)模塊都是可選的。應(yīng)用程序可以使用一個(gè)或另一個(gè)模塊，在某些情況下，可以同時(shí)使用這兩個(gè)模塊?—?例如，帶有反應(yīng)式WebClient的Spring MVC控制器。
依賴管理
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-r2dbc</artifactId>
</dependency>
<dependency>
<groupId>dev.miku</groupId>
<artifactId>r2dbc-mysql</artifactId>
</dependency>
</dependencies>應(yīng)用配置
spring:
r2dbc:
url: r2dbc:mysql://localhost:3306/reactive_db
username: root
password: 123123
pool:
initialSize: 100
maxSize: 100
---
logging:
level:
org.springframework.r2dbc: DEBUG #輸出執(zhí)行的sql關(guān)于MySQL的R2DBC詳細(xì)配置查看：
https://github.com/mirromutth/r2dbc-mysql
實(shí)體&Service 基本的CURD操作
@Table("T_USERS")
public class Users {
@Id
private Long id ;
private String name ;
private String sex ;
private Integer age ;
}
Service
@Resource
private R2dbcEntityTemplate template ;
@Transactional
public Mono<Users> insertByTemplate(Users users) {
return template.insert(users) ;
}
public Mono<Integer> removeByTemplate(Long id) {
Query query = Query.query(Criteria.where("id").is(id)) ;
return template.delete(query, Users.class) ;
}
public Mono<Integer> updateByTemplate(Users users) {
CriteriaDefinition criteria = Criteria.where("id").is(users.getId()) ;
Query query = Query.query(criteria) ;
Update update = Update.update("name", users.getName()) ;
return template.update(query, update, Users.class) ;
}
public Mono<Users> selectByTemplate(Long id) {
Query query = Query.query(Criteria.where("id").is(id)) ;
return template.select(query, Users.class).single() ;
}
public Flux<Users> selectByTemplate(Integer page, Integer size) {
Query query = Query.empty().offset((page - 1) * size).limit(size) ;
return template.select(query, Users.class) ;
}
public Mono<Long> selecyByTemplateCount() {
return template.select(Users.class).count() ;
}
public Mono<ResponseEntity<List<Users>>> selectByTemplatePager(Integer page, Integer size) {
Mono<List<Users>> datas = this.selectByTemplate(page, size).collectList() ;
Mono<Long> count = this.selecyByTemplateCount() ;
return datas.zipWith(count, (list, c) -> {
return ResponseEntity.ok().header("count", c + "").header("page", page + "").header("size", size + "").body(list) ;
}) ;
}Controller
@Resource
private UsersService us ;
@PostMapping("/insert")
public Mono<Users> insertByTemplate(@RequestBody Users users) {
return us.insertByTemplate(users) ;
}
@GetMapping("/remove/{id}")
public Mono<Integer> removeByTemplate(@PathVariable("id")Long id) {
return us.removeByTemplate(id) ;
}
@PostMapping("/update")
public Mono<Integer> updateByTemplate(@RequestBody Users users) {
return us.updateByTemplate(users) ;
}
@GetMapping("/query/{id}")
public Mono<Users> selectByTemplate(@PathVariable("id") Long id) {
return us.selectByTemplate(id).single() ;
}
@GetMapping("/pager")
public Mono<ResponseEntity<List<Users>>> selectByTemplate(Integer page, Integer size) {
return us.selectByTemplatePager(page, size) ;
}
@GetMapping("/count")
public Mono<Long> selecyByTemplateCount() {
return us.selecyByTemplateCount() ;
}R2DBC Repository
通過繼承ReactiveCrudRepository或者是ReactiveSortingRepository。Repository支持的方法查詢?nèi)缦卤硭荆?/p>
圖片
Repository修改操作：
interface ReactivePersonRepository extends ReactiveSortingRepository<Person, String> {
Mono<Integer> deleteByLastname(String lastname);
Mono<Void> deletePersonByLastname(String lastname);
Mono<Boolean> deletePersonByLastname(String lastname);
}自定義修改操作：
@Modifying
@Query("UPDATE person SET firstname = :firstname where lastname = :lastname")
Mono<Integer> setFixedFirstnameFor(String firstname, String lastname);支持樂觀鎖：
@Version注釋提供了與R2DBC上下文中JPA類似的語法，并確保更新僅應(yīng)用于具有匹配版本的行。因此，version屬性的實(shí)際值被添加到update查詢中，如果另一個(gè)操作同時(shí)更改了行，則更新不會產(chǎn)生任何影響。在這種情況下，將拋出OptimisticLockingFailureException。以下示例顯示了這些功能：
@Table
public class Person {
@Id Long id;
String firstname;
String lastname;
@Version
Long version;
}如下示例演示了樂觀鎖異常的觸發(fā)：
R2dbcEntityTemplate template = …;
// 1. 初始插入數(shù)據(jù) 此時(shí)version = 0
Mono<Person> daenerys = template.insert(new Person("Daenerys"));
// 2. 加載剛剛插入的數(shù)據(jù)，此時(shí)加載的version = 0
Person other = template.select(Person.class) .matching(query(where("id").is(daenerys.getId()))).first().block();
// 3. 更新數(shù)據(jù)，此處更新后該條數(shù)據(jù)的version = 1
daenerys.setLastname("Targaryen");
template.update(daenerys);
// 4. 更新數(shù)據(jù)，由于other中的version = 0 ；而數(shù)據(jù)庫已經(jīng)是1了，所以這里會觸發(fā)OptimisticLockingFailureException異常
template.update(other).subscribe();
