Tenormis's Blog

Appearance

Spring In Action 6th：响应式持久化数据 ​

1. 使用 R2DBC ​

R2DBC（Reactive Relational Database Connectivity）是一种使用响应式类型处理关系数据的相对较新的选项。它实际上是 JDBC 的一种响应式替代方案，可以在诸如 MySQL、PostgreSQL、H2 和 Oracle 等传统关系数据库中实现非阻塞持久化。由于它构建在 Reactive Streams 上，因此它与 JDBC 非常不同，并且是一个独立的规范，与 Java SE 无关。

Spring Data R2DBC 是 Spring Data 的一个子项目，为 R2DBC 提供自动存储库支持，类似于我们在《Spring In Action 6th：处理数据》中介绍的 Spring Data JDBC。然而，与 Spring Data JDBC 不同，Spring Data R2DBC 不要求严格遵循领域驱动设计的概念。实际上，正如你即将看到的，通过聚合根持久化数据在 Spring Data R2DBC 中需要比在 Spring Data JDBC 中更多的工作。

要使用 Spring Data R2DBC，你需要在项目构建中添加一个启动器依赖。对于一个使用 Maven 构建的项目，依赖关系如下：

XML
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-r2dbc</artifactId>
</dependency>

你还需要一个关系数据库来持久化数据，以及相应的 R2DBC 驱动程序。对于我们的项目，我们将使用内存中的 H2 数据库。因此，我们需要添加两个依赖项：H2 数据库库本身和 H2 R2DBC 驱动程序。以下是 Maven 依赖项：

XML
<dependency>
    <groupId>com.h2database</groupId>
    <artifactId>h2</artifactId>
    <scope>runtime</scope>
</dependency>
<dependency>
    <groupId>io.r2dbc</groupId>
    <artifactId>r2dbc-h2</artifactId>
    <scope>runtime</scope>
</dependency>

如果你使用的是不同的数据库，那么你需要添加相应的 R2DBC 驱动程序依赖项，以适配你选择的数据库。

现在依赖项已经就位，让我们看看 Spring Data R2DBC 是如何工作的。我们从定义领域实体开始。

1.1. 为 R2DBC 定义实体 ​

为了了解 Spring Data R2DBC，我们将仅重新创建 Taco Cloud 应用程序的持久层，仅关注于对 taco 和 order 数据进行持久化所必需的组件。这包括为 TacoOrder、Taco 和 Ingredient 创建领域实体，以及相应的每个实体的存储库。

我们将首先创建的领域实体类是 Ingredient 类。它将类似于下面的代码清单：

Java
@Data
@NoArgsConstructor
@RequiredArgsConstructor
@EqualsAndHashCode(exclude = "id")
public class Ingredient {
    @Id
    private Long id;
    private @NonNull String slug;
    private @NonNull String name;
    private @NonNull Type type;

    public enum Type {
        WRAP, PROTEIN, VEGGIES, CHEESE, SAUCE
    }
}

Note：这里 Long ID 的 @Id 注解是由 org.springframework.data.annotation 包中提供的。

正如你所看到的，这与我们之前创建的 Ingredient 类的其他版本并没有太大的不同。请注意以下两个显著的差异：

  • Spring Data R2DBC 要求属性具有 Setter 方法，因此大多数属性必须是非 final 的。但为了帮助 Lombok 创建一个必需的参数构造函数，我们使用 @NonNull 注解大多数属性。这将导致 Lombok 和 @RequiredArgsConstructor 注解将这些属性包含在构造函数中；

  • 在通过 Spring Data R2DBC 存储库保存对象时，如果对象的 ID 属性非空，则将其视为更新。在 Ingredient 的情况下，id 属性先前被定义为 String 并在创建时指定。但是在 Spring Data R2DBC 中这样做会导致错误。因此，我们将该 String ID 转移到一个名为 slug 的新属性中，它只是 Ingredient 的伪 ID，并使用由数据库生成的 Long ID 属性；

相应的数据库表在 schema.sql 中定义如下：

SQL
create table Ingredient (
  id   identity
, slug varchar(4)  not null
, name varchar(25) not null
, type varchar(10) not null
);

Taco 实体类与其 Spring Data JDBC 的对应类相似，如下所示：

Java
@Data
@NoArgsConstructor
@RequiredArgsConstructor
public class Taco {
    @Id
    private Long id;
    private @NonNull String name;
    private Set<Long> ingredientIds = new HashSet<>();

    public void addIngredient(Ingredient ingredient) {
        ingredientIds.add(ingredient.getId());
    }
}

Note：同样这里 Long ID 的 @Id 注解也是由 org.springframework.data.annotation 包中提供的，后续不再重复赘述。

与 Ingredient 类一样，实体字段必须具有 Setter 方法，因此使用了 @NonNull 而不是 final。

但特别有趣的是，Taco 不是有一个 Ingredient 对象的集合，而是有一个 Set<Long>，引用了属于这个 taco 的 Ingredient 对象的 ID。选择使用 Set 而不是 List 是为了确保唯一性。但为什么我们必须使用 Set<Long> 而不是 Set<Ingredient> 作为 ingredient 集合呢？

与其他 Spring Data 项目不同，Spring Data R2DBC 目前不支持实体之间的直接关系（至少在目前还不支持）。作为一个相对较新的项目，Spring Data R2DBC 仍在努力解决以非阻塞方式处理关系的一些挑战。这可能会在将来的 Spring Data R2DBC 版本中发生变化。

在那之前，我们不能让 Taco 引用一个 Ingredient 集合并期望持久层能正常工作。相反，在处理关系时，我们有以下选择：

  • 定义具有对相关对象的 ID 引用的实体。在这种情况下，数据库表中的相应列必须尽可能地定义为数组类型。H2 和 PostgreSQL 是支持数组列的两个数据库，但许多其他数据库不支持。此外，即使数据库支持数组列，可能也无法将条目定义为对所引用表的外键，从而无法强制执行引用完整性；

  • 定义实体及其相应的表以完全匹配彼此。对于集合，这意味着引用的对象将具有映射回引用表的列。例如，Taco 对象的表需要有一列指向 TacoOrder；

  • 将引用实体序列化为 JSON 并将 JSON 存储在一个大的 VARCHAR 列中。如果不需要通过引用对象进行查询，这种方法特别有效。然而，由于相应 VARCHAR 列的长度限制，JSON 序列化对象的大小可能有潜在限制。此外，我们无法利用数据库架构来保证引用完整性，因为引用的对象将作为一个简单的字符串值存储（其中可能包含任何内容）；

