响应式编程实战(12)-12

1 前言

无论是传统软件还是互联网应用，对于任何一个系统而言，数据的存储和访问都是不可缺少的。而数据访问层的构建可能会涉及多种不同形式的数据存储媒介，包括传统的关系型数据库，也包含各种 NoSQL。今天这一讲我们先讨论响应式数据访问的模型，以及 Spring 框架所提供的支持。

本文先引出全栈式响应式编程这一设计理念，在这一设计理念之下，需要解决的一大问题就是如何构建响应式数据访问。幸好，Spring Data 为我们提供了一套完整的解决方案，其中 Spring Data Reactive 组件是你需要重点掌握的。

2 全栈式响应式编程

讨论如何构建响应式数据访问组件，构建响应式数据访问组件的目的来自一个核心概念，即全栈式响应式编程。

所谓全栈式响应式编程，指的是响应式开发方式的有效性取决于整个请求链路的各个环节是否都采用了响应式编程模型，正如下图所展示的，全栈式响应式编程示意图：

响应式编程实战(12)-12_Data

如某环节或步骤不是响应式的，就会出现同步阻塞，从而导致背压机制无法生效。如果某一层组件（例如数据访问层）无法采用响应式编程模型，那么响应式编程的概念对于整个请求链路的其他层而言就没有意义。

常见的 Web 服务架构中，最典型的非响应式场景就是数据访问层中使用关系型数据库，因为传统的关系型数据库都基于非响应式的数据访问机制。

这种全栈式的响应式技术栈就需要数据访问层返回的是 Mono/Flux 流，而不是传统的实体对象。借助 Spring 家族的 Spring Data 组件，我们可以实现这一目标。

3 Spring Data 与数据访问抽象

Spring 家族中专门用于数据访问的开源框架，其核心理念是支持对所有存储媒介进行资源配置从而实现数据访问。

数据访问需要完成领域对象与存储数据之间的映射，并对外提供访问入口，Spring Data 基于 Repository 架构模式抽象出一套实现该模式的统一数据访问方式。

Spring Data 对数据访问过程的抽象主要体现在：

提供了一套 Repository 接口定义及实现
实现了各种多样化的查询支持

3.1 Repository

Repository 接口是 Spring Data 中对数据访问的最高层抽象，接口定义如下所示。

public interface Repository<T, ID> {
 }

空接口，通过泛型指定了领域实体对象的类型和 ID。Spring Data 中，存在一大批 Repository 接口的子接口和实现类，其中 CrudRepository 接口是对 Repository 接口的最常见扩展，其添加了对领域实体的 CRUD 操作功能：

public interface CrudRepository<T, ID> extends Repository<T, ID> {
   <S extends T> S save(S entity);=
   <S extends T> Iterable<S> saveAll(Iterable<S> entities);=
   Optional<T> findById(ID id);
   boolean existsById(ID id);
   Iterable<T> findAll();
   Iterable<T> findAllById(Iterable<ID> ids);
   long count();
   void deleteById(ID id);
   void delete(T entity);
   void deleteAll(Iterable<? extends T> entities);
   void deleteAll();
 }

这些方法都是自解释的，我们可以看到 CrudRepository 接口提供了保存单个实体、保存集合、根据 Id 查找实体、根据 Id 判断实体是否存在、查询所有实体、查询实体数量、根据 Id 删除实体、删除一个实体的集合以及删除所有实体等常见操作。

3.2 多样化查询

在日常开发过程中，对数据的查询操作需求远高于新增、删除和修改操作，所以在 Spring Data 中，除了对领域对象提供默认的 CRUD 操作之外，重点对查询场景做了高度抽象，最典型的就是提供:

@Query 注解
方法名衍生查询机制

通过 @Query 注解直接在代码中嵌入查询语句和条件，从而提供类似 ORM 框架所具有的强大功能：

public interface AccountRepository extends JpaRepository<Account, Long> {
    
   @Query("select a from Account a where a.accountName= ?1") 
   Account findByUserName(String accountName);   
 }

这里的 @Query 注解中使用的就是类似 SQL 语句的语法，但却能自动完成领域对象 Account 与数据库数据之间的相互映射。虽然上述代码示例给出的是针对关系型数据访问的 JpaRepository，但在 Spring Data 中存在一批 @Query 注解，分别用来针对不同的持久化媒介。

如MongoDB 中同样存在一个 @Query 注解，但该注解位于org.springframework.data.mongodb.repository 包中，定义如下。

package org.springframework.data.mongodb.repository;
  
 public @interface Query {
     String value() default "";
     String fields() default "";
     boolean count() default false;
     boolean exists() default false;
     boolean delete() default false;
 }

MongoDB 中 @Query 注解的 value 值是一串 JSON 字符串，用于指定需要查询的对象条件，我会在下一讲为你介绍具体的使用方法。

