Reactive Spring實戰 — 響應式MySql交互
- 2021 年 7 月 12 日
- 筆記
- Reactive Spring實戰
本文與大家探討Spring中如何實現MySql響應式交互。
Spring Data R2DBC項目是Spring提供的資料庫響應式編程框架。
R2DBC是Reactive Relational Database Connectivity的首字母縮寫詞。 R2DBC是一個API規範倡議，它聲明了一個響應式API，由驅動程式供應商實現，並以響應式編程的方式訪問他們的關係資料庫。
實現資料庫的響應式編程並不是容易的，傳統的JDBC協議是一個完全阻塞的 API，所以響應式編程對JDBC協議可以說是一種「顛覆」了。
這裡再強調一次響應式編程，響應式編程是一種非阻塞非同步的編程模式，而Spring響應式編程提供了一種友好、直觀、易於理解的編碼模式處理非同步結果（可參考前面的文章）。
也就是說，應用發送SQL給資料庫後，應用執行緒不需要阻塞等待資料庫返回結果，而是直接返回處理其他任務，等到資料庫SQL處理完成後，再由Spring調用執行緒處理結果。
到目前，Spring Data R2DBC項目支援以下資料庫：
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)
下面基於MySql，介紹一下Spring Data R2DBC使用方式。
引入依賴
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-r2dbc</artifactId>
</dependency>
<dependency>
<groupId>dev.miku</groupId>
<artifactId>r2dbc-mysql</artifactId>
<version>0.8.2.RELEASE</version>
</dependency>
配置文件
spring.r2dbc.url=r2dbcs:mysql://127.0.0.1:3306/bin-springreactive?useSSL=false
spring.r2dbc.username=...
spring.r2dbc.password=...
Spring Data R2DBC可以與Spring Data JPA結合使用，其實R2DBC與原來的JPA使用方式差別不大，使用非常簡單。
只是Spring Data JPA中方法返回的是真實的值，而R2DBC中，返回的是數據流Mono，Flux。
簡單介紹一個Spring Data JPA。Spring Data JPA是Spring基於ORM框架、JPA規範的基礎上封裝的一套 JPA (Java Persistence API) 應用框架，簡單說，就是類似Mybatis，Hibernate的框架（Spring Data JPA底層通過Hibernate操作資料庫）。
Repository是Spring Data R2DBC中的重要概念，封裝了對一個實體的操作，相當於一個dao（Data Access Object，數據訪問對象）。
假如應用中有一個實體DeliveryCompany，對應表delivery_company。
實體定義如下：
public class DeliveryCompany {
@Id
private long id;
private String name;
private String label;
private Integer level;
...
}
@Id註解標誌了id屬性。
下面我們定義一個DeliveryCompanyRepository介面，繼承與R2dbcRepository。
@Repository
public interface DeliveryCompanyRepository extends R2dbcRepository<DeliveryCompany,Long> {
...
}
R2dbcRepository是Spring實現的介面，該介面繼承與ReactiveCrudRepository，ReactiveCrudRepository介面提供了增刪改查的模板方法。
public interface ReactiveCrudRepository<T, ID> extends Repository<T, ID> {
<S extends T> Mono<S> save(S var1);
<S extends T> Flux<S> saveAll(Iterable<S> var1);
<S extends T> Flux<S> saveAll(Publisher<S> var1);
Mono<T> findById(ID var1);
Mono<T> findById(Publisher<ID> var1);
...
}
注意這裡的返回結果，是Mono、Flux等非同步結果，這就是響應式交互與非響應式交互的最大區別。
如果要自定義操作，有以下方式
（1） 通過方法名定義
只要我們按規則定義方法名，Spring就會為我們生成SQL。
// 按名稱查找
Flux<DeliveryCompany> findByName(String name);
// 查找給定範圍內的
Flux<DeliveryCompany> findByIdGreaterThan(Long startId);
// 查找大於給定id的數據
Flux<DeliveryCompany> findByIdGreaterThan(Long startId);
// 查詢名稱以給定字元串開頭的數據
Flux<DeliveryCompany> findByNameStartingWith(String start);
// 分頁
Flux<DeliveryCompany> findByIdGreaterThanEqual(Long startId, Pageable pageable);
注意，上面方法名需要按規範定義
findByName -> findBy<fieldName>
findByIdGreaterThan -> findBy<fieldName>GreaterThan
Spring會為我們生成對應的SQL，非常方便。這種方法可以滿足多數簡單的查詢。
對應的還有刪除操作
Mono<Integer> deleteByName(String name);
詳細的方法命名規則，則參考官方文檔。
（2）手動編寫SQL
對於複雜的SQL，開發人員也可以手寫SQL，
@Query("select id,name from delivery_company where id in (:ids)")
Flux<DeliveryCompany> findByIds2(List<Long> ids);
@Query("select id,name from delivery_company where name = :name")
Flux<DeliveryCompany> findByName2(String name);
@Modifying
@Query("update delivery_company set name = :name where id = :id")
Mono<DeliveryCompany> update2(@Param("id") long id, @Param("name") String name);
可以看到，編寫SQL也非常簡單，對於集合參數支援非常好。
目前未發現使用JPQL（Java Persistence Query Language）的方式，不過使用原生的SQL是沒有問題的。
如果大家使用過Mybatis，應該會用過以下判斷參數非空的做法
<select id="findByName2"
resultType="DeliveryCompany">
SELECT * FROM delivery_company
WHERE name = #{name}
<if test="label != null">
AND label like #{label}
</if>
</select>
可惜在JPA中非找到支援的方法，如果有同學知道，請不吝指教。
(3) 使用R2dbcEntityTemplate
另外，可以使用R2dbcEntityTemplate自動生成SQL
@Autowired
private R2dbcEntityTemplate template;
public Flux<DeliveryCompany> getByName3(String name) {
return template
.select(DeliveryCompany.class)
.from("delivery_company")
.matching(Query.query(Criteria.where("name").is(name))).all();
// Criteria.where("name").is(name).and
}
public Mono<Integer> update3(DeliveryCompany company) {
return template
.update(DeliveryCompany.class)
.inTable("delivery_company")
.matching(Query.query(Criteria.where("id").is(company.getId())))
.apply(Update.update("name", company.getName()));
}
這種方式可以實現判斷參數非空查詢，不過使用起來較為繁瑣（我們也可以對其進行一定的封裝以方便我們使用）。
(4)Spring Data R2DBC中同樣支援Querydsl，
我們定義的Repository可以繼承於ReactiveQuerydslPredicateExecutor
public interface ReactiveQuerydslPredicateExecutor<T> {
Mono<T> findOne(Predicate var1);
Flux<T> findAll(Predicate var1);
Flux<T> findAll(Predicate var1, Sort var2);
Flux<T> findAll(Predicate var1, OrderSpecifier... var2);
Flux<T> findAll(OrderSpecifier... var1);
Mono<Long> count(Predicate var1);
Mono<Boolean> exists(Predicate var1);
}
Spring Data R2DBC中同樣支援@QuerydslPredicate註解，這裡不再深入。
Spring Data R2DBC支援事務，使用方法很簡單，在業務方法添加@Transactional即可
@Transactional
public Flux<DeliveryCompany> save(List<DeliveryCompany> companyList) {
Flux<DeliveryCompany> result = Flux.just();
for (DeliveryCompany deliveryCompany : companyList) {
result = result.concat(result, repository.save(deliveryCompany));
}
return result;
}
為了展示事務的使用，這裡沒有調用Repository的saveAll方法，而是循環插入數據並返回最後的結果。
注意，最後的結果Flux、Mono一定要作為方法返回值，因為響應式編程的異常資訊保存在這些結果中（而不是在方法調用時拋出），所以這些結果必須作為方法返回值，否則Spring無法知道方法是否報錯，也就無法回退事務。
Spring Data R2DBC基本與Spring Data JPA的使用相同，所以本篇文章主要還是對Spring Data JPA使用方式的介紹。
我之前並沒有使用過Spring Data JPA，本篇文章主要還是入門介紹，還有很多東西沒有涉及，如id生成，多表查詢等，這裡不再一一介紹。
官方文檔：//docs.spring.io/spring-data/r2dbc/docs/1.3.2/reference/html/
文章完整程式碼：//gitee.com/binecy/bin-springreactive/tree/master/delivery-service
如果您覺得本文不錯，歡迎關注我的微信公眾號，系列文章持續更新中。您的關注是我堅持的動力！