尽管这些选项都不是理想的，但在权衡之后，我们将为 Taco 对象使用第一种方式。Taco 类有一个 Set<Long>，引用一个或多个 Ingredient 的 ID。这意味着相应的表必须有一个数组列来存储这些 ID。对于 H2 数据库，Taco 表的定义如下：

SQL
create table Taco (
  id             identity
, name           varchar(50) not null
, ingredient_ids array
);

在 ingredient_ids 列上使用的数组类型是特定于 H2 的。对于 PostgreSQL，该列可能被定义为 integer[]。请参考所选择数据库的文档，了解如何定义数组列。请注意，并非所有数据库实现都支持数组列，因此你可能需要选择建模关系的其他选项之一。

最后，TacoOrder 类如下所示：

Java
@Data
public class TacoOrder {
    @Id
    private Long id;
    private String deliveryName;
    private String deliveryStreet;
    private String deliveryCity;
    private String deliveryState;
    private String deliveryZip;
    private String ccNumber;
    private String ccExpiration;
    private String ccCVV;
    private Set<Long> tacoIds = new LinkedHashSet<>();
    private List<Taco> tacos = new ArrayList<>();

    public void addTaco(Taco taco) {
        this.tacos.add(taco);
    }
}

正如你所看到的，除了具有一些额外的属性之外，TacoOrder 类遵循与 Taco 类相同的模式。它通过一个 Set<Long> 引用其子 Taco 对象。

然而，稍后我们将看到如何将完整的 Taco 对象放入 TacoOrder，即使 Spring Data R2DBC 不直接支持这种方式的关系。Taco_Order 表的数据库表结构如下：

SQL
create table Taco_Order (
  id              identity
, delivery_name   varchar(50) not null
, delivery_street varchar(50) not null
, delivery_city   varchar(50) not null
, delivery_state  varchar(2)  not null
, delivery_zip    varchar(10) not null
, cc_number       varchar(16) not null
, cc_expiration   varchar(5)  not null
, cc_cvv          varchar(3)  not null
, taco_ids        array
);

就像 Taco 表一样，Taco-Order 表通过一个定义为数组列的 taco_ids 列引用其子 Taco。再次强调，这个表结构是为 H2 数据库设计的。请查阅你的数据库文档，了解有关支持和创建数组列的详细信息。

Note：以上数据库表结构的定义脚本位于根目录下的 schema.sql 文件（在项目中位于 src/main/resources 下）。如果你希望在数据库初始化过程中包含其他 SQL 脚本，可以在调用 populator.addPopulators() 时添加更多的 Resource-DatabasePopulator 对象。

现在我们已经定义了我们的实体和它们对应的数据库表结构，让我们创建仓库，通过这些仓库我们将保存和获取 taco 数据。

1.2. 定义响应式 Repository ​

在《Spring In Action 6th：处理数据》和《Spring In Action 6th：处理非关系型数据》中，我们将我们的仓库定义为扩展 Spring Data 的 CrudRepository 接口的接口。但是，那个基础仓库接口处理的是单个对象和 Iterable 集合。相比之下，我们期望一个响应式仓库会处理 Mono 和 Flux 对象。

这就是为什么 Spring Data 提供 ReactiveCrudRepository 来定义响应式仓库。ReactiveCrudRepository 的操作非常像 CrudRepository。要创建一个仓库，定义一个扩展 ReactiveCrudRepository 的接口，例如这样：

Java
public interface OrderRepository
    extends ReactiveCrudRepository<TacoOrder, Long> {
}

表面上，这个 OrderRepository 和我们之前的仓库之间的唯一区别是它扩展了 ReactiveCrudRepository 而不是 CrudRepository。但是，显著的不同之处在于，它的方法返回 Mono 和 Flux 类型，而不是单个 TacoOrder 或 Iterable<TacoOrder>。两个例子包括 findById() 方法，它返回一个 Mono<TacoOrder>，和 findAll()，它返回一个 Flux<TacoOrder>。

为了看到这个响应式仓库可能如何在实践中工作，假设你想获取所有的 TacoOrder 对象，并将它们的 deliveryName 打印到标准输出。在这种情况下，你可能会写一些像如下片段那样的代码：

Java
@Autowired
OrderRepository orderRepository;

...

orderRepository.findAll()
    .doOnNext(order -> {
        System.out.println("Deliver to: " + order.getDeliveryName());
    })
    .subscribe();

在这里，对 findAll() 的调用返回了一个 Flux<TacoOrder>，我们在其上添加了一个 doOnNext() 来打印 deliveryName。最后，调用 subscribe() 启动了数据通过 Flux 的流动。

在《Spring In Action 6th：处理数据》的 Spring Data JDBC 示例中，TacoOrder 是聚合根，Taco 是该聚合中的子元素。因此，Taco 对象作为 TacoOrder 的一部分被持久化，没有必要定义一个专门用于 Taco 持久化的仓库。但是，Spring Data R2DBC 不以这种方式支持适当的聚合根，所以我们需要一个 TacoRepository，通过它来持久化 Taco 对象：

Java
public interface TacoRepository
    extends ReactiveCrudRepository<Taco, Long> {
}

如你所见，TacoRepository 和 OrderRepository 没有太大的区别，它扩展了 ReactiveCrudRepository。另一边，IngredientRepository 则稍微有些有趣，如下所示：

Java
public interface IngredientRepository
        extends ReactiveCrudRepository<Ingredient, Long> {
    Mono<Ingredient> findBySlug(String slug);
}

