vlambda博客
学习文章列表

领域驱动设计 (DDD) 作者：Eric Evans (Addison-Wesley, 2004) 应该在每个软件工程师的书架上占据一席之地。这是 因为它定义并形式化 成功的微服务架构的重要理论基础。 DDD 建立了一个通用词汇表（即上下文、领域、模型和普遍存在的语言），并制定了一组维护模型完整性的原则。 DDD 最重要的后果之一是，根据 DDD 定义的单个有界上下文通常映射到单独的微服务中，如下所示：

图 7.1 由实施领域驱动设计和领域驱动设计提炼

由于 DDD 非常关注业务核心领域， 尤其是表达、创建和检索领域模型的工件，因此以下对象通常会在本章中被引用——entity, 值对象、聚合、存储库。

在使用 DDD 实现应用程序期间，如果服务中存在这样的层，则应将上述对象映射到应用程序持久层。这样的领域模型为逻辑和物理数据模型奠定了基础。

微服务架构引入的主要与持久性相关的变化可能是强烈鼓励 不 在服务之间共享数据存储。这意味着每个逻辑服务都拥有并管理其数据库（如果它需要数据库），理想情况下，没有其他服务可以以不同于服务 API 调用的方式访问数据。

解释这种分离的所有原因超出了本书的范围，但最重要的原因如下：

考虑下图：

图 7.2 每个服务的独立数据库

在物理级别，存储分离可以通过几种不同的方式实现。最简单的方法是为所有服务拥有一个数据库服务器和一个数据库，但使用单独的模式（每个微服务的模式）来划分它们。这样的配置可以很容易地实现，需要最少的服务器资源，并且在生产中不需要太多的管理，所以它在应用程序开发的第一阶段非常有吸引力。可以使用数据库访问控制规则强制执行数据分离。这样的做法很容易实现，但同时也很容易被破解。由于数据存储在同一个数据库中，开发人员有时很想编写一个查询来检索或更新属于多个服务的数据。通过仅损害一项服务来损害整个系统的安全性也更容易。下图显示了上述系统的设计：

图 7.3 每个服务的架构

服务可能共享单个数据库服务器，但具有具有不同访问凭据的不同数据库。这种方法改进了数据分离，因为编写能够访问外部数据的单个查询要困难得多。但是，这也使备份例程变得有点复杂。下图描述了这一点：

图 7.4 每个服务的数据库

每个服务都可能有它的数据库服务器。这种方法需要更多的管理，但为细粒度调整数据库服务器以满足具体服务的需求提供了一个良好的起点。此外，在这种情况下，可以仅 垂直和水平 扩展 需要这种可扩展性的数据库。下图显示了这一点：

图 7.5 每个服务的数据库服务器

在实现一个软件系统时，所有前面提到的技术可以 很容易 根据实际需求 系统。以下架构显示了这样一个系统的设计：

图 7.6 持久性策略的混合组合

此外，可以有不同的数据库服务器实例，我们现在可以一起使用不同的数据库引擎（SQL 和 NoSQL）来获得最佳结果。这种方法被称为多语言持久性。

2006 年， Neal Ford 提出了术语多语言编程 .这表达了这样的想法，即软件系统可以用不同的语言混合编写，以便从最好的语言中获得最显着的提升——适合业务或技术环境的语言。之后，引入了许多新的编程语言，目的是在特定领域做到最好，或者擅长多个领域。

与此同时，在数据持久性领域发生了另一种类似的思维转变。这导致人们质疑如果不同的应用程序部分根据业务或技术要求使用不同的持久性技术会发生什么。例如，为分布式 Web 应用程序存储 HTTP 会话和在社交网络上存储朋友图需要不同的操作特性，因此需要不同的数据库。如今，一个系统同时拥有两种或多种不同的数据库技术是相当标准的。

从历史上看，大多数关系数据库管理系统 （RDBMSes）是基于相同的 ACID 原则构建，并提供非常相似的 SQL 方言用于与存储进行通信。一般而言，RDBMS 适用于广泛的应用程序，但对于许多常见用例（例如，存储图、内存存储和分布式存储），它们很少具有最佳性能和操作特性。相比之下，NoSQL 数据库出现 最近有更广泛的基本原则，即使大多数 NoSQL 数据库不能有效地用作通用数据存储，也可以为常见用例提供更好的功能。下图描述了这一点：

图 7.7 每个服务都使用最适合其需求的持久化技术

此外，多语言持久性是以复杂性为代价的。每一种新的存储机制都会引入新的 API 和需要学习的范式、需要开发或采用的新客户端库，以及需要在开发和生产中解决的一组新标准问题。此外，对 NoSQL 数据库的错误使用可能意味着对服务的完全重新设计。使用 right 持久性技术（SQL 或 NoSQL）应该会让这更舒服，但挑战不会消失。

Spring Framework 有单独的子项目专门用于数据持久性，称为 Spring Data (http://projects .spring.io/spring-data）。本章的其余部分描述了 Spring Data 中可用的不同设计方法和数据库连接器，特别是针对响应式编程模式以及它们如何改变应用程序访问存储在多语言持久层中的数据的方式。

在一个设计合理的microservice架构中，所有的服务都是无状态的，所有的状态都存储在知道如何管理的特殊服务中数据持久化。在云环境中，无状态服务可实现高效的可扩展性和高可用性。然而，管理和扩展 database 服务器要困难得多，尤其是那些不是为云设计的服务器。大多数 云提供商通过使用他们的数据库即服务（DBaaS ) 解决方案。此类存储解决方案可能是普通数据库（MySQL、PostgreSQL 和 Redis）的定制版本，也可能是从头开始设计为仅在云中运行（AWS Redshift、Google BigTable 和 Microsoft CosmosDB）。

通常，云存储或数据库使用的算法按以下方式工作：

通常，这种方法允许客户（在我们的例子中是软件开发人员）和云提供商专注于他们的主要目标。云提供商实施了存储和处理客户数据的最有效方式，最大限度地减少了底层基础设施的支出。同时，客户端专注于应用程序的主要业务目标，不会花时间配置数据库服务器、复制或备份。这种关注点分离对客户来说并不总是最有利可图的，甚至可能根本不可能。然而，当相关时，它有时允许我们只用少数工程师构建成功且广泛使用的应用程序。

一些最知名的云原生数据存储和数据库服务如下：

在实际业务系统中，为了处理客户端请求，通常需要查询两个或多个服务拥有的数据。例如，客户可能希望查看他们的所有订单以及与他们的订单相对应的付款状态。在微服务架构之前，这可以通过单个连接查询来实现，但现在这违反了最佳实践。为了处理多服务请求，需要实现一个适配器服务来查询订单和支付服务，应用所有需要的转换，并将聚合结果返回给客户端。此外，很明显，如果两个服务通信很多或高度依赖于彼此，它们可能是合并为一个服务的正确候选者（如果这种服务合并不会损害域驱动设计）。下图描述了这一点：

图 7.8 从 Order 服务和 Payment 服务聚合数据的适配器服务

阅读策略非常简单，但要同时实施和更新需要几个服务的策略要困难得多。假设客户想要下订单，但只能 验证适当的库存余额和付款信息。每个服务都有自己的数据库，因此在一个工作流中，一个业务事务涉及两个或多个微服务和数据库。 解决这样的问题有几种方法，但最流行的两种方法是  分布式事务和事件驱动架构。

最终一致性

让我们回过头来分析分布式事务在软件系统中的作用。很明显，我们使用分布式事务来确定系统状态，或者换句话说，我们消除了系统中某些状态可能不一致的不确定性。然而，这种消除不确定性的要求非常严格。 实施领域驱动设计的作者 Vaughn Vernon 提出将不确定性嵌入到领域模型中。据他说，如果很难保护系统免受不一致状态的影响，并且无论我们如何努力对抗它仍然会出现不一致，那么接受不确定性并将其嵌入到常规业务工作流程中可能是有益的。

例如，我们的系统仍然可以通过引入一个名为 verifying payment info 的新状态来创建没有经过验证的付款信息的订单。这个新事件将一个不确定的情况（支付信息可能有效或无效）转换为一个单独的业务步骤，该步骤可能被占用有限的时间（直到付款信息得到验证）。使用这种方法，我们不需要我们的系统始终保持一致。相反，需要确保系统对每个业务交易的状态有一致的看法。这种未来的一致性称为最终一致性。 以下 图描述了这一点：

图 7.9 Order 和 Payment 服务都更新私有数据库并通过消息传递工作流状态

通常，最终的一致性保证足以构建一个成功推进其操作的健全系统。此外，任何分布式系统都必须处理最终一致性，以便可用（处理用户请求）和分区容错（在组件之间的网络中断中幸存）。

 

命令查询职责分离

Command Query Responsibility Segregation（CQRS）常用同时 事件溯源。 CQRS 由两部分组成：

• The writing part receives state-update commands and stores them in an underlying event store, but does not return an entity state.
• The reading part does not change the state and returns it for a requested query type. State representations for distinct queries are stored in views, which are recalculated asynchronously after update events are received as commands.

CQRS 模式工作 的方式如下：

图 7.10 订单服务的 CQRS 实现。写入部分存储更新命令，读取部分异步计算预期查询的视图

 

CQRS 模式允许软件系统以流式方式处理大量数据，同时，它可以快速响应有关当前系统状态的各种查询。

有多种类型的数据库称为嵌入式数据库。此类数据库在应用程序进程本身内部运行，不需要通过网络进行任何通信。对于嵌入式数据库，没有硬性要求要有有线协议，即使有些已经或可能作为嵌入式模式或作为单独的服务运行。在本章的后面，我们将在几个示例中使用 H2 嵌入式数据库。