方法名衍生查询也是 Spring Data 的查询特色之一，通过在方法命名上直接使用查询字段和参数，Spring Data 就能自动识别相应的查询条件并组装对应的查询语句。

示例：

public interface AccountRepository extends JpaRepository<Account, Long> {
    
        List<Account> findByFirstNameAndLastName(String firstName, String 
   lastName);
 }

上面的例子通过 findByFirstNameAndLastname 这样符合普通语义的方法名，并在参数列表中按照方法名中参数的顺序和名称（即第一个参数是 fistName，第二个参数 lastName）传入相应的参数，Spring Data 就能自动组装 SQL 语句从而实现衍生查询。

响应式数据访问模型

从 Spring Boot 2 开始，针对支持响应式访问的各种数据库，Spring Data 提供了响应式版本的 Repository 支持。

Spring Data Reactive 抽象

理想情况下，在查询数据库时，我们希望以与“12 | WebClient：如何实现非阻塞式的跨服务远程调用”中介绍的 WebClient 类似的方式使用数据存储库。实际上，Spring Data Commons 模块为 ReactiveCrudRepository 接口提供了这样的契约。

与 CrudRepository 接口类似，ReactiveCrudRepository 接口同样继承自 Repository 接口，提供了针对数据流的 CRUD 操作。ReactiveCrudRepository 接口定义如下所示。

public interface ReactiveCrudRepository<T, ID> extends 
   Repository<T, ID> {
     <S extends T> Mono<S> save(S entity);
     <S extends T> Flux<S> saveAll(Iterable<S> entities);
     <S extends T> Flux<S> saveAll(Publisher<S> entityStream);
     Mono<T> findById(ID id);
     Mono<T> findById(Publisher<ID> id);
     Mono<Boolean> existsById(ID id);
     Mono<Boolean> existsById(Publisher<ID> id);
     Flux<T> findAll();
     Flux<T> findAllById(Iterable<ID> ids);
     Flux<T> findAllById(Publisher<ID> idStream);
     Mono<Long> count();
     Mono<Void> deleteById(ID id);
     Mono<Void> deleteById(Publisher<ID> id);
     Mono<Void> delete(T entity);
     Mono<Void> deleteAll(Iterable<? extends T> entities);
     Mono<Void> deleteAll(Publisher<? extends T> entityStream);
     Mono<Void> deleteAll();
 }

ReactiveCrudRepository 中所有方法的返回值都是 Mono/Flux，满足全栈响应式编程模型的需求。

ReactiveSortingRepository 接口进一步对 ReactiveCrudRepository 接口做了扩展，添加了排序功能，ReactiveSortingRepository 接口定义如下。

public interface ReactiveSortingRepository<T, ID> extends 
   ReactiveCrudRepository<T, ID> {
  
         Flux<T> findAll(Sort sort);
 }

位于 ReactiveSortingRepository 接口之上的就是各个与具体数据库操作相关的接口。以下一讲要介绍的 ReactiveMongoRepository 接口为例，它在 ReactiveSortingRepository 接口基础上进一步添加了针对 MongoDB 的各种特有操作。ReactiveMongoRepository 接口定义：

public interface ReactiveMongoRepository<T, ID> extends 
 ReactiveSortingRepository<T, ID>, ReactiveQueryByExampleExecutor<T> {
  
   <S extends T> Mono<S> insert(S entity);
   <S extends T> Flux<S> insert(Iterable<S> entities);
   <S extends T> Flux<S> insert(Publisher<S> entities);
   <S extends T> Flux<S> findAll(Example<S> example);
   <S extends T> Flux<S> findAll(Example<S> example, Sort sort);
 }

可以看到该接口同时扩展了 ReactiveSortingRepository 和 ReactiveQueryByExampleExecutor 接口，而 ReactiveQueryByExampleExecutor 接口提供了针对 QueryByExample（按示例查询）机制的响应式实现版本。

以上介绍的 Spring Data Reactive 中相关核心接口之间的继承关系，Spring Data Reactive 中核心接口继承关系图：

响应式编程实战(12)-12_响应式_02

响应式数据存储库

目前，Spring Data Reactive Repository 支持包括 MongoDB、Cassandra、Redis、Couchbase 等多款主流的 NoSQL 数据库。Spring Data 项目有几个单独的模块，分别针对这些 NoSQL 数据库提供了响应式的数据访问支持。本文将结合 ReactiveSpringCSS 案例中使用到的 MongoDB 和 Redis 给出相关的代码示例。

正在孵化的模块，目前只包含一个组件：Spring Data R2DBC。R2DBC 是Reactive Relational Database Connectivity的简写，代表响应式关系型数据库连接，相当于响应式数据访问领域的 JDBC 规范。

创建响应式数据访问层组件