与我们的其他两个响应式仓库一样，IngredientRepository 扩展了 ReactiveCrudRepository。但是，因为我们可能需要一种根据 slug 值查找 Ingredient 对象的方法，所以 IngredientRepository 包含了一个返回 Mono<Ingredient> 的 findBySlug() 方法。

现在让我们看看如何编写测试来验证我们的仓库是否工作正常。

1.3. 测试 R2DBC Repository ​

Spring Data R2DBC 包含了对 R2DBC 仓库编写集成测试的支持。具体来说，当 @DataR2dbcTest 注解被放置在一个测试类上时，Spring 会创建一个应用程序上下文，其中生成的 Spring Data R2DBC 仓库作为可以注入到测试类中的 bean。结合我们在前面的章节中使用过的 StepVerifier，这使我们能够针对我们创建的所有仓库编写自动化测试。

为了简洁，我们将仅关注一个单独的测试类：IngredientRepositoryTest。这将测试 IngredientRepository，验证它是否可以保存 Ingredient 对象，获取单个 Ingredient，以及获取所有保存的 Ingredient 对象。代码如下所示：

Java
@DataR2dbcTest
public class IngredientRepositoryTest {
    @Autowired
    IngredientRepository ingredientRepo;

    @BeforeEach
    public void setup() {
        Flux<Ingredient> deleteAndInsert = ingredientRepo.deleteAll()
                .thenMany(ingredientRepo.saveAll(
                        Flux.just(
                                new Ingredient("FLTO", "Flour Tortilla", Type.WRAP),
                                new Ingredient("GRBF", "Ground Beef", Type.PROTEIN),
                                new Ingredient("CHED", "Cheddar Cheese", Type.CHEESE)
                        )));
        StepVerifier.create(deleteAndInsert)
                .expectNextCount(3)
                .verifyComplete();
    }

    @Test
    public void shouldSaveAndFetchIngredients() {
        StepVerifier.create(ingredientRepo.findAll())
                .recordWith(ArrayList::new)
                .thenConsumeWhile(x -> true)
                .consumeRecordedWith(ingredients -> {
                    assertThat(ingredients).hasSize(3);
                    assertThat(ingredients).contains(
                            new Ingredient("FLTO", "Flour Tortilla", Type.WRAP));
                    assertThat(ingredients).contains(
                            new Ingredient("GRBF", "Ground Beef", Type.PROTEIN));
                    assertThat(ingredients).contains(
                            new Ingredient("CHED", "Cheddar Cheese", Type.CHEESE));
                })
                .verifyComplete();
        StepVerifier.create(ingredientRepo.findBySlug("FLTO"))
                .assertNext(ingredient -> {
                    ingredient.equals(new Ingredient("FLTO", "Flour Tortilla",
                            Type.WRAP));
                });
    }
}

虽然我们只关注了测试 IngredientRepository，但同样的技术可以用来测试任何 Spring Data R2BDC 生成的仓库。

到目前为止，一切都好。我们现在已经定义了我们的领域类型和它们各自的仓库。我们还编写了一个测试来验证它们是否工作。如果我们愿意，我们可以按原样使用它们。但是，这些仓库使 TacoOrder 的持久化不便，因为我们必须首先创建和持久化作为该订单一部分的 Taco 对象，然后持久化引用子 Taco 对象的 TacoOrder 对象。当读取 TacoOrder 时，我们只会收到一组 Taco ID，而不是完全定义的 Taco 对象。

如果我们能将 TacoOrder 作为聚合根进行持久化，并且它的子 Taco 对象也能随之持久化，那就太好了。同样，如果我们能获取一个 TacoOrder 并且它完全定义了完整的 Taco 对象，而不仅仅是 ID，那就太好了。让我们定义一个位于 OrderRepository 和 TacoRepository 前面的服务级类，以模仿之前 OrderRepository 的持久化行为。

1.4. 定义 OrderRepository 聚合根服务 ​

将 TacoOrder 和 Taco 对象一起持久化的第一步是在 TacoOrder 类中添加一个 Taco 集合属性，这样 TacoOrder 就成为了聚合根。下面将展示这一步骤。

Java
@Data
public class TacoOrder {
    ...

    @Transient
    private transient List<Taco> tacos = new ArrayList<>();

    public void addTaco(Taco taco) {
        this.tacos.add(taco);
        if (taco.getId() != null) {
            this.tacoIds.add(taco.getId());
        }
    }
}

除了在 TacoOrder 类中添加一个名为 tacos 的新 List<Taco> 属性外，addTaco() 方法现在将给定的 Taco 添加到该列表中（同时也将其 id 添加到 tacoIds 集合中）。

然而，需要注意的是，tacos 属性被注解为 @Transient（同时也被标记为 Java 的 transient 关键字）。这表明 Spring Data R2DBC 不应尝试持久化此属性。如果没有 @Transient 注解，Spring Data R2DBC 会尝试持久化它，并由于不支持此类关系而导致错误。

当 TacoOrder 被保存时，只有 tacoIds 属性会被写入数据库，tacos 属性将被忽略。即便如此，至少现在 TacoOrder 有一个地方可以存放 Taco 对象。这将在保存 TacoOrder 时保存 Taco 对象以及在获取 TacoOrder 时读取 Taco 对象时派上用场。

现在我们可以创建一个服务 bean，该服务 bean 可以保存和读取 TacoOrder 对象以及它们各自的 Taco 对象。让我们从保存 TacoOrder 开始。以下示例代码中定义的 TacoOrderAggregateService 类有一个 save() 方法：

Java
@Service
@RequiredArgsConstructor
public class TacoOrderAggregateService {
    private final TacoRepository tacoRepo;
    private final OrderRepository orderRepo;

    public Mono<TacoOrder> save(TacoOrder tacoOrder) {
        return Mono.just(tacoOrder)
                .flatMap(order -> {
                    List<Taco> tacos = order.getTacos();
                    order.setTacos(new ArrayList<>());
                    return tacoRepo.saveAll(tacos)
                            .map(taco -> {
                                order.addTaco(taco);
                                return order;
                            })
                            .last();
                })
                .flatMap(orderRepo::save);
    }
}