但是，大多数软件使用在不同服务器（或不同容器）上的不同进程中运行的数据库。应用程序使用一个名为 数据库驱动程序的专用客户端库来通信 与外部数据库。此外，有线协议定义了数据库驱动程序和数据库本身的通信方式。它定义了客户端和数据库之间发送的消息顺序的格式。在大多数情况下，有线协议与语言无关，因此 Java 应用程序可以查询用 C++ 编写的数据库。

由于有线协议通常设计为在 TCP/IP 上工作，因此无需阻塞有线协议。与同步 HTTP 通信一样，协议本身并没有阻塞，它是客户端在等待结果的同时决定阻塞。此外，TCP 是一种异步协议，通过滑动窗口实现的流量控制来支持背压。但是，滑动窗口 approach 已针对通过网络发送字节块进行了调整，可能无法反映应用程序背压的需求以最好的方式。例如，当从数据库接收行时，更自然地以行数请求数据处理的下一部分，而不是依赖于定义网络缓冲区大小的系统设置。当然，有线协议可能有意使用另一种机制，甚至是多种机制的组合来实现背压，但必须记住，TCP 机制也一直在幕后工作。

也可以使用更高级的协议作为数据库有线协议的基础。例如，我们可能会使用 HTTP2、WebSockets、gRPC 或 RSocket。 第 8 章， 使用 Cloud Streams 扩展，对 RSocket 和 gRPC 协议进行了简要比较。

除了背压问题之外，还有不同的方法可以在客户端和数据库之间传递大数据集。例如，客户端插入数万行数据，或者分析查询结果包含数百万行。为了简单起见，让我们只考虑后一个用例。一般来说，有几种方法可以传达这样的结果集：

下图显示了分块结果流的交互流：

图 7.11 使用块迭代查询结果

一般来说，不同的数据库 在其有线协议中实现 一种或多种方法 。例如，MySQL 知道如何将数据作为一个整体或作为一个流逐行发送。同时，PostgreSQL数据库有一个概念称为portal，它允许客户端传播准备接收的尽可能多的数据行。上图描述了 Java 应用程序如何使用这种方法。

在这个级别上，一个设计良好的数据库有线协议可能已经具备反应所需的所有特性。同时，即使是最直接的协议也可能被包含一个反应式驱动程序，该驱动程序可能使用 TCP 控制流进行背压传播。

数据库驱动程序是一个库，它使数据库有线协议适应语言结构，例如方法调用、回调或潜在的反应流。在关系数据库的情况下，驱动程序通常实现语言级别的 API，例如用于 Python 的 DB-API 或用于 Java 的 JDBC。

以同步 阻塞方式编写的软件使用相同的数据访问方法也就不足为奇了。此外，通常通过驱动程序与外部数据库的通信与与外部 HTTP 服务的通信没有什么不同。例如，Apache Phoenix JDBC 驱动程序基于 Apache Calcite 框架的 Avatica 组件，并使用 JSON 或 HTTP 上的协议缓冲区。因此，理论上，我们也可以将响应式设计应用于数据库通信协议并获得非常相似的好处，就像 Spring WebFlux 模块中的响应式 WebClient 。下图显示，从网络通信的角度来看，HTTP 请求和数据库查询非常相似：

图7.12 阻塞HTTP请求和数据库请求的类似阻塞IO行为

通常，数据库驱动程序的阻塞特性是由上层 API 决定的，而不是由有线协议决定的。因此，实现具有适当语言级 API 的反应式数据库驱动程序应该不难。本章稍后将介绍此类 API 的候选者。同时，NoSQL 数据库驱动程序没有既定的语言级 API 可以实现，因此可以自由地异步或响应式地实现自己的 API。例如，MongoDB、Cassandra 和 Couchbase 决定采用这条路线，它们现在提供异步或响应式驱动程序。

1997 年首次发布 Java 数据库连接 (JDBC) 此后一直定义应用程序如何与数据库通信（主要关系）， 为数据访问提供统一的 API一个Java平台。最新的 API 修订版 4.3 于 2017 年发布，并包含在 Java SE 9 中。

JDBC 允许多个数据库客户端 drivers 存在并被同一个应用程序使用。 JDBC Driver Manager 负责正确注册、加载和使用所需的驱动程序实现。加载驱动程序时，客户端可以使用适当的访问凭据创建连接。 JDBC 连接使得初始化和执行语句成为可能，例如 SQL 的 SELECT,CREATE,插入、更新和DELETE。更新数据库状态作为执行结果的语句返回一些受影响的行，查询语句返回java.sql.ResultSet，它是result.ResultSet 是很久以前设计的，有一个奇怪的 API。例如，枚举一行中的列的索引从1 开始，而不是从0。

ResultSet 接口设计用于向后迭代，甚至是随机访问，但这种级别的兼容性要求驱动程序在允许任何处理之前加载所有行。为简单起见，我们假设 ResultSet 类似于结果行的简单迭代器。这种假设允许底层实现对分块的结果集进行操作，并按需从数据库中加载批次。任何底层异步实现都必须包装到 JDBC 级别的同步阻塞调用中。

在性能方面，JDBC 允许对非选择查询进行批处理。这应该允许在更少的网络请求中与数据库进行通信。但是，由于 JDBC 被设计为同步和阻塞的，因此在处理大型数据集时这无济于事。

尽管 JDBC 被设计为业务逻辑级 API，但它操作 使用表、行和列，而不是 span class="emphasis">entities 和 aggregates 作为领域驱动设计的建议。因此，如今，JDBC 被认为太低级而无法直接使用。为此，Spring 生态系统具有 Spring Data JDBC 和 Spring Data JPA 模块。还有很多完善的库封装了 JDBC 并提供了更令人愉快的 API 来使用。这种库的一个例子是Jdbi。它不仅提出了流畅的 API，而且与 Spring 生态系统有很好的集成。

使关系数据库访问反应

由于 JDBC 是 Java 世界中用于数据访问的 primary 语言级 API（至少对于关系数据源而言），它塑造建立在它之上的所有抽象级别的行为。之前，我们展示了不建议在反应式应用程序中使用阻塞 API，因为它们限制了应用程序的可扩展性。因此，对于我们来说，拥有一个适当的语言级数据库访问 API 以用于反应式应用程序至关重要。不幸的是，没有简单的解决方案可以为此目的稍微调整 JDBC。目前，有两个很有前途的 API 草案可能适合这一领域，我们将在本章后面讨论它们。下图描述了制作反应式 JDBC API 所需的内容：

图 7.13 当前的 JDBC 栈和潜在的响应式替换

为了简化环绕原始JDBC的麻烦，Spring提供 Spring JDBC 模块，它很老但描述得很好。该模块提供了几个版本的JdbcTemplate类，可帮助执行查询并将关系行映射到实体中。它还处理资源的创建和释放，有助于省略常见错误，例如忘记关闭准备好的语句或连接。JdbcTemplate还捕获 JDBC 异常并将它们转换为通用org.springframework.dao异常。

假设我们在 SQL 数据库中有书籍集合，实体由以下 Java 类表示：