基于 Spring Data Reactive 抽象，在 Spring Boot 中创建响应式数据访问层组件的常见方式一般有三种，分别是继承 ReactiveCrudRepository 接口、继承数据库专用的 Reactive Repository 接口，以及自定义数据访问层接口。

继承 ReactiveCrudRepository 接口

如果基本的 CRUD 操作就能满足我们的需求，那么继承 ReactiveCrudRepository 接口来创建响应式数据访问层组件是最直接的方法，如下代码展示了这一使用方式。

public interface OrderReactiveRepository 
         extends ReactiveCrudRepository<Order, Long> {
  
         Mono<Order> getOrderByOrderNumber(String orderNumber);    
 }

在上述代码中，假如我们的领域实体是 Order，包含了主键 Id、订单编号 OrderNumber 等属性，那么我们就可以定义一个 OrderReactiveRepository 接口，然后让该接口继承自 ReactiveCrudRepository 接口。根据需要，我们可以完全使用 ReactiveCrudRepository 接口的内置方法，也可以使用前面介绍的方法名衍生查询来实现丰富的自定义查询，如上述代码中所示的 getOrderByOrderNumber() 方法。

继承数据库专用的 Reactive Repository 接口

如果需要使用针对某种数据库的特有操作，我们也可以继承数据库专用的 Reactive Repository 接口。在如下示例中，OrderReactiveRepository 接口就继承了 MongoDB 专用的 ReactiveMongoRepository 接口。

public interface OrderReactiveRepository
         extends ReactiveMongoRepository<Product, String> {
  
         Mono<Order> getOrderByOrderNumber(String orderNumber);    
 }

自定义数据访问层接口

最后，我们也可以摒弃 Spring Data 的 Repository 接口，而采用完全自定义的数据访问层接口。如下代码定义了一个 OrderReactiveRepository 接口，我们看到该接口没有继承自 Spring Data 中任何一层次的 Repository 接口。

public interface OrderReactiveRepository{
    Mono<Boolean> saveOrder(Order order);
     
    Mono<Boolean> updateOrder(Order order);
     
    Mono<Boolean> deleteOrder(Long orderId);
     
    Mono<Product> findOrderById(Long orderId);
     
    Flux<Product> findAllOrders();
 }

针对这种实现方式，我们需要构建针对 OrderReactiveRepository 接口的实现类，而在实现类中一般会使用 Spring 提供的各种响应式数据库访问模板类，来实现相应的数据访问功能。例如，针对 MongoDB，我们就可以使用 ReactiveMongoTemplate 模板类。

后续内容中，以上三种创建响应式数据访问层组件的实现方法我们都会用到。

总结

数据访问是一切应用系统的基础，也是全栈响应式访问链路中的最后一环。Spring 作为一款集成性的开发框架，专门提供了 Spring Data 组件来对数据访问过程进行抽象。在新版的 Spring 中，也专门对 Spring Data 组件进行了升级，并整合了响应式访问模型。

FAQ

在使用 Spring Data 时，创建响应式数据访问层组件有哪些具体的方式？

使用 Spring Data R2DBC：Spring Data R2DBC 是 Spring Data 的一个子项目，它提供了响应式的关系型数据库访问支持，使用 R2DBC 标准来实现异步和非阻塞的数据库访问。通过使用 Spring Data R2DBC，我们可以创建响应式的数据访问层组件，并使用 Reactor 或 RxJava 等响应式编程库来处理异步和流式数据。
使用 Spring Data MongoDB Reactive：Spring Data MongoDB Reactive 是 Spring Data 的一个子项目，它提供了响应式的 MongoDB 数据库访问支持。通过使用 Spring Data MongoDB Reactive，我们可以创建响应式的数据访问层组件，并使用 Reactor 或 RxJava 等响应式编程库来处理异步和流式数据。
使用 Spring WebFlux 和任意的非阻塞数据访问库：Spring WebFlux 是 Spring 框架提供的一个响应式 Web 框架，它支持使用任意的非阻塞数据访问库来实现响应式的数据访问。通过使用 Spring WebFlux 和一个非阻塞的数据访问库（例如 Reactive Redis、Reactive Cassandra 等），我们可以创建响应式的数据访问层组件，并使用 Reactor 或 RxJava 等响应式编程库来处理异步和流式数据。

无论使用哪种方式，创建响应式的数据访问层组件的核心思想都是一致的：使用异步和非阻塞的方式来实现数据访问，并使用响应式编程库来处理异步和流式数据。这样可以大大提高系统的并发性和吞吐量，同时也能更好地支持现代化的微服务架构。

在今天内容的基础上，后文我们将基于 Spring Data 框架中的 Spring Data MongoDB 来访问 NoSQL 数据库，并结合 ReactiveSpringCSS 案例完成对响应式数据访问层组件的设计和实现。

标签：实战,12,响应,Spring,编程,接口,访问,Mono,Data
From： https://blog.51cto.com/JavaEdge/6680947