虽然上述代码行数不多，但在 save() 方法中有很多需要解释的内容。首先，作为参数接收的 TacoOrder 被使用 Mono.just() 方法包装在一个 Mono 中。这使我们能够在 save() 方法的其余部分中将其作为响应式类型进行处理。

我们接下来要做的是对我们刚刚创建的 Mono<TacoOrder> 应用 flatMap()。map() 和 flatMap() 都是对通过 Mono 或 Flux 传递的数据对象进行转换的选项，但是因为我们在进行转换过程中执行的操作将产生一个 Mono<TacoOrder>，所以 flatMap() 操作确保我们在映射后继续处理一个 Mono<TacoOrder>，而不是一个 Mono<Mono<TacoOrder>>，如果我们使用 map()，就会出现这种情况。

映射的目的是确保 TacoOrder 最终拥有子 Taco 对象的 ID，并在此过程中保存这些 Taco 对象。每个 Taco 对象的 ID 对于新的 TacoOrder 来说最初可能是空的，我们在 Taco 对象被保存后才会知道 ID。

在从 TacoOrder 中获取 List<Taco>（我们将在保存 Taco 对象时使用）后，我们将 tacos 属性重置为一个空列表。我们将用保存后分配了 ID 的新 Taco 对象重建该列表。

在注入的 TacoRepository 上调用 saveAll() 方法保存了我们所有的 Taco 对象。saveAll() 方法返回一个 Flux<Taco>，我们然后通过 map() 方法遍历它。在这种情况下，转换操作是次要的，每个 Taco 对象被添加回 TacoOrder 是主要的。但是为了确保最终在 Flux 上的是 TacoOrder 而不是 Taco，映射操作返回的是 TacoOrder 而不是 Taco。调用 last() 确保我们不会因为映射操作而有重复的 TacoOrder 对象（每个 Taco 一个）。

此时，所有 Taco 对象应该已经被保存并推回到父 TacoOrder 对象中，同时带有它们新分配的 ID。剩下的就是保存 TacoOrder，这就是最后的 flatMap() 调用所做的。再次，我们在这里选择 flatMap() 是为了确保从 OrderRepository.save() 调用返回的 Mono<TacoOrder> 不会被包装在另一个 Mono 中。我们希望我们的 save() 方法返回一个 Mono<TacoOrder>，而不是一个 Mono<Mono<TacoOrder>>。

现在让我们看一下一个将通过其 ID 读取 TacoOrder，并重新构造所有子 Taco 对象的方法。下面的代码示例显示了一个新的 findById() 方法，用于此目的：

Java
public Mono<TacoOrder> findById(Long id) {
    return orderRepo
            .findById(id)
            .flatMap(order -> {
                return tacoRepo.findAllById(order.getTacoIds())
                        .map(taco -> {
                            order.addTaco(taco);
                            return order;
                        })
                        .last();
            });
}

首先，我们通过在 OrderRepository 上调用 findById() 方法来获取 TacoOrder。这将返回一个 Mono<TacoOrder>，然后我们对其进行 flatMap() 转换，将只有 Taco ID 的 TacoOrder 转换为包含完整 Taco 对象的 TacoOrder。

给 flatMap() 方法的 lambda 表达式调用 TacoRepository.findAllById() 方法，一次性获取 tacoIds 属性中引用的所有 Taco 对象。这将产生一个 Flux<Taco>，我们通过 map() 方法遍历它，将每个 Taco 添加到父 TacoOrder 中，就像我们在用 saveAll() 保存所有 Taco 对象后在 save() 方法中所做的一样。

再次，map() 操作更多的是作为遍历 Taco 对象的手段，而不是作为转换。但是给 map() 的 lambda 表达式每次都返回父 TacoOrder，所以我们最终得到的是 Flux<TacoOrder> 而不是 Flux<Taco>。调用 last() 取出 Flux 中的最后一个条目并返回一个 Mono<TacoOrder>，这就是我们从 findById() 方法返回的内容。

如果你还没有进入响应式思维模式，save() 和 findById() 方法中的代码可能会有点混乱。响应式编程需要一种不同的思维方式，一开始可能会让人感到困惑，但是随着你的响应式编程技能的提高，你会认识到它其实是非常优雅的。

像任何其它代码一样，编写测试以确保 TacoOrderAggregateService 按预期工作是个好主意。测试也将作为如何使用 TacoOrderAggregateService 的示例。下面的代码示例显示了一个针对 TacoOrderAggregateService 的测试：

Java
@DataR2dbcTest
@DirtiesContext
public class TacoOrderAggregateServiceTests {
    @Autowired
    TacoRepository tacoRepo;

    @Autowired
    OrderRepository orderRepo;

    TacoOrderAggregateService service;

    @BeforeEach
    public void setup() {
        this.service = new TacoOrderAggregateService(tacoRepo, orderRepo);
    }

    @Test
    public void shouldSaveAndFetchOrders() {
        TacoOrder newOrder = new TacoOrder();
        newOrder.setDeliveryName("Test Customer");
        newOrder.setDeliveryStreet("1234 North Street");
        newOrder.setDeliveryCity("Notrees");
        newOrder.setDeliveryState("TX");
        newOrder.setDeliveryZip("79759");
        newOrder.setCcNumber("4111111111111111");
        newOrder.setCcExpiration("12/24");
        newOrder.setCcCVV("123");
        newOrder.addTaco(new Taco("Test Taco One"));
        newOrder.addTaco(new Taco("Test Taco Two"));
        StepVerifier.create(service.save(newOrder))
                .assertNext(this::assertOrder)
                .verifyComplete();
        StepVerifier.create(service.findById(1L))
                .assertNext(this::assertOrder)
                .verifyComplete();
    }