classBook{privateint id;private String title;publicBook(){}publicBook(int id, String title){this.id = id;this.title = title;}//getters and setters ...}

使用JdbcTemplate和一个通用的BeanPropertyRowMapper， 我们可以通过以下方式创建一个Spring存储库：

@RepositoryclassBookJdbcRepository{@Autowired
   JdbcTemplate jdbcTemplate;public Book findById(int id){return jdbcTemplate.queryForObject("SELECT * FROM book WHERE id=?",newObject[]{ id },newBeanPropertyRowMapper<>(Book.class));}}

或者，我们可以提供我们自己的映射器类来指导 Spring 如何将一个 ResultSet 转换为域实体：

classBookMapperimplementsRowMapper<Book>{@Overridepublic Book mapRow(ResultSet rs,int rowNum)throws SQLException {returnnewBook(rs.getInt("id"), rs.getString("title"));}}

让我们使用BookMapper类来实现BookJdbcRepository.findAll()方法：

public List<Book>findAll(){return jdbcTemplate.query("SELECT * FROM book",newBookMapper());}

JdbcTemplate 的另一项改进是由 NamedParameterJdbcTemplate 类实现的。这增加了对使用人类可读名称而不是 ? 占位符传递 JDBC 参数的支持。因此，准备好的 SQL 查询及其对应的 Java 代码可能如下所示：

SELECT*FROM book WHERE title = :searchtitle

这是一个经典的准备好的 SQL 语句：

SELECT*FROM book WHERE title = ?

这似乎只是一个很小的改进，但命名参数比索引提供了更好的代码可读性，尤其是当查询需要六个参数时。

总之，Spring JDBC 模块由高级抽象使用的实用程序、帮助类和工具组成。更高级别的 API 不限制 Spring JDBC 模块，因此它可以相对容易地吸收所需的响应式支持，因为底层 API 也支持它。

Spring Data JDBC 是 Spring Data 系列中一个相当新的模块。它旨在简化基于 JDBC 的存储库的实现。 Spring Data 存储库，包括基于 JDBC 的存储库，由 < em>领域驱动设计 作者：Eric Evans。这意味着我们建议每个聚合根都有一个存储库。 Spring Data JDBC 具有用于简单聚合的 CRUD 操作，支持 @Query注解和实体生命周期事件。

要使用 Spring Data JDBC，我们必须修改 Book实体并应用org.springframework.data.annotation.Id注解到 id字段。存储库要求实体具有唯一标识符，因此重构以在存储库中使用的Book实体如下所示：

classBook{@Idprivateint id;private String title;// other parts are unchanged}

现在让我们定义BookRepository接口，从CrudRepository<Book, Integer>派生它：

@RepositorypublicinterfaceBookSpringDataJdbcRepositoryextendsCrudRepository<Book, Integer>{// (1)@Query("SELECT * FROM book WHERE LENGTH(title) = "+// (2)"(SELECT MAX(LENGTH(title)) FROM book)")
   List<Book>findByLongestTitle();// (2.1)@Query("SELECT * FROM book WHERE LENGTH(title) = "+"(SELECT MIN(LENGTH(title)) FROM book)")
   Stream<Book>findByShortestTitle();// (3)@Async// (4)@Query("SELECT * FROM book b "+"WHERE b.title = :title")
   CompletableFuture<Book>findBookByTitleAsync(// (4.1)@Param("title") String title);@Async// (5)@Query("SELECT * FROM book b "+"WHERE b.id > :fromId AND b.id < :toId")
   CompletableFuture<Stream<Book>>findBooksByIdBetweenAsync(// (5.1)@Param("fromId") Integer from,@Param("toId") Integer to);}

前面的代码是做下面的事情：

为了通知 Spring 需要使用 Spring Data JDBC 生成BookRepository实现，我们必须向 Spring Boot 应用程序添加下一个依赖项：

compile('org.springframework.data:spring-data-jdbc:1.0.0.RELEASE')

还需要在应用程序配置中添加 @EnableJdbcRepositories注解。在底层，Spring Data JDBC使用Spring JDBC和< code class="literal">NamedParameterJdbcTemplate，之前讨论过。

Spring Data JDBC是一个非常小的模块，它提供了一种为小型微服务构建简单持久层的便捷方法。但是，它被有意设计为简单，而不是针对 ORM 方面，例如缓存、实体的延迟加载和复杂的实体关系。出于这些目的，Java 生态系统有一个单独的规范，称为 Java Persistence API（JPA< /strong>)。 

使 Spring Data JDBC 反应式

Spring Data JDBC 是一个更大的项目 Spring Data Relational 的一部分。 Spring Data JDBC 需要 JDBC，这是一个完全阻塞的 API，不适合完全反应式堆栈。在撰写本文时，Spring Data 团队开发了 R2DBC 规范，该规范允许驱动程序提供与数据库的完全反应式和非阻塞集成。这些努力可能会在 Spring Data R2DBC 模块中采用，该模块将成为 Spring Data Relational 项目的一部分。下图显示了 Spring Data Relational 的潜在反应堆栈：

图 7.14 当前的 Spring Data JDBC 栈和潜在的响应式替换

JPA第一次出现是在2006年（最新版本2.2，2013年发布，有时也叫JPA2），以及它旨在描述 Java 应用程序中的关系数据管理。如今，JPA 是定义应用程序如何组织持久层的标准。它由 API 本身和 Java Persistence Query Language (JPQL)。 JPQL 是一种类似于 SQL 的平台独立语言，它通过 存储库而不是数据库查询JPA 实体对象。

与用于数据库访问的标准 的JDBC 相比，JPA 是一个标准 用于 对象关系映射（ ORM），它允许将代码中的对象映射到数据库中的表。 ORM 通常在后台使用 JDBC 和动态生成的 SQL 查询，但这种机制大多对应用程序开发人员隐藏。 JPA 不仅允许映射实体，还允许映射实体关系，以便轻松加载关联对象。 

最常用的 JPA 实现是 Hibernate (http://hibernate.org) 和 EclipseLink (<类="ulink" href="http://www.eclipse.org/eclipselink" target="_blank">http://www.eclipse.org/eclipselink）。这两者都实现了 JPA 2.2 并且可以互换。除了实现 JPA 标准外，这两个项目还提出了一组额外的特性，这些特性未在规范中定义，但在某些情况下可能很方便。例如，EclipseLink 使得处理数据库更改事件和描述实体到多个数据库中的表的映射成为可能。另一方面，Hibernate 提出了对时间戳和自然 ID 的更好支持。这两个库都支持多租户。但是，我们应该明白，当使用 exclusive 功能时，这些库不再是可互换的。

使用 JPA 实现而不是纯 JDBC 的另一个原因是由于 Hibernate 和 EclipseLink 提供的缓存特性。这两个库都允许我们最小化在一级会话缓存中缓存实体的实际数据库请求数量，甚至在二级外部缓存中。仅此功能可能会对应用程序性能产生显着影响。

Spring Data JPA 同样允许我们像 Spring Data JDBC 一样构建存储库，但在内部 它 使用更强大的 JPA ——基于实施。 Spring Data JPA 对 Hibernate 和 EclipseLink 都有 excellent 支持。 Spring Data JPA 动态生成基于方法名称约定的 JPA 查询，提供 Generic DAO 模式的实现并添加对 Querydsl 库的支持 (http://www.querydsl.com)，它支持优雅的、类型安全的、基于 Java 的查询。

现在，让我们创建一个简单的应用程序来演示 Spring Data JPA 的基础知识。以下依赖项获取 Spring Boot 应用程序所需的所有模块：

compile('org.springframework.boot:spring-boot-starter-data-jpa')

Spring Boot 足够聪明，可以推断出正在使用 Spring Data JPA，因此甚至不需要添加 @EnableJpaRepositories 注释（但如果我们愿意，我们可以这样做）。 Book实体应如下所示：

@Entity@Table(name ="book")publicclassBook{@Idprivateint id;private String title;// Constructors, getters, setters...}

标有javax.persistence.Entity注解的Book实体允许setting JPQL 查询中使用的实体名称。 javax.persistence.Table注解定义了表的坐标，也可以定义约束和索引。需要注意的是，我们必须使用 javax.persistence.Id 而不是org.springframework.data.annotation.Id注解注解。

现在，让我们定义一个 CRUD 存储库，其中包含一个使用命名约定生成查询的自定义方法，以及另一个使用 JPQL 查询的方法：

@RepositoryinterfaceBookJpaRepositoryextendsCrudRepository<Book, Integer>{

   Iterable<Book>findByIdBetween(int lower,int upper);@Query("SELECT b FROM Book b WHERE LENGTH(b.title) = "+"(SELECT MIN(LENGTH(b2.title)) FROM Book b2)")
   Iterable<Book>findShortestTitle();}

类路径中的 JDBC 驱动程序、一个 Spring Boot 依赖项、Book实体类和BookJpaRepository接口足以提供一个基于下一个技术堆栈的基本但非常通用的持久层——Spring Data JPA、JPA、JPQL、Hibernate 和 JDBC。

Spring Data JPA 和 Spring Data JDBC 都是 优秀 解决方案，用于连接至少提供 JDBC 驱动程序的关系型数据库，但是大多数 NoSQL 数据库 不这样做。在这种情况下，Spring Data 项目有几个单独的模块，分别针对流行的 NoSQL 数据库。 Spring 团队积极为 MongoDB、Redis、Apache Cassandra、Apache Solr、Gemfire、Geode 和 LDAP 开发模块。同时，社区为以下数据库和存储开发模块——Aerospike、ArangoDB、Couchbase、Azure Cosmos DB、DynamoDB、Elasticsearch、Neo4j、Google Cloud Spanner、Hazelcast 和 Vault。

使用 MongoDB 的代码几乎 与 Spring Data JDBC 示例中的代码相同。要使用 MongoDB 存储库，我们必须添加以下依赖项：

compile('org.springframework.boot:spring-boot-starter-data-mongodb')

假设我们必须实现一个在线图书目录。解决方案应该基于 MongoDB 和 Spring Framework。为此，我们可以使用以下 Java 类定义Book实体：

@Document(collection ="book")// (1)publicclassBook{@Id// (2)private ObjectId id;// (3)@Indexed// (4)private String title;@Indexedprivate List<String> authors;// (5)@Field("pubYear")// (6)privateint publishingYear;// constructors, getters and setters// ...}

在这里，我们使用 @Documentannotation(1) 而不是 JPA@Entity 来自org.springframework.data.mongodb.core.mapping 包。此注释特定于 MongoDB，并且可以引用正确的数据库集合。此外，为了定义实体的内部 ID，我们使用 MongoDB 特定类型，org.bson.types.ObjectId(3) ， 结合 Spring Data annotationorg.springframework.data.annotation.Id(2)。我们的实体以及我们的数据库 document 将包含一个title字段，其中也将被 MongoDB 索引。为此，我们使用@Indexedannotation(4) 装饰该字段。此注释提供了一些有关索引详细信息的配置选项。此外，一本书可能有一个或多个作者，我们通过将 authors 字段的类型声明为 List<String> 来表示这一点。 (5)。 authors字段也被索引。请注意，这里我们没有创建对具有多对多关系的单独author表的引用，因为它很可能是用关系数据库实现的，而是我们嵌入作者姓名作为book实体的子文档。最后，我们定义publishingYear字段。实体和数据库中的字段名称不同。  @Field 注解允许对这种情况进行自定义映射 (6)。

在数据库中，这样的book实体将由以下 JSON 文档表示：

{"_id": ObjectId("5b1c0908eb696eddfadc0b1b"),                  /*(1)*/
    "title":"The Expanse: Leviathan Wakes","pubYear":2011,                                              /*(2)*/
    "authors":[                                                  /*(3)*/
        "Daniel Abraham",                                          /*   */
        "Ty Franck"                                                /*   */
    ],"_class":"org.rpis5.chapters.chapter_07.mongo_repo.Book"     /*(4)*/
}

正如我们所见，MongoDB 使用专门设计的数据类型来表示文档的 ID (1)。< /span>在这种情况下， publishingYear映射到pubYear字段(2 ) 和作者由数组(3)表示。此外，Spring Data MongoDB 添加了 support_class 字段，该字段描述了用于对象-文档映射的 Java 类。

对于 MongoDB，存储库接口应该扩展org.springframework.data.mongodb.repository.MongoRepository接口(1)，它反过来扩展了CrudRepository， 我们已经在前面的例子中使用过：

@RepositorypublicinterfaceBookSpringDataMongoRepositoryextendsMongoRepository<Book, Integer>{// (1)

   Iterable<Book>findByAuthorsOrderByPublishingYearDesc(// (2)
       String... authors
   );@Query("{ 'authors.1': { $exists: true } }")// (3)
   Iterable<Book>booksWithFewAuthors();}

当然，MongoDB 存储库支持基于命名约定的查询生成，因此findByAuthorsOrderByPublishingYearDesc< /code>method 按作者搜索书籍并返回按出版年份排序的结果，从最近的出版物开始。此外，org.springframework.data.mongodb.repository.Query注解允许我们编写特定于 MongoDB 的查询。例如，前面的查询(3) 巧妙地搜索不止一位作者的书籍。

应用程序的其余部分应该与使用 Spring Data JDBC 或 Spring Data JPA 的情况相同。

尽管我们已经谈到了使用 Spring 进行数据持久化的主要方法，但我们几乎没有触及这个领域的皮毛。我们完全省略了事务管理、数据库初始化和迁移（Liquibase、Flyway），这些都是实体映射、缓存和性能调优的最佳实践。所有这些领域都可以写不止一本书，但我们必须继续前进并研究如何以反应方式进行持久性。

使用 Spring Framework 实现对 NoSQL 数据库的响应式支持需要整个底层基础架构提供响应式或异步 API。一般来说，NoSQL 数据库出现的时间相对较晚，并且发展迅速，因此没有很多基础设施受到同步阻塞 API 的严重限制。因此，与使用 JDBC 驱动程序的关系数据库相比，使用 NoSQL 数据库实现反应式持久性应该更容易。到目前为止，Spring Data 有一些响应式数据连接器，MongoDB 就是其中之一。这将在后面的 使用 Spring Data 进行响应式数据访问 部分进行介绍。

在研究 Spring Framework 甚至 Java 的 persistence 选项时，我们查看了 JDBC、JPA、Hibernate、EclipseLink 、Spring Data JDBC 和 Spring Data JDBC，所有这些 API 和库本质上都是同步和阻塞的。尽管它们几乎总是用于从涉及网络调用的外部服务中检索数据，但它们不允许非阻塞交互。因此，所有前面提到的  API 和库都与响应式范例发生冲突。向数据库发出查询的 Java 线程注定要被阻塞，直到第一条数据到达或发生超时，从响应式应用程序的资源管理的角度来看，这是非常浪费的。如第 6 章中所述， WebFlux 异步非阻塞通信，这种方法极大地限制了应用程序的吞吐量，并且需要更多的服务器资源，因此也需要更多的资金。

不管是 HTTP 请求还是数据库请求，以阻塞的方式发出 IO 请求都是很浪费的。此外，基于 JDBC 的通信通常使用整个连接池来并行执行查询。相比之下，广泛使用的 HTTP2 协议允许使用同一个 TCP 连接同时发送和接收多个资源。这种方法减少了占用的 TCP 套接字的数量，并允许客户端和服务器（在我们的例子中，这是数据库）更大的并发性。考虑下图：

图 7.15 普通数据库通信与允许多路复用的通信协议（如 HTTP2）之间的比较

当然，连接池的存在是为了在打开新连接时节省时间。也可以在 JDBC 下实现一个通信层，以利用 HTTP2 中的多路复用，但是，JDBC 级别之前的代码必须是同步和阻塞的。

同样，在处理占用少量批次的大型查询结果时，与数据库游标（一种允许对查询结果记录进行迭代的控制结构）的通信看起来像上图的左侧。 第 3 章，反应式流 - 新流的标准 ，从 Reactive Streams 的角度详细分析了通信选项之间的差异，但同样的论点也适用于网络交互。

即使数据库提供了 asynchronous 能够高效通信并利用连接多路复用的非阻塞驱动程序，我们也无法获得在使用 JDBC、JPA 或 Spring Data JPA 时充分发挥它的潜力。因此，要构建一个完全响应式的应用程序，我们必须放弃同步技术并使用响应式类型制作 API。

总结本节，传统和完善的 JDBC 和 JPA 实现可能成为现代反应式应用程序的瓶颈。 JDBC 和 JPA 最有可能在运行时使用过多的线程和过多的内存，同时需要积极的缓存来限制冗长的同步请求和阻塞 IO。

不是同步模型不好；它只是不太适合反应式应用程序，很可能成为一个限制因素。但是，这些模型可能会成功共存。同步和反应式方法都有其优点和缺点。例如，到目前为止，反应式持久性方法无法提出任何在特性方面甚至接近 JPA 的 ORM 解决方案。

尽管 同步 数据访问并不是在实现持久层时消耗服务器资源的最有效方式，但它仍然是一种高度有价值的方法，主要用于构建阻塞 Web 应用程序时。 JDBC 可能是用于访问数据的最流行和最通用的 API，但它几乎完全隐藏了应用程序和数据库之间客户端-服务器通信的复杂性。 Spring Data JDBC 和 Spring Data JPA 为数据持久性提供了更高级的工具，隐藏了查询转换和事务管理的巨大复杂性。所有这些都经过了实战考验，大大简化了现代应用程序的开发方式。

同步数据访问简单，易于调试，易于测试。通过监控线程池也很容易跟踪资源使用情况。同步方法提供了大量不需要任何背压支持的工具（例如 JPA 和 Spring Data 连接器），并且在使用迭代器和同步流时仍然可能很有效。此外，大多数现代数据库在内部使用阻塞模型，因此使用阻塞驱动程序进行交互是很自然的。这种同步方法对本地和分布式事务有很好的支持。在用 C 或 C++ 编写的本机驱动程序上实现包装器也很容易。

同步 数据访问的唯一缺点在于执行的阻塞方式，这与使用反应范式（Netty、Reactor、WebFlux）。

在简要回顾一下同步数据访问技术之后，我们可能会转向响应式数据持久性的探索，看看 Spring Data 的响应式连接器如何在不损害 Spring Data 存储库的多功能性的情况下实现高性能的承诺。

要使用 MongoDB 而不是 的同步数据访问，我们必须向我们的 Gradle 项目添加以下依赖项：

compile 'org.springframework.boot:spring-boot-starter-data-mongodb-reactive'

假设我们想要重构上一节中的简单 MongoDB 应用程序，使其具有响应性。在这种情况下，我们可以将Book实体保持原样，而不做任何修改。与 MongoDB 对象-文档映射关联的所有注释对于同步和反应式 MongoDB 模块都是相同的。但是，在存储库中，我们 现在 必须用反应类型替换普通类型：

publicinterface ReactiveSpringDataMongoBookRepository
   extendsReactiveMongoRepository<Book, Integer>{// (1)@Meta(maxScanDocuments =3)// (2)
   Flux<Book>findByAuthorsOrderByPublishingYearDesc(// (3)
       Flux<String> authors
   );@Query("{ 'authors.1': { $exists: true } }")// (4)
   Flux<Book>booksWithFewAuthors();}

因此，我们的存储库 现在 扩展了ReactiveMongoRepository接口(1)  ;而不是MongoRepository。反过来，ReactiveMongoRepository扩展了ReactiveCrudRepository接口，这是所有响应式连接器的通用接口。

ReactiveCrudRepository 接口是同步 Spring Data 中 CrudRepository 接口的反应式等效接口。 Reactive Spring Data Repositories 使用相同的注解并支持大多数同步提供的特性。因此，响应式 Mongo 存储库支持通过方法名称约定(3) 进行查询，@Query 使用手写 MongoDB 注释查询(4)，以及@Meta注解以及一些额外的查询调整能力( 2） 。它还支持 constructs 用于运行Query by Example  (QBE) 请求。然而，与同步MongoRepository相比，ReactiveMongoRepository扩展了ReactiveSortingRepository接口，它提供了请求特定结果顺序的能力，但不提供分页支持。数据分页的问题在 分页支持 部分中讨论。

像往常一样，我们可以在我们的应用程序中注入一个 ReactiveSpringDataMongoBookRepository 类型的bean，然后Spring Data会提供所需的bean。 以下 代码展示了如何使用响应式存储库将几本书插入MongoDB：

@Autowiredprivate ReactiveSpringDataMongoBookRepository rxBookRepository;      // (1)...
Flux<Book> books = Flux.just(// (2)newBook("The Martian",2011,"Andy Weir"),newBook("Blue Mars",1996,"Kim Stanley Robinson"));

rxBookRepository
   .saveAll(books)// (3).then()// (4).doOnSuccess(ignore -> log.info("Books saved in DB"))// (5).subscribe();// (6)

让我们了解一下前面的代码在这里做了什么：

现在，让我们使用 Reactive Streams 查询 MongoDB。要打印流经反应流的查询结果，我们可以使用以下方便的辅助方法：

privatevoidreportResults(String message, Flux<Book> books){// (1)
   books
      .map(Book::toString)// (2).reduce(// (3)newStringBuilder(),// (3.1)(sb, b)-> sb.append(" - ")// (3.2).append(b).append("\n")).doOnNext(sb -> log.info(message +"\n{}", sb))// (5).subscribe();// (6)}

让我们了解一下 preceding 代码在这里做了什么：

现在，让我们阅读并报告数据库中的所有书籍：

Flux<Book> allBooks = rxBookRepository.findAll();reportResults("All books in DB:", allBooks);

以下代码使用方法命名约定搜索 Andy Weir 的所有书籍：

Flux<Book> andyWeirBooks = rxBookRepository
   .findByAuthorsOrderByPublishingYearDesc(Mono.just("Andy Weir"));reportResults("All books by Andy Weir:", andyWeirBooks);

此外，前面的代码使用 Mono 类型传递搜索条件，并且仅在该Mono 类型时才开始实际的数据库查询产生一个onNext事件。因此，响应式 repository 成为响应式流的自然组成部分，其中传入和传出流是响应式的。

现在，让我们实现一个稍微复杂的 业务用例。我们可能想更新一本书的出版年份，而我们只有书名。因此，首先，我们必须找到所需的图书实例，然后更新出版年份，并将图书保存到数据库中。为了使我们的用例更加复杂，我们假设 title 和 year 值都是异步检索的，有一些延迟，并由 Mono 类型传递。另外，我们想知道我们更新的请求是否成功。到目前为止，我们并不要求更新是原子的，并假设总是不超过一本书具有相同的标题。因此，根据这些要求，我们可以设计以下业务方法 API：

public Mono<Book>updatedBookYearByTitle(// (1)
                    Mono<String> title,// (2)
                    Mono<Integer> newPublishingYear)// (3)

前面的代码执行以下操作：

我们现在可以创建一个测试场景来检查我们的实现是如何updatedBookYearByTitle作品：

Instant start =now();// (1)
Mono<String> title = Mono.delay(Duration.ofSeconds(1))// (2).thenReturn("Artemis")//.doOnSubscribe(s -> log.info("Subscribed for title"))//.doOnNext(t ->//
      log.info("Book title resolved: {}", t));// (2.1)

Mono<Integer> publishingYear = Mono.delay(Duration.ofSeconds(2))// (3).thenReturn(2017)//.doOnSubscribe(s -> log.info("Subscribed for publishing year")) //.doOnNext(t ->//
      log.info("New publishing year resolved: {}", t));// (3.1)updatedBookYearByTitle(title, publishingYear)// (4).doOnNext(b ->
      log.info("Publishing year updated for book: {}", b))// (4.1).hasElement()// (4.2).doOnSuccess(status ->
      log.info("Updated finished {}, took: {}",// (5)
           status ?"successfully":"unsuccessfully",between(start,now())))// (5.1).subscribe();// (6)

前面的代码执行以下操作：

从前面的代码中，我们可以得出结论 workflow 的运行速度不能超过两秒，因为这是解决问题所需的时间出版年份。但是，它可能会运行更长时间。让我们进行实现的第一次迭代：

private Mono<Book>updatedBookYearByTitle(/* First Iteration */
   Mono<String> title,
   Mono<Integer> newPublishingYear
){return rxBookRepository.findOneByTitle(title)// (1).flatMap(book -> newPublishingYear                             // (2).flatMap(year ->{// (3)
            book.setPublishingYear(year);// (4)return rxBookRepository.save(book);// (5)}));}

使用这种方法，在方法的开头，我们使用提供的对 title(1)。一旦找到Book实体(2)，我们就订阅新的出版年份值。然后，一旦新的出版年份值到达，我们就更新Book实体(4) 并调用< code class="literal">save存储库的方法。此代码产生以下输出：

Subscribed for title
Book title resolved: Artemis
Subscribed for publishing year
New publishing year resolved:2017
Publishing year updated for book:Book(publishingYear=2017...
Updated finished successfully, took: PT3.027S

所以，这本书更新了，但是从日志中我们可以看出，我们是在收到书名后才订阅新的出版年份，所以该方法总共花费了三秒多的时间来计算结果。我们可以做得更好。为了启动并发检索过程，我们必须在工作流开始时订阅这两个流。以下代码描述了如何使用 zip 方法执行此操作：

private Mono<Book>updatedBookYearByTitle(/* Second Iteration */
   Mono<String> title,
   Mono<Integer> newPublishingYear
){return Mono.zip(title, newPublishingYear)// (1).flatMap((Tuple2<String, Integer> data)->{// (2)
         String titleVal = data.getT1();// (2.1)
         Integer yearVal = data.getT2();// (2.2)return rxBookRepository
            .findOneByTitle(Mono.just(titleVal))// (3).flatMap(book ->{
               book.setPublishingYear(yearVal);// (3.1)return rxBookRepository.save(book);// (3.2)});});}

在这里，我们zip两个值并同时订阅它们(1)。一旦两个值都准备好了，我们的流就会收到一个Tuple2 容器，其中包含感兴趣的值(2)。现在我们必须解压值(2.1) 和(2.2) 使用data.getT1()和data.getT2()调用。在(3)点，我们查询Book实体，一旦到达，我们就更新出版年份并保存实体到数据库。在 second 迭代之后，我们的应用程序显示以下输出：

Subscribed for title
Subscribed for publishing year
Book title resolved: Artemis
New publishing year resolved:2017
Publishing year updated for the book:Book(publishingYear=2017...
Updated finished successfully, took: PT2.032S

现在，我们可能会看到我们首先订阅了两个流 当两个值都到达时，我们更新了 book 实体。在第二种方法中，我们花费大约两秒钟而不是三秒钟来执行操作。它更快，但需要使用 Tuple2 类型，这需要额外的代码行以及进行转换。为了提高可读性并删除getT1()和getT2()调用，我们可以添加Reactor Addons 模块，它为这种情况提供了一些语法糖。

通过以下新依赖，我们可以改进之前的代码示例：

compile('io.projectreactor.addons:reactor-extra')

这是我们可以改进的方法：

private Mono<Book>updatedBookYearByTitle(/* Third Iteration */
   Mono<String> title,
   Mono<Integer> newPublishingYear
){return Mono.zip(title, newPublishingYear).flatMap(
         TupleUtils.function((titleValue, yearValue)->// (1)
            rxBookRepository
               .findOneByTitle(Mono.just(titleValue))// (2).flatMap(book ->{
                  book.setPublishingYear(yearValue);return rxBookRepository.save(book);})));}

在这里，在 (1) 处，我们可以将 Tuple2 对象的手动解构替换为 TupleUtils类中的">function方法并使用已经解构的值。由于 function 方法是静态的，因此生成的代码非常流畅和冗长：

return Mono.zip(title, newPublishingYear).flatMap(function((titleValue, yearValue)->{...}));

此外，在 (2) 点，我们将 titleValue 再次包装到 Mono 对象。我们可以使用已经具有正确类型的原始title对象，但在这种情况下，我们将订阅两次title流并将接收以下输出。请注意，我们触发了两次标题解析代码：

Subscribed for title
Subscribed for publishing year
Book title resolved: Artemis
New publishing year resolved:2017
Subscribed for title
Book title resolved: Artemis
Publishing year updated for the book:Book(publishingYear=2017...
Updated finished successfully, took: PT3.029S

还有一点是，在第三次迭代中，我们发出 database 请求，仅在收到书名和新出版后才加载图书价值观。但是，我们可能会在出版年份请求仍在进行中但标题值已经存在时开始加载图书实体。第四次迭代展示了如何构建这个反应式工作流程：

private Mono<Book>updatedBookYearByTitle(/* Forth Iteration */
   Mono<String> title,
   Mono<Integer> newPublishingYear
){return Mono.zip(// (1)
      newPublishingYear,// (1.1)
      rxBookRepository.findOneByTitle(title)// (1.2)).flatMap(function((yearValue, bookValue)->{// (2)
      bookValue.setPublishingYear(yearValue);//return rxBookRepository.save(bookValue);// (2.1)}));}

在这里，使用zip运算符(1) 我们订阅新的出版年份值(1.1) 和书籍实体(1.2) 同时。当两个值都到达时 (2)，我们更新实体的出版年份并请求图书保存过程(2.1) .此外，与此业务用例的所有先前迭代一样，即使工作流至少需要两秒钟才能完成，也不会阻塞任何线程。因此，此代码非常有效地使用计算资源。

本练习的重点是演示，借助 Reactive Streams 的功能和 Project Reactor API 的多功能性，很容易构建不同的异步工作流，甚至涉及数据持久层。只需几个反应式操作符，我们就可以彻底改变数据流经系统的方式。然而，并非所有这样的反应流替代方案都是平等的。有些可能运行得更快，而另一些可能运行得较慢，并且在许多情况下，最明显的解决方案并不是最合适的解决方案。因此，在编写反应式管道时，请考虑反应式运算符的替代组合，不要选择想到的第一个，而是选择最适合业务请求的选项。

现在，让我们深入了解 reactive 存储库的工作原理。 Spring Data 中的响应式存储库通过调整底层数据库驱动程序功能来工作。下面可能有一个响应式流兼容的驱动程序或一个 asynchronous 驱动程序，可以封装到响应式 API 中。在这里，我们将了解反应式 MongoDB 存储库如何使用符合反应式流的 MongoDB 驱动程序，以及反应式 Cassandra 存储库如何构建在异步驱动程序上。

首先，ReactiveMongoRepository接口扩展了更通用的接口——ReactiveSortingRepository和ReactiveQueryByExampleExecutor。 ReactiveQueryByExampleExecutor接口允许使用 QBE 语言执行查询。 ReactiveSortingRepository接口扩展了更通用的ReactiveCrudRepository接口并添加了findAll方法，它允许请求查询结果的排序顺序。

由于很多响应式连接器使用ReactiveCrudRepository 接口，让我们仔细看一下。ReactiveCrudRepository声明了保存方法，查找，并删除实体。 单声道<T> save(T entity)方法保存entity，然后返回保存的实体进行进一步操作。请注意，保存操作可能会完全更改实体对象。 单声道<T> findById(ID id)操作consumes实体的id并且 返回 结果包装到Mono。 findAllById 方法有两个覆盖，每个都使用 Iterable<ID> 集合形式的 ID  Publisher<ID> 的形式。除了响应式方法之外，ReactiveCrudRepository和CrudRepository之间唯一显着的区别在于ReactiveCrudRepository 不支持分页，不允许事务性操作。本章稍后将介绍使用 Spring Data 对反应式持久性的事务支持。但是，现在实现分页策略是开发人员的责任。

publicinterfaceReactiveBookRepositoryextendsReactiveSortingRepository<Book, Long>{

    Flux<Book>findByAuthor(String author, Pageable pageable);}

Flux<Book> result = reactiveBookRepository
    .findByAuthor('Andy Weir', PageRequest.of(1,5));

public Flux<T>findAllById(Publisher<ID> ids){return Flux.from(ids).buffer().flatMap(this::findAllById);}

public<S extendsT> Flux<S>insert(Publisher<S> entities){return Flux.from(entities).flatMap(entity -> mongoOperations.insert(entity,...));}

publicclassRxMongoTemplateQueryService{privatefinal ReactiveMongoOperations mongoOperations;// (1)// Constructor...public Flux<Book>findBooksByTitle(String titleRegExp){// (2)
      Query query = Query.query(newCriteria("title")// (3).regex(titleRegExp))).limit(100);return mongoOperations
         .find(query, Book.class,"book");// (4)}}

使用响应式驱动程序 (MongoDB)

Spring Data 中的 Reactive MongoDB 连接基于 MongoDB Reactive Streams Java Driver ( https://github.com/mongodb/mongo-java-driver-reactivestreams）。驱动程序提供异步流处理 具有非阻塞背压。反过来，响应式驱动程序构建在 MongoDB Async Java 驱动程序之上 (http://mongodb.github.io/mongo-java-driver/3.8/driver-async)。异步驱动是低级的和 有一个基于回调的API，所以使用起来不如更高级的好级反应流驱动程序。我们必须注意，除了 MongoDB Reactive Streams Java Driver 还有 MongoDB RxJava Driver (http://mongodb.github.io/mongo-java-driver-rx)，它也是建立在相同的异步 MongoDB 驱动程序之上的。因此，对于 MongoDB 连接，Java 生态系统有一个同步驱动程序、一个异步驱动程序和两个反应驱动程序。

 

当然，如果我们需要比ReactiveMongoTemplate提供更多的查询过程控制，我们可以直接使用响应式驱动。使用这种方法，前面使用纯反应式驱动程序的示例如下：

publicclassRxMongoDriverQueryService{privatefinal MongoClient mongoClient;// (1)public Flux<Book>findBooksByTitleRegex(String regex){// (2)return Flux.defer(()->{// (3)
         Bson query = Filters.regex(titleRegex);// (3.1)return mongoClient                                        //.getDatabase("test-database")// (3.2).getCollection("book")// (3.3).find(query);// (3.4)}).map(doc ->newBook(// (4)
            doc.getObjectId("id"),
            doc.getString("title"),
            doc.getInteger("pubYear"),// ... other mapping routine));}}

Copy

我们看一下前面的代码：

1. 服务是指com.mongodb.reactivestreams.client.MongoClient<的instance /代码>界面。当数据源配置正确时，这个实例应该可以作为 Spring bean 访问。
2. 该服务定义了findBooksByTitleRegex方法，该方法返回一个带有Book实体的Flux .
3. 我们必须返回一个新的Flux实例，它将执行推迟到实际订阅发生的时间。在该 lambda 中，我们使用 com.mongodb.client.model.Filters 定义一个具有 org.bson.conversions.Bson 类型的新查询帮助类。然后我们通过名称引用数据库(3.2) 以及集合(3.3) 。除非我们使用 find 方法发送先前准备好的查询(3.4) 否则不会发生与数据库的通信。
4. 一旦结果开始返回，我们可以将 MongoDB 文档传输到域实体中，如果需要的话。

尽管在前面的示例中我们在数据库驱动程序级别工作，但当我们使用 Reactive Streams 操作时，它仍然非常舒适。此外，我们不需要手动处理背压，因为 MongoDB Reactive Streams Java Driver 已经支持它。 Reactive MongoDB 连接使用基于批量大小的背压需求。这种方法是合理的默认设置，但在使用小的需求增量时会产生许多往返。下图突出显示了响应式 MongoDB 存储库所需的所有 abstraction 层：

图 7.16 Reactive MongoDB stack with Spring Data

使用异步驱动程序 (Cassandra)

我们已经描述了 reactive Mongo 存储库是如何构建在响应式驱动程序之上的。现在，让我们看看反应式 Cassandra 存储库如何适应异步驱动程序。

与 ReactiveMongoRepository 类似，响应式 Cassandra 连接器为我们提供了ReactiveCassandraRepository接口，它还扩展了更通用的ReactiveCrudRepository。 ReactiveCassandraRepository接口由SimpleReactiveCassandraRepository实现，而后者又使用ReactiveCassandraOperations 低级操作的接口。ReactiveCassandraOperations由ReactiveCassandraTemplate类实现。当然，ReactiveCassandraTemplate可以直接在应用中使用，类似于ReactiveMongoTemplate。

ReactiveCassandraTemplate 类内部使用ReactiveCqlOperations。ReactiveCassandraTemplate与 Spring Data 实体如as org.springframework.data.cassandra.core.query.Query，而ReactiveCqlOperations用CQL语句操作 （由<代码 class="literal">String) 可由 Cassandra 驱动程序识别。 ReactiveCqlOperations接口由ReactiveCqlTemplate类实现。反过来，ReactiveCqlTemplate使用ReactiveSession接口进行实际的数据库查询。ReactiveSession由DefaultBridgedReactiveSession类实现，该类将驱动程序提供的异步Session方法连接到响应式执行模式。

让我们更深入地了解DefaultBridgedReactiveSession类如何将异步 API 适配为反应式 API。 execute方法接收一个语句（例如，一个SELECT语句）并且 reactively 返回结果。 execute方法 及其adaptFuturehelper方法 如下所示：

public Mono<ReactiveResultSet>execute(Statement statement){// (1)return Mono.create(sink ->{// (2)try{
      ListenableFuture<ResultSet> future =this.session            // (3).executeAsync(statement);
      ListenableFuture<ReactiveResultSet> resultSetFuture =
          Futures.transform(// (4)
            future, DefaultReactiveResultSet::new);adaptFuture(resultSetFuture, sink);// (5)}catch(Exception cause){
      sink.error(cause);// (6)}});}<T>voidadaptFuture(// (7)
      ListenableFuture<T> future, MonoSink<T> sink
){
  future.addListener(()->{// (7.1)if(future.isDone()){try{
        sink.success(future.get());// (7.2)}catch(Exception cause){
        sink.error(cause);// (7.3)}}}, Runnable::run);}

Copy

让我们看一下前面的代码：

1. 首先，execute方法返回的不是Flux，而是一个 Mono带有ReactiveResultSet的实例。ReactiveResultSet包装异步com.datastax。 driver.core.ResultSet， 它支持分页，以便在返回ResultSet实例时获取结果的第一页，仅在之后获取下一页第一个结果的所有结果都已被消耗。ReactiveResultSet使用以下方法签名适应该行为—Flux<Row>行（）。
2. 我们使用 create 方法创建一个新的Mono实例，该方法将操作推迟到订阅的那一刻。
3. 这是在驱动程序的异步Session实例上执行的异步查询。请注意，Cassandra 驱动程序使用 Guava 的ListenableFuture 来返回结果。
4. 异步ResultSet被包装到一个名为ReactiveResultSet的响应式副本中。
5. 在这里，我们调用adaptFuture辅助方法，该方法将ListenableFuture映射到Mono 。
6. 如果有任何错误，我们必须通知我们的反应订阅者。
7. adaptFuture方法只是简单地为future添加一个新的监听器(7.1)，所以当结果出现时，它会生成一个响应式(7.1) code class="literal">onNextsignal(7.2)。它还通知订阅者有关执行错误（如果有）(7.3)。

需要注意的是，多页ResultSet允许调用fetchMoreResults方法异步获取后续页面数据。< code class="literal">ReactiveResultSet 在internally Flux<Row> rows()方法。 尽管这种方法有效，但在 Casandra 收到完全反应式驱动程序之前，它仍被视为一种中间解决方案。

下图显示了响应式 Spring Data Cassandra 模块的内部架构：

图 7.17 Reactive Cassandra 堆栈与 Spring Data

事务是数据库的标记，它定义了具有许多应该以原子方式执行的逻辑操作的单个单元的边界。因此，我们有一个事务被初始化的时间点，然后对事务对象进行一些操作，然后发生决策时刻。在那一刻，客户端和数据库决定是否应该成功提交或回滚事务。

在同步世界中，transaction 对象通常保存在ThreadLocal容器。但是， ThreadLocal 不适合将数据与反应流相关联，因为用户无法控制线程切换。事务需要将底层资源绑定到具体化流。使用 Project Reactor，这可以通过利用 Reactor 上下文来实现，如 Chapter 4, < em>Project Reactor - 响应式应用程序的基础。

@Document(collection ="wallet")// (1)publicclassWallet{@Idprivate ObjectId id;// (2)private String owner;privateint balance;// Constructors, getters, and setters are omitted...publicbooleanhasEnoughFunds(int amount){// (3)return balance >= amount;}publicvoidwithdraw(int amount){// (4)if(!hasEnoughFunds(amount)){thrownew IllegalStateException("Not enough funds!");}this.balance =this.balance - amount;}publicvoiddeposit(int amount){// (5)this.balance =this.balance + amount;}}

publicinterfaceWalletRepositoryextendsReactiveMongoRepository<Wallet, ObjectId>{// (1)
   Mono<Wallet>findByOwner(Mono<String> owner);// (2)}

publicinterfaceWalletService{

   Mono<TxResult>transferMoney(// (1)
      Mono<String> fromOwner,
      Mono<String> toOwner,
      Mono<Integer> amount);

   Mono<Statistics>reportAllWallets();// (2)enum TxResult {// (3)
      SUCCESS,
      NOT_ENOUGH_FUNDS,
      TX_CONFLICT
   }classStatistics{// (4)// Implementation is omitted ...}}

public Mono<OperationStats>runSimulation(){return Flux.range(0, iterations)// (1).flatMap(i -> Mono
         .delay(Duration.ofMillis(rnd.nextInt(10)))// (2).publishOn(simulationScheduler)// (3).flatMap(_i ->{
            String fromOwner =randomOwner();// (4)
            String toOwner =randomOwnerExcept(fromOwner);//int amount =randomTransferAmount();//return walletService.transferMoney(// (5)
               Mono.just(fromOwner),
               Mono.just(toOwner),
               Mono.just(amount));})).reduce(// (6)
         OperationStats.start(),
         OperationStats::countTxResult);}

publicclassTransactionalWalletServiceimplementsWalletService{privatefinal ReactiveMongoTemplate mongoTemplate;// (1)@Overridepublic Mono<TxResult>transferMoney(// (2)
      Mono<String> fromOwner,//
      Mono<String> toOwner,//
      Mono<Integer> requestAmount                                 //){//return Mono.zip(fromOwner, toOwner, requestAmount)// (2.1).flatMap(function((from, to, amount)->{// (2.2)returndoTransferMoney(from, to, amount)// (2.3).retryBackoff(// (2.4)20, Duration.ofMillis(1),//
                  Duration.ofMillis(500),0.1//)//.onErrorReturn(TxResult.c);// (2.5)}));}private Mono<TxResult>doTransferMoney(// (3)
      String from, String to, Integer amount                      // (3.1)){//return mongoTemplate.inTransaction().execute(session ->// (3.2)
         session                                                  //.findOne(queryForOwner(from), Wallet.class)// (3.3).flatMap(fromWallet -> session                        //.findOne(queryForOwner(to), Wallet.class)// (3.4).flatMap(toWallet ->{//if(fromWallet.hasEnoughFunds(amount)){// (3.5)
                     fromWallet.withdraw(amount);// (3.6)
                     toWallet.deposit(amount);// (3.7)return session.save(fromWallet)// (3.8).then(session.save(toWallet))// (3.9).then(Mono.just(TxResult.SUCCESS));// (3.10)}else{//return Mono.just(TxResult.NOT_ENOUGH_FUNDS);// (3.11)}//})))//.onErrorResume(e ->// (3.12)
             Mono.error(newRuntimeException("Conflict")))//.last();// (3.13)}private Query queryForOwner(String owner){// (4)return Query.query(newCriteria("owner").is(owner));// (4.1)}}

The number of accounts in the system:500
The total balance of the system before the simulation:500,000$
Running the money transferring simulation(10,000 iterations)...
The simulation is finished
Transfer operations statistic:- successful transfer operations:6,238- not enough funds:3,762- conflicts:0
All wallet operations:- total withdraw operations:6,238- total deposit operations:6,238
The total balance of the system after the simulation:500,000$

在撰写本文时，Spring Data 2.1 已为四个 NoSQL 数据库（即 MongoDB、Cassandra、Couchbase 和 Redis）提供 database 连接器. Spring Data 很可能也会支持其他数据存储，尤其是那些利用 Spring WebFlux WebClient 通过 HTTP 进行通信的数据存储。

在这里，我们不会介绍 Spring Data 反应式连接器的所有 特性 或其实现细节。在前面的部分中，我们介绍了 MongoDB 和 Cassandra 的很多内容。但是，让我们重点介绍每个反应式连接器的主要显着特征。

由于反应式连接领域相对较新，禁止在许多应用程序中使用该方法存在一些限制：

但是，我们可能会看到越来越多的 NoSQL 解决方案提供了响应式驱动程序，或者至少提供了易于包装到响应式 API 中的异步驱动程序。此外，目前正在对 Java 中用于数据访问的语言级 API 领域进行重大改进。在撰写本文时，有两个突出的建议可以满足这一需求，即 ADBA 和 R2DBC。现在，让我们更仔细地看看它们。

异步数据库访问（ADBA） 定义了一个非用于 Java 平台的阻塞数据库访问 API。在撰写本文时，它仍是一个草案， JDBC 专家组 正在讨论它应该是什么样子。 ADBA 是在 JavaOne 2016 会议上宣布的，并且已经讨论 几年了。 ADBA 旨在补充当前的 JDBC API，并提出一种面向高吞吐量程序的异步替代方案（而不是替代方案）。 ADBA 旨在支持流畅的编程风格，并为编写数据库查询提供构建器模式。 ADBA 不是 JDBC 的扩展，并且不依赖于它。准备就绪后，ADBA 很可能将存在于 java.sql2 包中。

ADBA 它是一个异步 API，因此在进行网络调用时不应阻塞任何方法调用。所有潜在的阻塞操作都表示为单独的 ADBA 操作。客户端应用程序构建并提交一个 ADBA 操作或 ADBA 操作图。实现 ADBA 的驱动程序异步执行操作并通过java.util.concurrent.CompletionStage或回调报告结果。准备就绪后，通过 ADBA 请求问题的异步 SQL 可能如下所示：

CompletionStage<List<String>> employeeNames =// (1)
connection
   .<Integer>rowOperation("select * from employee")// (2).onError(this::userDefinedErrorHandler)// (3).collect(Collector.of(// (4)()->newArrayList<String>(),(ArrayList<String> cont, Result.RowColumn row)->
          cont = cont.add(row.at("name").get(String.class)),(l, r)-> l,
      container -> container)).submit()// (5).getCompletionStage();// (6)

请注意，前面的代码仍然基于 ADBA 草案，因此 API 可能随时更改。但是，编号的行表示以下含义：

当然，ADBA 将提供编写和执行 更复杂的数据库查询的能力，包括条件查询、并行查询和依赖查询操作。 ADBA 支持事务。但是，与 JDBC 相比，ADBA 并非旨在由业务代码直接使用（即使有可能），而是旨在为更高级的库和框架提供异步基础。

在撰写本文时，ADBA 只有一个实现，称为 AoJ。 AoJ (https://gitlab.com/asyncjdbc/asyncjdbc/tree/master< /a>) 是一个 experimental 库，它通过在单独的线程池上调用标准 JDBC 来实现 ADBA API 的子集. AoJ 不适合生产使用，但提供了玩 ADBA 的能力，而无需实现成熟的异步驱动程序。

有一些关于 ADBA 不仅可以通过 CompletionStage 而且还可以通过响应式 Publisher 返回结果的能力的对话。来自 Java Flow API。然而，目前尚不清楚 Flow API 将如何集成到 ADBA 中，或者 ADBA 是否会公开一个反应式 API。在撰写本文时，这个话题仍然是热议的话题。

在这一点上，我们必须再次声明，Java 的CompletionStage 所代表的异步行为可能总是被响应式Publisher 实现所替代。但是，这种说法对相反的情况无效。反应性行为可以用CompletionStage或 CompletableFuture 仅通过一些折衷来表示，即通过降低背压传播。此外，对于 CompletionStage<List<T>>，没有实际的数据流语义，因为客户端需要等待完整的结果集。利用 Collector API 进行流式传输似乎并不是一个真正的选择。此外， CompletableFuture 一旦提交就开始执行，而 Publisher 只有在提交时才开始执行它收到订阅。用于数据库访问的反应式 API 将同时满足反应式语言级 API 和异步语言级 API 的需求。这是因为任何响应式 API 都可以快速转换为异步 API，而无需任何语义妥协。但是，在大多数情况下，异步 API 可能会成为反应式 API，但需要做出一些妥协。这就是为什么从本书作者的角度来看，支持 Reactive Streams 的 ADBA 似乎比只支持异步的 ADBA 更有益。

Java 中下一代数据访问 API 的替代候选者称为 R2DBC。它提供的不仅仅是完全异步的 ADBA，并且证明了用于关系数据访问的反应式 API 具有巨大的潜力。所以，让我们更仔细地看看它。

反应式关系数据库连接 (R2DBC) (https://github.com/r2dbc) 是一个探索完全反应式数据库 API 可能看起来像。 Spring Data 团队领导 R2DBC 计划，并使用它在 reactive 应用程序中的响应式数据访问上下文中探索和验证想法. R2DBC 在 Spring OnePlatform 2018 会议上公开宣布。 R2DBC 的目标是定义一个响应式 database 访问 API，支持背压。 Spring Data 团队在响应式 NoSQL 持久性方面获得了一些出色的经验，因此他们决定提出他们对真正响应式语言级数据访问 API 的设想。此外，R2DBC 可能成为 Spring Data 中关系响应式存储库的底层 API。在撰写本文时，R2DBC 仍处于试验阶段，尚不清楚它是否或何时可能成为生产就绪软件。

R2DBC 项目由以下部分组成：

R2DBC 客户端允许使用 PostgreSQL 数据库，如下所示：

PostgresqlConnectionFactory pgConnectionFactory =// (1)
    PostgresqlConnectionConfiguration.builder().host("<host>").database("<database>").username("<username>").password("<password>").build();

R2dbc r2dbc =newR2dbc(pgConnectionFactory);// (2)

r2dbc.inTransaction(handle ->// (3)
   handle
      .execute("insert into book (id, title, publishing_year) "+// (3.1)"values ($1, $2, $3)",20,"The Sands of Mars",1951).doOnNext(n-> log.info("{} rows inserted", n))// (3.2)).thenMany(r2dbc.inTransaction(handle ->// (4)
      handle.select("SELECT title FROM book")// (4.1).mapResult(result ->// (4.2)
            result.map((row, rowMetadata)->// (4.3)
               row.get("title", String.class)))))// (4.4).subscribe(elem -> log.info(" - Title: {}", elem));// (5)

让我们描述一下前面的代码：

正如我们所见，R2DBC Client 提供了一个具有响应式风格的流畅 API。这个 API 在响应式应用程序的代码库中感觉非常自然。

publicinterfaceBookRepositoryextendsReactiveCrudRepository<Book, Integer>{@Query("SELECT * FROM book WHERE publishing_year = "+"(SELECT MAX(publishing_year) FROM book)")
   Flux<Book>findTheLatestBooks();}

BookRepository createRepository(PostgresqlConnectionFactory fct){// (1)
   TransactionalDatabaseClient txClient =// (2)
      TransactionalDatabaseClient.create(fct);
   RelationalMappingContext cnt =newRelationalMappingContext();// (3)returnnewR2dbcRepositoryFactory(txClient, cnt)// (4).getRepository(BookRepository.class);// (5)}

bookRepository.findTheLatestBooks().doOnNext(book -> log.info("Book: {}", book)).count().doOnSuccess(count -> log.info("DB contains {} latest books", count)).subscribe();

David Moten 的 rxjava2-jdbc 库 (https ://github.com/davidmoten/rxjava2-jdbc) 的创建是为了以不阻塞 reactive 应用程序。该库基于 RxJava 2 构建，使用专用线程池和非阻塞连接池的概念。因此，请求在等待空闲连接时不会阻塞线程。一旦连接可用，查询就会在连接上开始执行 并阻塞线程。应用程序可能不会管理用于阻塞请求的专用调度程序，因为库会这样做。此外，该库有一个流畅的 DSL，它允许我们发出 SQL 语句并以反应流的形式接收结果。 让我们定义 Book 实体并正确注释它以与 rxjava2-jdbc:

@Query("select id, title, publishing_year "+// (3)"from book order by publishing_year")publicinterfaceBook{// (1)@Column String id();// (2)@Column String title();@Column Integer publishing_year();}

在前面的代码中，我们可以看到以下内容：

现在让我们定义一个简单的 repository 来查找在特定时期出版的书籍：

publicclassRxBookRepository{privatestaticfinal String SELECT_BY_YEAR_BETWEEN =// (1)"select * from book where "+"publishing_year >= :from and publishing_year <= :to";privatefinal String url ="jdbc:h2:mem:db";privatefinalint poolSize =25;privatefinal Database database = Database.from(url, poolSize);// (2)public Flowable<Book>findByYearBetween(// (3)
      Single<Integer> from,
      Single<Integer> to
   ){return Single
         .zip(from, to, Tuple2::new)// (3.1).flatMapPublisher(tuple -> database                       //.select(SELECT_BY_YEAR_BETWEEN)// (3.2).parameter("from", tuple._1())// (3.3).parameter("to", tuple._2())//.autoMap(Book.class));// (3.5)}}

我们来描述RxBookRepository类的实现如下：

rxjava2-jdbc 库支持 大多数 JDBC 驱动程序。此外，该库对事务有一些支持。事务中的所有操作都必须在同一个连接上执行。事务的提交/回滚自动发生。

rxjava2-jdbc 库很简洁，减少了一些潜在的线程块，并且可以响应式地使用关系数据库。然而，到目前为止，它仍然是新的，可能无法处理复杂的反应式工作流程，尤其是那些涉及事务的工作流程。 rxjava2-jdbc 库也需要所有 SQL 查询的定义。

有时我们可能已经有一个CrudRepository实例，其中包含所有必需的机制数据访问（无需手动查询或实体映射）。但是，我们不能直接在反应式应用程序中使用它。在这种情况下，很容易编写我们自己的反应式适配器，它的行为类似于 rxjava2-jdbc 库，但在存储库级别。应用此方法时请谨慎使用 JPA。使用延迟加载时，我们很快就会遇到代理问题。因此，假设我们有以下由 JPA 定义的Book实体：

@Entity@Table(name ="book")publicclassBook{@Id@GeneratedValue(strategy = GenerationType.IDENTITY)private Integer id;private String title;private Integer publishingYear;// Constructors, getters, setters...}

此外，我们还有以下 Spring Data JPA 存储库：

@RepositorypublicinterfaceBookJpaRepositoryextendsCrudRepository<Book, Integer>{// (1)

   Iterable<Book>findByIdBetween(int lower,int upper);// (2)@Query("SELECT b FROM Book b WHERE "+"LENGTH(b.title)=(SELECT MIN(LENGTH(b2.title)) FROM Book b2)") // (3)
   Iterable<Book>findShortestTitle();}

BookJpaRepository具有以下特征：

BookJpaRepository接口可以很好地与阻塞 JPA 基础设施配合使用。 BookJpaRepositoryrepository 的所有方法都返回非反应类型。要将BookJpaRepository接口包装到反应式API中并接收其大部分功能，我们可以定义一个抽象适配器并使用其他方法对其进行扩展以映射findByIdBetween和findShortestTitle方法。抽象适配器可以被重用于适配任何CrudRepository实例。适配器可能如下所示：

publicabstractclassReactiveCrudRepositoryAdapter
                <T, ID, I extendsCrudRepository<T, ID>>// (1)implementsReactiveCrudRepository<T, ID>{//protectedfinal I delegate;// (2)protectedfinal Scheduler scheduler;//// Constructor...@Overridepublic<S extendsT> Mono<S>save(S entity){// (3)return Mono                                                  //.fromCallable(()-> delegate.save(entity))// (3.1).subscribeOn(scheduler);// (3.2)}@Overridepublic Mono<T>findById(Publisher<ID> id){// (4)return Mono.from(id)// (4.1).flatMap(actualId ->// (4.2)
            delegate.findById(actualId)// (4.3).map(Mono::just)// (4.4).orElseGet(Mono::empty))// (4.5)
.subscribeOn(scheduler);// (4.6)}@Overridepublic Mono<Void>deleteAll(Publisher<?extendsT> entities){  // (5)return Flux.from(entities)// (5.1).flatMap(entity -> Mono                                   //.fromRunnable(()-> delegate.delete(entity))// (5.2).subscribeOn(scheduler))// (5.3).then();// (5.4)}// All other methods of ReactiveCrudRepository...}

让我们描述一下前面的代码：

现在，让我们在具体的 reactive 存储库实现中定义缺少的自定义方法：

publicclassRxBookRepositoryextendsReactiveCrudRepositoryAdapter<Book, Integer, BookJpaRepository>{publicRxBookRepository(
      BookJpaRepository delegate,
      Scheduler scheduler
   ){super(delegate, scheduler);}public Flux<Book>findByIdBetween(// (1)
      Publisher<Integer> lowerPublisher,                           //
      Publisher<Integer> upperPublisher                            //
   ){                                                             //return Mono.zip(// (1.1)
         Mono.from(lowerPublisher),                                //
         Mono.from(upperPublisher)                                 //).flatMapMany(                                               //function((low, upp)->// (1.2)
            Flux                                                   //
               .fromIterable(delegate.findByIdBetween(low, upp))// (1.3).subscribeOn(scheduler)// (1.4)))                                                        //.subscribeOn(scheduler);// (1.5)}public Flux<Book>findShortestTitle(){// (2)return Mono.fromCallable(delegate::findShortestTitle)// (2.1).subscribeOn(scheduler)// (2.2).flatMapMany(Flux::fromIterable);// (2.3)}}

RxBookRepository类扩展了抽象ReactiveCrudRepositoryAdapter类，引用了BookJpaRepository和< code class="literal">Scheduler实例，并定义如下方法：

现在，我们最终可以使用以下代码将阻塞BookJpaRepository包装到响应式RxBookRepository中：

Scheduler scheduler = Schedulers.newParallel("JPA",10);
BookJpaRepository jpaRepository =getBlockingRepository(...);

RxBookRepository rxRepository =newRxBookRepository(jpaRepository, scheduler);

Flux<Book> books = rxRepository
   .findByIdBetween(Mono.just(17), Mono.just(22));

books
   .subscribe(b -> log.info("Book: {}", b));

并非所有阻塞特征都可以如此轻松地映射。例如，JPA 延迟加载 很可能 被描述的方法破坏。此外，事务的支持将需要类似于 rxjava2-jdbc 库中的额外工作。或者，我们需要在 granularity 包装同步操作，其中没有事务扩展超过一个阻塞调用。

此处描述的方法不会神奇地将阻塞请求转换为反应式非阻塞执行。一些构成JPA调度器的线程 仍然 被阻塞。然而，对调度程序的详细监控和明智的池管理应该有助于在应用程序的性能和资源使用之间建立一个可接受的平衡。