    private void assertOrder(TacoOrder savedOrder) {
        assertThat(savedOrder.getId()).isEqualTo(1L);
        assertThat(savedOrder.getDeliveryName()).isEqualTo("Test Customer");
        assertThat(savedOrder.getDeliveryName()).isEqualTo("Test Customer");
        assertThat(savedOrder.getDeliveryStreet()).isEqualTo("1234 North Street");
        assertThat(savedOrder.getDeliveryCity()).isEqualTo("Notrees");
        assertThat(savedOrder.getDeliveryState()).isEqualTo("TX");
        assertThat(savedOrder.getDeliveryZip()).isEqualTo("79759");
        assertThat(savedOrder.getCcNumber()).isEqualTo("4111111111111111");
        assertThat(savedOrder.getCcExpiration()).isEqualTo("12/24");
        assertThat(savedOrder.getCcCVV()).isEqualTo("123");
        assertThat(savedOrder.getTacoIds()).hasSize(2);
        assertThat(savedOrder.getTacos().get(0).getId()).isEqualTo(1L);
        assertThat(savedOrder.getTacos().get(0).getName())
                .isEqualTo("Test Taco One");
        assertThat(savedOrder.getTacos().get(1).getId()).isEqualTo(2L);
        assertThat(savedOrder.getTacos().get(1).getName())
                .isEqualTo("Test Taco Two");
    }
}

上述测试类中包含很多行代码，但其中大部分是在 assertOrder() 方法中断言 TacoOrder 的内容。我们在审查这个测试时，将重点关注其他部分。

测试类被注解为 @DataR2dbcTest，以便让 Spring 创建一个应用程序上下文，其中包含我们所有的仓库作为 bean。@DataR2dbcTest 寻找一个被 @SpringBootConfiguration 注解的配置类来定义 Spring 应用程序上下文。

在单模块项目中，被 @SpringBootApplication（本身被 @SpringBootConfiguration 注解）注解的引导类起到了这个作用。但在我们的多模块项目中，这个测试类不在与引导类相同的项目中，所以我们需要一个像这样的简单配置类：

Java
@SpringBootConfiguration
@EnableAutoConfiguration
public class TestConfig {
}

这不仅满足了对一个用 @SpringBootConfiguration 注解的类的需求，而且还启用了自动配置，确保了（包括其他事项在内）仓库实现将被创建。就其本身而言，TacoOrderAggregateServiceTests 应该可以顺利通过。但是在可能在测试运行之间共享 JVM 和 Spring 应用上下文的 IDE 中，运行此测试与其他持久化测试可能会导致冲突的数据被写入内存中的 H2 数据库。这里使用 @DirtiesContext 注解来确保在测试运行之间重置 Spring 应用上下文，从而在每次运行时都有一个新的空的 H2 数据库。

setup() 方法使用注入到测试类中的 TacoRepository 和 OrderRepository 对象创建了一个 TacoOrderAggregateService 实例。TacoOrderAggregateService 被赋值给一个实例变量，以便测试方法可以使用它。

现在我们终于准备好测试我们的聚合服务了。shouldSaveAndFetchOrders() 的前几行构建了一个 TacoOrder 对象，并用一对测试 Taco 对象填充了它。然后通过 TacoOrderAggregateService 的 save() 方法保存 TacoOrder，该方法返回一个表示已保存订单的 Mono<TacoOrder>。使用 StepVerifier，我们断言返回的 Mono 中的 TacoOrder 符合我们的期望，包括它包含子 Taco 对象。

接下来，我们调用服务的 findById() 方法，该方法也返回一个 Mono<TacoOrder>。与调用 save() 一样，使用 StepVerifier 来逐步通过返回的 Mono 中的每个 TacoOrder（应该只有一个），并断言它符合我们的期望。

在两种 StepVerifier 情况下，调用 verifyComplete() 确保 Mono 中没有更多的对象，并且 Mono 是完整的。

值得注意的是，尽管我们可以应用类似的聚合操作来确保 Taco 对象总是包含完全定义的 Ingredient 对象，但我们选择不这样做，因为 Ingredient 是它自己的聚合根，可能被多个 Taco 对象引用。因此，每个 Taco 只会携带一个 Set<Long> 来引用 Ingredient ID，然后可以通过 IngredientRepository 单独查找。

尽管聚合实体可能需要更多的工作，但 Spring Data R2DBC 提供了一种以响应方式处理关系数据的方法。但这并不是 Spring 提供的唯一响应持久化选项。让我们看看如何使用响应式 Spring Data 仓库处理 MongoDB。

2. 使用 MongoDB 响应式保存文档 ​

在《Spring In Action 6th：处理非关系型数据》中我们利用 Spring Data MongoDB 定义了针对 MongoDB 文档数据库的持久化。在本节中，我们将重新探讨如何使用 Spring Data 的响应式支持来实现 MongoDB 的持久化。

首先，你需要创建一个项目，该项目包含 Spring Data Reactive MongoDB 启动器。在 Maven 中具体的依赖项如下：

XML
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
</dependency>

在《Spring In Action 6th：处理非关系型数据》，我们使用 Flapdoodle 嵌入式 MongoDB 数据库进行测试。不幸的是，当面对响应式仓库时，Flapdoodle 的表现并不尽如人意。在运行测试时，你需要有一个实际的 MongoDB 数据库在 27017 端口运行并监听。

现在我们准备开始为响应式 MongoDB 持久化编写代码。我们将从构成我们领域的文档类型开始。

2.1. 定义文档类型 ​

和之前一样，我们需要创建定义我们应用程序领域的类。在我们这样做的时候，我们需要用 Spring Data MongoDB 的 @Document 注解来注解它们，以指示它们是要存储在 MongoDB 中的文档。先从 Ingredient 类开始：

Java
@Data
@AllArgsConstructor
@NoArgsConstructor(access = AccessLevel.PRIVATE, force = true)
@Document
public class Ingredient {
    @Id
    private String id;
    private String name;
    private Type type;

    public enum Type {
        WRAP, PROTEIN, VEGGIES, CHEESE, SAUCE
    }
}

细心的你可能会注意到，这个 Ingredient 类和我们在《Spring In Action 6th：处理非关系型数据》创建的那个是一模一样的。实际上，无论是通过响应式仓库还是非响应式仓库进行持久化，MongoDB 的 @Document 类都是一样的。这意味着 Taco 和 TacoOrder 类也将和我们之前创建的那些一样。但是为了完整性，以及为了你不需要回头去翻阅，我们将在这里重复它们。

接下来是 Taco：

Java
@Data
public class Taco {
    @NotNull
    @Size(min = 5, message = "Name must be at least 5 characters long")
    private String name;

    private Date createdAt = new Date();

    @Size(min = 1, message = "You must choose at least 1 ingredient")
    private List<Ingredient> ingredients = new ArrayList<>();

    public void addIngredient(Ingredient ingredient) {
        this.ingredients.add(ingredient);
    }
}

注意，与 Ingredient 不同，Taco 类没有用 @Document 注解。这是因为它本身并不作为一个文档保存，而是作为 Taco-Order 聚合根的一部分保存。

另一方面，因为 TacoOrder 是一个聚合根，所以它被注解为 @Document，如下一段代码所示：

Java
@Data
@Document
public class TacoOrder implements Serializable {
    private static final long serialVersionUID = 1L;

    @Id
    private String id;

    private Date placedAt = new Date();

    ...

    private List<Taco> tacos = new ArrayList<>();

    public void addTaco(Taco taco) {
        this.tacos.add(taco);
    }
}

再次强调，对于响应式 MongoDB 仓库来说，领域文档类与非响应式仓库没有任何区别。接下来你会看到，响应式 MongoDB 仓库本身与非响应式的仓库只有非常微小的差别。

2.2. 定义响应式 MongoDB Repository ​

现在我们需要定义两个仓库，一个是 TacoOrder 聚合根，另一个是 Ingredient。我们不需要为 Taco 定义仓库，因为它是 TacoOrder 根的子节点。

这里展示的 IngredientRepository 接口，你现在应该已经很熟悉了：

Java
@CrossOrigin(origins = "http://localhost:8080")
public interface IngredientRepository
        extends ReactiveCrudRepository<Ingredient, String> {
}

这个 IngredientRepository 接口与我们在《Spring In Action 6th：处理非关系型数据》定义的那个只有微小的差别，它扩展了 ReactiveCrudRepository 而不是 CrudRepository。而且与我们为 Spring Data R2DBC 持久化创建的那个也只有一点区别，就是它没有包含 findBySlug() 方法。

同样，OrderRepository 几乎与我们在《Spring In Action 6th：处理非关系型数据》创建的也几乎相同，如下所示：

Java
public interface OrderRepository
        extends ReactiveCrudRepository<TacoOrder, String> {
    Flux<TacoOrder> findByUserOrderByPlacedAtDesc(
            User user, Pageable pageable);
}

归根结底，响应式和非响应式 MongoDB 仓库之间的唯一区别在于它们是扩展 ReactiveCrudRepository 还是 CrudRepository。然而，选择扩展 ReactiveCrudRepository，这些仓库的客户端必须准备好处理像 Flux 和 Mono 这样的响应式类型。当我们为响应式仓库编写测试时，这一点就变得显而易见，这就是我们接下来要做的事情。

2.3. 测试响应式 MongoDB Repository ​

编写 MongoDB 仓库测试的关键是用 @DataMongoTest 注解测试类。这个注解执行的功能类似于我们在本章早些时候使用的 @DataR2dbcTest 注解。它确保创建了一个 Spring 应用程序上下文，生成仓库相关的 bean，以便可以注入到测试中。从那里，测试类可以使用这些注入的仓库来设置测试数据，并对数据库执行其他操作。

例如，考虑以下示例中的 IngredientRepositoryTest，它测试了 IngredientRepository，断言 Ingredient 对象可以被写入和从数据库中读取。

Java
@DataMongoTest
public class IngredientRepositoryTest {
    @Autowired
    IngredientRepository ingredientRepo;

    @BeforeEach
    public void setup() {
        Flux<Ingredient> deleteAndInsert = ingredientRepo.deleteAll()
                .thenMany(ingredientRepo.saveAll(
                        Flux.just(
                                new Ingredient("FLTO", "Flour Tortilla", Type.WRAP),
                                new Ingredient("GRBF", "Ground Beef", Type.PROTEIN),
                                new Ingredient("CHED", "Cheddar Cheese", Type.CHEESE)
                        )));
        StepVerifier.create(deleteAndInsert)
                .expectNextCount(3)
                .verifyComplete();
    }

    @Test
    public void shouldSaveAndFetchIngredients() {
        StepVerifier.create(ingredientRepo.findAll())
                .recordWith(ArrayList::new)
                .thenConsumeWhile(x -> true)
                .consumeRecordedWith(ingredients -> {
                    assertThat(ingredients).hasSize(3);
                    assertThat(ingredients).contains(
                            new Ingredient("FLTO", "Flour Tortilla", Type.WRAP));
                    assertThat(ingredients).contains(
                            new Ingredient("GRBF", "Ground Beef", Type.PROTEIN));
                    assertThat(ingredients).contains(
                            new Ingredient("CHED", "Cheddar Cheese", Type.CHEESE));
                })
                .verifyComplete();
        StepVerifier.create(ingredientRepo.findById("FLTO"))
                .assertNext(ingredient -> {
                    ingredient.equals(new Ingredient("FLTO", "Flour Tortilla",
                            Type.WRAP));
                });
    }
}

这个测试与我们在本章早些时候编写的基于 R2DBC 的仓库测试相似，但仍有些许不同。它首先将三个 Ingredient 对象写入数据库。然后，它使用两个 StepVerifier 实例来验证可以通过仓库读取 Ingredient 对象，首先是所有 Ingredient 对象的集合，然后是通过 ID 获取单个 Ingredient。

此外，就像早些时候的基于 R2DBC 的测试一样，@DataMongoTest 注解会寻找一个带有 @SpringBootConfiguration 注解的类来创建应用程序上下文。之前创建的测试在这里也可以工作。

这里独特的是，第一个 StepVerifier 将所有的 Ingredient 对象收集到一个 ArrayList 中，然后断言该 ArrayList 包含每一个 Ingredient。findAll() 方法并不能保证结果文档的一致排序，这使得使用 assertNext() 或 expectNext() 容易失败。通过将所有结果的 Ingredient 对象收集到一个列表中，我们可以断言该列表包含所有三个对象，而不管它们的顺序如何。

对于 OrderRepository 的测试看起来也非常相似，如下所示：

Java
@DataMongoTest
public class OrderRepositoryTest {
    @Autowired
    OrderRepository orderRepo;

    @BeforeEach
    public void setup() {
        orderRepo.deleteAll().subscribe();
    }

    @Test
    public void shouldSaveAndFetchOrders() {
        TacoOrder order = createOrder();
        StepVerifier
                .create(orderRepo.save(order))
                .expectNext(order)
                .verifyComplete();
        StepVerifier
                .create(orderRepo.findById(order.getId()))
                .expectNext(order)
                .verifyComplete();
        StepVerifier
                .create(orderRepo.findAll())
                .expectNext(order)
                .verifyComplete();
    }

    private TacoOrder createOrder() {
        TacoOrder order = new TacoOrder();
        ...
        return order;
    }
}

shouldSaveAndFetchOrders() 方法首先做的是构造一个订单，包括客户和支付信息以及一些 tacos。为了简洁起见，以上省略了 createOrder() 方法的细节。然后，它使用一个 StepVerifier 来保存 TacoOrder 对象，并断言 save() 方法返回保存的 TacoOrder。然后，它尝试通过其 ID 获取订单，并断言它接收到完整的 TacoOrder。最后，它获取所有的 TacoOrder 对象，并断言它应该只有一个且是预期的 TacoOrder。

如前所述，你需要一个可用的 MongoDB 服务器并监听在 27017 端口上才能运行这个测试。Flapdoodle 嵌入式 MongoDB 与响应式仓库不兼容。如果你的机器上安装了 Docker，你可以轻松地启动一个暴露在 27017 端口上的 MongoDB 服务器，像这样：

Bash
$ docker run -p27017:27017 mongo

通过其它方式设置 MongoDB 也是可以的。有关更多详细信息，请参阅 MongoDB 的官方文档。

现在我们已经看到了如何为 R2BDC 和 MongoDB 创建响应式仓库，让我们再看一下 Spring Data 用于响应式持久化的另一个选项：Cassandra。

3. 使用 Cassandra 响应式保存数据 ​

要开始使用 Cassandra 数据库的响应式持久化，你需要将以下启动器依赖项添加到你的项目构建中。这个依赖项取代了我们之前使用的任何 Mongo 或 R2DBC 的依赖项。

XML
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-cassandra-reactive</artifactId>
</dependency>

然后，你需要声明一些关于 Cassandra 键空间的细节以及如何管理 schema。在你的 application.yml 文件中，添加以下几行：

YAML
spring:
  data:
    rest:
      base-path: /data-api
    cassandra:
      keyspace-name: tacocloud
      schema-action: recreate
      local-datacenter: datacenter1

这是我们在《Spring In Action 6th：处理非关系型数据》中使用非响应式 Cassandra 仓库时使用的相同的 YAML 配置。需要注意的关键是 keyspace-name。在你的 Cassandra 集群中创建一个与该名称相同的键空间是非常重要的。

你还需要在你的本地机器上运行一个 Cassandra 集群，并监听 9042 端口。最简单的方法是使用 Docker，如下所示：

Bash
$ docker network create cassandra-net
$ docker run --name my-cassandra \
           --network cassandra-net \
           -p 9042:9042 \
           -d cassandra:latest

如果你的 Cassandra 集群在另一台机器或端口上，你需要在 application.yml 中指定联系点和端口，如《Spring In Action 6th：处理非关系型数据》中所示。要创建键空间，请运行 CQL shell 并使用 create keyspace 命令，如下所示：

Bash
$ docker run -it --network cassandra-net --rm cassandra cqlsh my-cassandra
cqlsh> create keyspace tacocloud
WITH replication = {'class': 'SimpleStrategy', 'replication_factor' : 1};

现在你已经有了一个 Cassandra 集群，一个新的 tacocloud 键空间，以及在你的项目中的 Spring Data Cassandra Reactive 启动器，你已经准备好开始定义领域类了。

3.1. 为 Cassandra 定义实体类 ​

正如我们在使用 Mongo 进行持久化时的情况一样，选择响应式还是非响应式的 Cassandra 持久化对于你如何定义你的领域类别没有任何影响。我们将使用的 Ingredient、Taco 和 TacoOrder 的领域类别与我们在《Spring In Action 6th：处理非关系型数据》中创建的完全相同。这里展示了一个使用 Cassandra 注解的 Ingredient 类：

Java
@Data
@AllArgsConstructor
@NoArgsConstructor(access = AccessLevel.PRIVATE, force = true)
@Table("ingredients")
public class Ingredient {
    @PrimaryKey
    private String id;
    private String name;
    private Type type;

    public enum Type {
        WRAP, PROTEIN, VEGGIES, CHEESE, SAUCE
    }
}

至于 Taco 类，它在下方示例中用类似的 Cassandra 持久化注解进行了定义：

Java
@Data
@RestResource(rel = "tacos", path = "tacos")
@Table("tacos")
public class Taco {
    @PrimaryKeyColumn(type = PrimaryKeyType.PARTITIONED)
    private UUID id = Uuids.timeBased();

    @NotNull
    @Size(min = 5, message = "Name must be at least 5 characters long")
    private String name;

    @PrimaryKeyColumn(type = PrimaryKeyType.CLUSTERED,
            ordering = Ordering.DESCENDING)
    private Date createdAt = new Date();

    @Size(min = 1, message = "You must choose at least 1 ingredient")
    @Column("ingredients")
    private List<IngredientUDT> ingredients = new ArrayList<>();

    public void addIngredient(Ingredient ingredient) {
        this.ingredients.add(new IngredientUDT(ingredient.getName(),
                ingredient.getType()));
    }
}

因为 Taco 类通过用户定义的类型引用 Ingredient 对象，你还需要 IngredientUDT 类，如下所示：

Java
@Data
@AllArgsConstructor
@NoArgsConstructor(access = AccessLevel.PRIVATE, force = true)
@UserDefinedType("ingredient")
public class IngredientUDT {
    private String name;
    private Ingredient.Type type;
}

我们三个领域类别的最后一个，TacoOrder，如下所示：

Java
@Data
@Table("tacoorders")
public class TacoOrder implements Serializable {
    private static final long serialVersionUID = 1L;

    @PrimaryKey
    private UUID id = Uuids.timeBased();

    private Date placedAt = new Date();

    @Column("user")
    private UserUDT user;

    private String deliveryName;
    private String deliveryStreet;
    private String deliveryCity;
    private String deliveryState;
    private String deliveryZip;
    private String ccNumber;
    private String ccExpiration;
    private String ccCVV;

    @Column("tacos")
    private List<TacoUDT> tacos = new ArrayList<>();

    public void addTaco(Taco taco) {
        this.addTaco(new TacoUDT(taco.getName(), taco.getIngredients()));
    }

    public void addTaco(TacoUDT tacoUDT) {
        this.tacos.add(tacoUDT);
    }
}

就像 Taco 类通过用户定义的类型引用 Ingredient 一样，TacoOrder 类通过 TacoUDT 类引用 Taco，如下所示：

Java
@Data
@UserDefinedType("taco")
public class TacoUDT {
    private final String name;
    private final List<IngredientUDT> ingredients;
}

需要重申的是，这些与它们的非响应式类是完全相同的。我在这里只是重复了它们，这样你就不必回去翻阅之前的内容了。

现在让我们定义持久化这些对象的仓库。

3.2. 创建响应式 Cassandra Repository ​

到现在为止，你可能已经预料到响应式的 Cassandra 仓库会与等效的非响应式仓库看起来非常相似。如果是这样，那太好了！你已经开始理解，无论仓库是否是响应式的，只要有可能，Spring Data 都会尝试维持类似的编程模型。

你可能已经猜到，使仓库变为响应式的唯一关键区别是接口扩展了 ReactiveCrudRepository，如下面的 IngredientRepository 接口所示：

Java
public interface IngredientRepository
        extends ReactiveCrudRepository<Ingredient, String> {
}

自然，对于 OrderRepository 也是如此，如下所示：

Java
public interface OrderRepository
        extends ReactiveCrudRepository<TacoOrder, UUID> {
    Flux<TacoOrder> findByUserOrderByPlacedAtDesc(
            User user, Pageable pageable);
}

事实上，这些仓库不仅让人想起它们的非响应式对应物，而且与我们在本章早些时候编写的 MongoDB 仓库也没有太大的区别。除了 Cassandra 使用 UUID 作为 TacoOrder 的 ID 类型，而不是 String 之外，它们几乎是完全相同的。这再次展示了 Spring Data 项目在可能的情况下所采用的一致性。

让我们通过编写一些测试来结束我们对编写响应式 Cassandra 仓库的探讨，以验证它们是否有效。

3.3. 测试响应式 Cassandra Repository ​

在这个时候，你可能不会感到惊讶，测试响应式 Cassandra 仓库与测试响应式 MongoDB 仓库非常相似。例如，看一下下方示例中的 IngredientRepositoryTest，看看你能否发现它与之前的区别：

Java
@DataCassandraTest
public class IngredientRepositoryTest {
    @Autowired
    IngredientRepository ingredientRepo;

    @BeforeEach
    public void setup() {
        Flux<Ingredient> deleteAndInsert = ingredientRepo.deleteAll()
                .thenMany(ingredientRepo.saveAll(
                        Flux.just(
                                new Ingredient("FLTO", "Flour Tortilla", Type.WRAP),
                                new Ingredient("GRBF", "Ground Beef", Type.PROTEIN),
                                new Ingredient("CHED", "Cheddar Cheese", Type.CHEESE)
                        )));
        StepVerifier.create(deleteAndInsert)
                .expectNextCount(3)
                .verifyComplete();
    }

    @Test
    public void shouldSaveAndFetchIngredients() {
        StepVerifier.create(ingredientRepo.findAll())
                .recordWith(ArrayList::new)
                .thenConsumeWhile(x -> true)
                .consumeRecordedWith(ingredients -> {
                    assertThat(ingredients).hasSize(3);
                    assertThat(ingredients).contains(
                            new Ingredient("FLTO", "Flour Tortilla", Type.WRAP));
                    assertThat(ingredients).contains(
                            new Ingredient("GRBF", "Ground Beef", Type.PROTEIN));
                    assertThat(ingredients).contains(
                            new Ingredient("CHED", "Cheddar Cheese", Type.CHEESE));
                })
                .verifyComplete();
        StepVerifier.create(ingredientRepo.findById("FLTO"))
                .assertNext(ingredient -> {
                    ingredient.equals(new Ingredient("FLTO", "Flour Tortilla",
                            Type.WRAP));
                });
    }
}

你看到了吗？MongoDB 版本是用 @DataMongoTest 注解的，而这个新的 Cassandra 版本是用 @DataCassandraTest 注解的。就是这样！其它一切都是相同的。

对于 OrderRepositoryTest 也是如此。将 @DataMongoTest 替换为 @DataCassandraTest，其他所有的都是相同的，如下所示：

Java
@DataCassandraTest
public class OrderRepositoryTest {
    ...
}

各种 Spring Data 项目之间的一致性甚至扩展到了测试的编写方式。这使得在不同的持久化到不同类型数据库的项目之间切换变得容易，而无需对它们的开发方式进行太多不同的思考。

4. 总结 ​

  • Spring Data 支持各种数据库类型的响应式持久化，包括关系数据库（使用 R2DBC）、MongoDB 和 Cassandra；

  • Spring Data R2DBC 为关系持久化提供了一个响应式选项，但尚未直接支持领域类别中的关系；

  • 由于缺乏直接的关系支持，Spring Data R2DBC 仓库需要对领域和数据库表设计采取不同的方法；

  • Spring Data MongoDB 和 Spring Data Cassandra 为编写 MongoDB 和 Cassandra 数据库的响应式仓库提供了几乎相同的编程模型；

  • 使用 Spring Data 测试注解和 StepVerifier，你可以测试从 Spring 应用上下文自动创建的响应式仓库；

转载请注明作者和出处

Copyright © 2025-至今 Tenormis